[Feature][transform-v2] SQL transform support schema change

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/pom.xml

@@ -25,7 +25,9 @@
     <artifactId>seatunnel-transforms-v2-e2e-part-2</artifactId>
 
     <name>SeaTunnel : E2E : Transforms V2 : Part 2</name>
-
+    <properties>
+        <mysql.version>8.0.31</mysql.version>
+    </properties>
     <dependencies>
 
         <dependency>
@@ -35,7 +37,39 @@
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+        <!-- test dependencies on TestContainers -->
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mysql</artifactId>
+            <version>${testcontainer.version}</version>
+            <scope>test</scope>
+        </dependency>
 
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-cdc-mysql</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>${mysql.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>1.17.3</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-elasticsearch</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLSchemaChangeIT.java

@@ -0,0 +1,208 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.e2e.transform;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlContainer;
+import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlVersion;
+import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.UniqueDatabase;
+import org.apache.seatunnel.connectors.seatunnel.elasticsearch.client.EsRestClient;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.shaded.com.google.common.collect.Lists;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+@Slf4j
+@DisabledOnContainer(
+        value = {},
+        type = {EngineType.SPARK, EngineType.FLINK},
+        disabledReason =
+                "Currently SPARK do not support cdc. In addition, currently only the zeta engine supports schema evolution for pr https://github.com/apache/seatunnel/pull/5125.")
+public class TestSQLSchemaChangeIT extends TestSuiteBase implements TestResource {
+
+    private ElasticsearchContainer container;
+
+    private static final MySqlContainer MYSQL_CONTAINER = createMySqlContainer(MySqlVersion.V8_0);
+    private static final String MYSQL_HOST = "mysql_cdc_e2e";
+    private static final String MYSQL_USER_NAME = "mysqluser";
+    private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+    private static final String DATABASE = "shop";
+    protected static final String DRIVER_JAR =
+            "https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";
+    private final UniqueDatabase shopDatabase = new UniqueDatabase(MYSQL_CONTAINER, DATABASE);
+
+    private EsRestClient esRestClient;
+
+    @BeforeEach
+    @Override
+    public void startUp() throws Exception {
+        container =
+                new ElasticsearchContainer(
+                                DockerImageName.parse("elasticsearch:8.9.0")
+                                        .asCompatibleSubstituteFor(
+                                                "docker.elastic.co/elasticsearch/elasticsearch"))
+                        .withNetwork(NETWORK)
+                        .withEnv("cluster.routing.allocation.disk.threshold_enabled", "false")
+                        .withNetworkAliases("elasticsearch")
+                        .withPassword("elasticsearch")
+                        .withStartupAttempts(5)
+                        .withStartupTimeout(Duration.ofMinutes(5))
+                        .withLogConsumer(
+                                new Slf4jLogConsumer(
+                                        DockerLoggerFactory.getLogger("elasticsearch:8.9.0")));
+        Startables.deepStart(Stream.of(container)).join();
+        log.info("Elasticsearch container started");
+        esRestClient =
+                EsRestClient.createInstance(
+                        Lists.newArrayList("https://" + container.getHttpHostAddress()),
+                        Optional.of("elastic"),
+                        Optional.of("elasticsearch"),
+                        false,
+                        false,
+                        Optional.empty(),
+                        Optional.empty(),
+                        Optional.empty(),
+                        Optional.empty());
+
+        Startables.deepStart(Stream.of(MYSQL_CONTAINER)).join();
+        shopDatabase.createAndInitialize();
+    }
+
+    @TestContainerExtension
+    private final ContainerExtendedFactory extendedFactory =
+            container -> {
+                Container.ExecResult extraCommands =
+                        container.execInContainer(
+                                "bash",
+                                "-c",
+                                "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib && cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
+                                        + DRIVER_JAR);
+                Assertions.assertEquals(0, extraCommands.getExitCode());
+            };
+
+    private static MySqlContainer createMySqlContainer(MySqlVersion version) {
+        MySqlContainer mySqlContainer =
+                new MySqlContainer(version)
+                        .withConfigurationOverride("docker/server-gtids/my.cnf")
+                        .withSetupSQL("docker/setup.sql")
+                        .withNetwork(NETWORK)
+                        .withNetworkAliases(MYSQL_HOST)
+                        .withDatabaseName(DATABASE)
+                        .withUsername(MYSQL_USER_NAME)
+                        .withPassword(MYSQL_USER_PASSWORD)
+                        .withLogConsumer(
+                                new Slf4jLogConsumer(
+                                        DockerLoggerFactory.getLogger("mysql-docker-image")));
+        mySqlContainer.setPortBindings(Lists.newArrayList(String.format("%s:%s", 3306, 3306)));
+        return mySqlContainer;
+    }
+
+    @TestTemplate
+    public void testSchemaChange(TestContainer container) throws InterruptedException {
+
+        String jobId = String.valueOf(JobIdGenerator.newJobId());
+        String jobConfigFile = "/sql_transform/mysqlcdc_to_sql_transform_with_schema_change.conf";
+        CompletableFuture.runAsync(
+                () -> {
+                    try {
+                        container.executeJob(jobConfigFile, jobId);
+                    } catch (Exception e) {
+                        log.error("Commit task exception :" + e.getMessage());
+                        throw new RuntimeException(e);
+                    }
+                });
+
+        TimeUnit.SECONDS.sleep(20);
+        shopDatabase.setTemplateName("add_columns").createAndInitialize();
+
+        await().atMost(120, TimeUnit.SECONDS)
+                .pollInterval(3, TimeUnit.SECONDS)
+                .ignoreExceptions()
+                .untilAsserted(
+                        () -> {
+                            Container.ExecResult execResult =
+                                    this.container.execInContainer(
+                                            "bash",
+                                            "-c",
+                                            "curl -k -u elastic:elasticsearch https://localhost:9200/schema_change_index/_mapping");
+                            ObjectNode jsonNodes = JsonUtils.parseObject(execResult.getStdout());
+                            JsonNode schemaChangeIndex =
+                                    jsonNodes
+                                            .get("schema_change_index")
+                                            .get("mappings")
+                                            .get("properties");
+                            Assertions.assertEquals(
+                                    schemaChangeIndex.get("add_column1").get("type").asText(),
+                                    "text");
+                            Assertions.assertEquals(
+                                    schemaChangeIndex.get("add_column2").get("type").asText(),
+                                    "integer");
+                            Assertions.assertEquals(
+                                    schemaChangeIndex.get("add_column3").get("type").asText(),
+                                    "float");
+                            Assertions.assertEquals(
+                                    schemaChangeIndex.get("add_column4").get("type").asText(),
+                                    "date");
+                            Container.ExecResult indexCountResult =
+                                    this.container.execInContainer(
+                                            "bash",
+                                            "-c",
+                                            "curl -k -u elastic:elasticsearch https://localhost:9200/schema_change_index/_count");
+                            Assertions.assertTrue(
+                                    indexCountResult.getStdout().contains("\"count\":18"));
+                        });
+    }
+
+    @AfterEach
+    @Override
+    public void tearDown() {
+        if (Objects.nonNull(esRestClient)) {
+            esRestClient.close();
+        }
+        container.close();
+    }
+}

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/add_columns.sql

@@ -0,0 +1,39 @@
+--
+-- 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 ADD COLUMN add_column1 varchar(64) not null default 'yy',ADD COLUMN add_column2 int not null default 1;
+
+insert into products
+values (119,"scooter","Small 2-wheel scooter",3.14,'xx',1),
+       (120,"car battery","12V car battery",8.1,'xx',2),
+       (121,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,'xx',3),
+       (122,"hammer","12oz carpenter's hammer",0.75,'xx',4),
+       (123,"hammer","14oz carpenter's hammer",0.875,'xx',5),
+       (124,"hammer","16oz carpenter's hammer",1.0,'xx',6),
+       (125,"rocks","box of assorted rocks",5.3,'xx',7),
+       (126,"jacket","water resistent black wind breaker",0.1,'xx',8),
+       (127,"spare tire","24 inch spare tire",22.2,'xx',9);
+
+
+alter table products ADD COLUMN add_column3 float not null default 1.1;
+alter table products ADD COLUMN add_column4 datetime not null default now();

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/shop.sql

@@ -0,0 +1,42 @@
+--
+-- 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;
+
+drop table if exists products;
+-- Create and populate our products using a single insert with many rows
+CREATE TABLE products (
+                          id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+                          name VARCHAR(255) NOT NULL DEFAULT 'SeaTunnel',
+                          description VARCHAR(512),
+                          weight FLOAT
+);
+
+INSERT INTO products
+VALUES (101,"scooter","Small 2-wheel scooter",3.14),
+       (102,"car battery","12V car battery",8.1),
+       (103,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8),
+       (104,"hammer","12oz carpenter's hammer",0.75),
+       (105,"hammer","14oz carpenter's hammer",0.875),
+       (106,"hammer","16oz carpenter's hammer",1.0),
+       (107,"rocks","box of assorted rocks",5.3),
+       (108,"jacket","water resistent black wind breaker",0.1),
+       (109,"spare tire","24 inch spare tire",22.2);