[Fix][Connector-starrocks] Fix drop column bug for starrocks (#8216)

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/util/SchemaUtils.java

@@ -236,7 +236,7 @@ public static boolean columnExists(Connection connection, TablePath tablePath, S
         String selectColumnSQL =
                 String.format(
                         "SELECT %s FROM %s WHERE 1 != 1",
-                        quoteIdentifier(column), tablePath.getTableName());
+                        quoteIdentifier(column), tablePath.getFullName());
         try (Statement statement = connection.createStatement()) {
             return statement.execute(selectColumnSQL);
         } catch (SQLException e) {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksSchemaChangeIT.java

@@ -191,6 +191,9 @@ public void testStarRocksSinkWithSchemaEvolutionCase(TestContainer container)
         assertSchemaEvolutionForAddColumns(
                 DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection);
 
+        assertSchemaEvolutionForDropColumns(
+                DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection);
+
         // savepoint 1
         Assertions.assertEquals(0, container.savepointJob(jobId).getExitCode());
 
@@ -303,6 +306,27 @@ private void assertSchemaEvolutionForAddColumns(
                         });
     }
 
+    private void assertSchemaEvolutionForDropColumns(
+            String database,
+            String sourceTable,
+            String sinkTable,
+            Connection sourceConnection,
+            Connection sinkConnection) {
+
+        // case1 add columns with cdc data at same time
+        shopDatabase.setTemplateName("drop_columns_validate_schema.sql").createAndInitialize();
+        await().atMost(60000, TimeUnit.MILLISECONDS)
+                .untilAsserted(
+                        () ->
+                                Assertions.assertIterableEquals(
+                                        query(
+                                                String.format(QUERY_COLUMNS, database, sourceTable),
+                                                sourceConnection),
+                                        query(
+                                                String.format(QUERY_COLUMNS, database, sinkTable),
+                                                sinkConnection)));
+    }
+
     private void assertTableStructureAndData(
             String database,
             String sourceTable,

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/ddl/drop_columns.sql

@@ -21,20 +21,6 @@
 CREATE DATABASE IF NOT EXISTS `shop`;
 use shop;
 
-alter table products drop column add_column4,drop column add_column6;
-insert into products
-values (137,"scooter","Small 2-wheel scooter",3.14,'xx',1,1.1),
-       (138,"car battery","12V car battery",8.1,'xx',2,1.2),
-       (139,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,'xx',3,1.3),
-       (140,"hammer","12oz carpenter's hammer",0.75,'xx',4,1.4),
-       (141,"hammer","14oz carpenter's hammer",0.875,'xx',5,1.5),
-       (142,"hammer","16oz carpenter's hammer",1.0,'xx',6,1.6),
-       (143,"rocks","box of assorted rocks",5.3,'xx',7,1.7),
-       (144,"jacket","water resistent black wind breaker",0.1,'xx',8,1.8),
-       (145,"spare tire","24 inch spare tire",22.2,'xx',9,1.9);
-update products set name = 'dailai' where id in (140,141,142);
-delete from products where id < 137;
-
 
 alter table products drop column add_column1,drop column add_column3;
 insert into products

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/ddl/drop_columns_validate_schema.sql

@@ -0,0 +1,36 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements.  See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License.  You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+-- ----------------------------------------------------------------------------------------------------------------
+-- DATABASE:  shop
+-- ----------------------------------------------------------------------------------------------------------------
+CREATE DATABASE IF NOT EXISTS `shop`;
+use shop;
+
+alter table products drop column add_column4,drop column add_column6;
+insert into products
+values (137,"scooter","Small 2-wheel scooter",3.14,'xx',1,1.1),
+       (138,"car battery","12V car battery",8.1,'xx',2,1.2),
+       (139,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,'xx',3,1.3),
+       (140,"hammer","12oz carpenter's hammer",0.75,'xx',4,1.4),
+       (141,"hammer","14oz carpenter's hammer",0.875,'xx',5,1.5),
+       (142,"hammer","16oz carpenter's hammer",1.0,'xx',6,1.6),
+       (143,"rocks","box of assorted rocks",5.3,'xx',7,1.7),
+       (144,"jacket","water resistent black wind breaker",0.1,'xx',8,1.8),
+       (145,"spare tire","24 inch spare tire",22.2,'xx',9,1.9);
+update products set name = 'dailai' where id in (140,141,142);
+delete from products where id < 137;