diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/pom.xml b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/pom.xml index 9f8df61fb3f..c52270b2d9a 100644 --- a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/pom.xml +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/pom.xml @@ -25,7 +25,9 @@ seatunnel-transforms-v2-e2e-part-2 SeaTunnel : E2E : Transforms V2 : Part 2 - + + 8.0.31 + @@ -35,7 +37,45 @@ test-jar test - + + + org.apache.seatunnel + connector-starrocks + ${project.version} + test + + + org.apache.seatunnel + connector-fake + ${project.version} + test + + + mysql + mysql-connector-java + ${mysql.version} + test + + + org.apache.seatunnel + connector-cdc-mysql + ${project.version} + test-jar + test + + + org.apache.seatunnel + connector-assert + ${project.version} + test + + + + org.testcontainers + mysql + ${testcontainer.version} + test + diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLSchemaChangeIT.java b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLSchemaChangeIT.java new file mode 100644 index 00000000000..b11584b1562 --- /dev/null +++ b/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,471 @@ +/* + * 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.google.common.collect.Lists; + +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.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +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.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.lifecycle.Startables; +import org.testcontainers.utility.DockerLoggerFactory; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.net.URL; +import java.net.URLClassLoader; +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.sql.Timestamp; +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.awaitility.Awaitility.await; +import static org.awaitility.Awaitility.given; + +@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 static final String DATABASE = "shop"; + private static final String SOURCE_TABLE = "products"; + 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 DOCKER_IMAGE = "starrocks/allin1-ubuntu:3.3.4"; + private static final String DRIVER_CLASS = "com.mysql.cj.jdbc.Driver"; + private static final String HOST = "starrocks_cdc_e2e"; + private static final int SR_PROXY_PORT = 8080; + private static final int QUERY_PORT = 9030; + private static final int HTTP_PORT = 8030; + private static final int BE_HTTP_PORT = 8040; + private static final String USERNAME = "root"; + private static final String PASSWORD = ""; + private static final String SINK_TABLE = "products"; + private static final String CREATE_DATABASE = "CREATE DATABASE IF NOT EXISTS " + DATABASE; + private static final String SR_DRIVER_JAR = + "https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar"; + + private Connection starRocksConnection; + private Connection mysqlConnection; + private GenericContainer starRocksServer; + + public static final DateTimeFormatter DATE_TIME_FORMATTER = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + + private static final String QUERY = "select * from %s.%s order by id"; + private static final String QUERY_COLUMNS = + "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s' ORDER by COLUMN_NAME;"; + private static final String PROJECTION_QUERY = + "select id,name,description,weight,add_column1,add_column2,add_column3 from %s.%s order by id;"; + + private static final MySqlContainer MYSQL_CONTAINER = createMySqlContainer(MySqlVersion.V8_0); + + private final UniqueDatabase shopDatabase = + new UniqueDatabase(MYSQL_CONTAINER, DATABASE, "mysqluser", "mysqlpw", DATABASE); + + @TestContainerExtension + private final ContainerExtendedFactory extendedFactory = + container -> { + Container.ExecResult extraCommands = + container.execInContainer( + "bash", + "-c", + "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib && cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O " + + SR_DRIVER_JAR); + Assertions.assertEquals(0, extraCommands.getExitCode()); + }; + + private static MySqlContainer createMySqlContainer(MySqlVersion version) { + return 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"))); + } + + private void initializeJdbcConnection() throws Exception { + URLClassLoader urlClassLoader = + new URLClassLoader( + new URL[] {new URL(SR_DRIVER_JAR)}, + TestSQLSchemaChangeIT.class.getClassLoader()); + Thread.currentThread().setContextClassLoader(urlClassLoader); + Driver driver = (Driver) urlClassLoader.loadClass(DRIVER_CLASS).newInstance(); + Properties props = new Properties(); + props.put("user", USERNAME); + props.put("password", PASSWORD); + starRocksConnection = + driver.connect( + String.format("jdbc:mysql://%s:%s", starRocksServer.getHost(), QUERY_PORT), + props); + } + + private void initializeStarRocksServer() { + starRocksServer = + new GenericContainer<>(DOCKER_IMAGE) + .withNetwork(NETWORK) + .withNetworkAliases(HOST) + .withLogConsumer( + new Slf4jLogConsumer(DockerLoggerFactory.getLogger(DOCKER_IMAGE))); + starRocksServer.setPortBindings( + Lists.newArrayList( + String.format("%s:%s", QUERY_PORT, QUERY_PORT), + String.format("%s:%s", HTTP_PORT, HTTP_PORT), + String.format("%s:%s", BE_HTTP_PORT, BE_HTTP_PORT))); + Startables.deepStart(Stream.of(starRocksServer)).join(); + log.info("StarRocks container started"); + // wait for starrocks fully start + given().ignoreExceptions() + .await() + .atMost(360, TimeUnit.SECONDS) + .untilAsserted(this::initializeJdbcConnection); + } + + @TestTemplate + public void testStarRocksSinkWithSchemaEvolutionCase(TestContainer container) + throws InterruptedException, IOException, SQLException { + 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); + + // verify multi table sink + verifyDataConsistency("orders"); + verifyDataConsistency("customers"); + + // waiting for case1 completed + assertSchemaEvolutionForAddColumns( + DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection); + + assertSchemaEvolutionForDropColumns( + DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection); + + insertNewDataIntoMySQL(); + insertNewDataIntoMySQL(); + // verify incremental + verifyDataConsistency("orders"); + + // savepoint 1 + Assertions.assertEquals(0, container.savepointJob(jobId).getExitCode()); + insertNewDataIntoMySQL(); + // case2 drop columns with cdc data at same time + shopDatabase.setTemplateName("drop_columns").createAndInitialize(); + + // restore 1 + CompletableFuture.supplyAsync( + () -> { + try { + container.restoreJob(jobConfigFile, jobId); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + return null; + }); + + // waiting for case2 completed + assertTableStructureAndData( + DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection); + + // savepoint 2 + Assertions.assertEquals(0, container.savepointJob(jobId).getExitCode()); + + // case3 change column name with cdc data at same time + shopDatabase.setTemplateName("change_columns").createAndInitialize(); + + // case4 modify column data type with cdc data at same time + shopDatabase.setTemplateName("modify_columns").createAndInitialize(); + + // restore 2 + CompletableFuture.supplyAsync( + () -> { + try { + container.restoreJob(jobConfigFile, jobId); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + return null; + }); + + // waiting for case3/case4 completed + assertTableStructureAndData( + DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection, starRocksConnection); + insertNewDataIntoMySQL(); + // verify restore + verifyDataConsistency("orders"); + } + + private void insertNewDataIntoMySQL() throws SQLException { + mysqlConnection + .createStatement() + .execute( + "INSERT INTO orders (id, customer_id, order_date, total_amount, status) " + + "VALUES (null, 1, '2025-01-04 13:00:00', 498.99, 'pending')"); + } + + private void verifyDataConsistency(String tableName) { + await().atMost(10000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + String.format(QUERY, DATABASE, tableName), + mysqlConnection), + query( + String.format(QUERY, DATABASE, tableName), + starRocksConnection))); + } + + private void assertSchemaEvolutionForAddColumns( + String database, + String sourceTable, + String sinkTable, + Connection sourceConnection, + Connection sinkConnection) { + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + String.format(QUERY, database, sourceTable), + sourceConnection), + query( + String.format(QUERY, database, sinkTable), + sinkConnection))); + + // case1 add columns with cdc data at same time + shopDatabase.setTemplateName("add_columns").createAndInitialize(); + try { + TimeUnit.SECONDS.sleep(10); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + String.format(QUERY_COLUMNS, database, sourceTable), + sourceConnection), + query( + String.format(QUERY_COLUMNS, database, sinkTable), + sinkConnection))); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> { + Assertions.assertIterableEquals( + query( + String.format( + QUERY.replaceAll( + "order by id", + "where id >= 128 order by id"), + database, + sourceTable), + sourceConnection), + query( + String.format( + QUERY.replaceAll( + "order by id", + "where id >= 128 order by id"), + database, + sinkTable), + sinkConnection)); + }); + + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> { + Assertions.assertIterableEquals( + query( + String.format(PROJECTION_QUERY, database, sourceTable), + sourceConnection), + query( + String.format(PROJECTION_QUERY, database, sinkTable), + sinkConnection)); + }); + } + + 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").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, + String sinkTable, + Connection sourceConnection, + Connection sinkConnection) { + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + String.format(QUERY_COLUMNS, database, sourceTable), + sourceConnection), + query( + String.format(QUERY_COLUMNS, database, sinkTable), + sinkConnection))); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query( + String.format(QUERY, database, sourceTable), + sourceConnection), + query( + String.format(QUERY, database, sinkTable), + sinkConnection))); + } + + private Connection getMysqlJdbcConnection() throws SQLException { + return DriverManager.getConnection( + MYSQL_CONTAINER.getJdbcUrl(), + MYSQL_CONTAINER.getUsername(), + MYSQL_CONTAINER.getPassword()); + } + + @BeforeAll + @Override + public void startUp() throws SQLException { + initializeStarRocksServer(); + log.info("The second stage: Starting Mysql containers..."); + Startables.deepStart(Stream.of(MYSQL_CONTAINER)).join(); + log.info("Mysql Containers are started"); + shopDatabase.createAndInitialize(); + log.info("Mysql ddl execution is complete"); + initializeJdbcTable(); + mysqlConnection = getMysqlJdbcConnection(); + } + + @AfterAll + @Override + public void tearDown() throws SQLException { + if (MYSQL_CONTAINER != null) { + MYSQL_CONTAINER.close(); + } + if (starRocksServer != null) { + starRocksServer.close(); + } + if (starRocksConnection != null) { + starRocksConnection.close(); + } + if (mysqlConnection != null) { + mysqlConnection.close(); + } + } + + private void initializeJdbcTable() { + try (Statement statement = starRocksConnection.createStatement()) { + // create databases + statement.execute(CREATE_DATABASE); + } catch (SQLException e) { + throw new RuntimeException("Initializing table failed!", e); + } + } + + private List> query(String sql, Connection connection) { + try { + ResultSet resultSet = connection.createStatement().executeQuery(sql); + List> result = new ArrayList<>(); + int columnCount = resultSet.getMetaData().getColumnCount(); + while (resultSet.next()) { + ArrayList objects = new ArrayList<>(); + for (int i = 1; i <= columnCount; i++) { + if (resultSet.getObject(i) instanceof Timestamp) { + Timestamp timestamp = resultSet.getTimestamp(i); + objects.add(timestamp.toLocalDateTime().format(DATE_TIME_FORMATTER)); + break; + } + if (resultSet.getObject(i) instanceof LocalDateTime) { + LocalDateTime localDateTime = resultSet.getObject(i, LocalDateTime.class); + objects.add(localDateTime.format(DATE_TIME_FORMATTER)); + break; + } + objects.add(resultSet.getObject(i)); + } + log.debug(String.format("Print query, sql: %s, data: %s", sql, objects)); + result.add(objects); + } + return result; + } catch (SQLException e) { + throw new RuntimeException(e); + } + } +} diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/add_columns.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/add_columns.sql new file mode 100644 index 00000000000..2a1212aa95d --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/add_columns.sql @@ -0,0 +1,83 @@ +-- +-- 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; +INSERT INTO products +VALUES (110,"scooter","Small 2-wheel scooter",3.14), + (111,"car battery","12V car battery",8.1), + (112,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8), + (113,"hammer","12oz carpenter's hammer",0.75), + (114,"hammer","14oz carpenter's hammer",0.875), + (115,"hammer","16oz carpenter's hammer",1.0), + (116,"rocks","box of assorted rocks",5.3), + (117,"jacket","water resistent black wind breaker",0.1), + (118,"spare tire","24 inch spare tire",22.2); +update products set name = 'dailai' where id = 101; +delete from products where id = 102; + +alter table products ADD COLUMN add_column1 varchar(64) not null default 'yy',ADD COLUMN add_column2 int not null default 1; +update products set add_column1 = 'swm1', add_column2 = 2; + +update products set name = 'dailai' where id = 110; +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); +delete from products where id = 118; + +alter table products ADD COLUMN add_column3 float not null default 1.1; +update products set add_column3 = 3.3; +alter table products ADD COLUMN add_column4 timestamp not null default current_timestamp(); +update products set add_column4 = current_timestamp(); + +delete from products where id = 113; +insert into products +values (128,"scooter","Small 2-wheel scooter",3.14,'xx',1,1.1,'2023-02-02 09:09:09'), + (129,"car battery","12V car battery",8.1,'xx',2,1.2,'2023-02-02 09:09:09'), + (130,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,'xx',3,1.3,'2023-02-02 09:09:09'), + (131,"hammer","12oz carpenter's hammer",0.75,'xx',4,1.4,'2023-02-02 09:09:09'), + (132,"hammer","14oz carpenter's hammer",0.875,'xx',5,1.5,'2023-02-02 09:09:09'), + (133,"hammer","16oz carpenter's hammer",1.0,'xx',6,1.6,'2023-02-02 09:09:09'), + (134,"rocks","box of assorted rocks",5.3,'xx',7,1.7,'2023-02-02 09:09:09'), + (135,"jacket","water resistent black wind breaker",0.1,'xx',8,1.8,'2023-02-02 09:09:09'), + (136,"spare tire","24 inch spare tire",22.2,'xx',9,1.9,'2023-02-02 09:09:09'); +update products set name = 'dailai' where id = 135; + +alter table products ADD COLUMN add_column6 varchar(64) not null default 'ff' after id; +update products set add_column6 = 'swm6'; + +delete from products where id = 115; +insert into products +values (173,'tt',"scooter","Small 2-wheel scooter",3.14,'xx',1,1.1,'2023-02-02 09:09:09'), + (174,'tt',"car battery","12V car battery",8.1,'xx',2,1.2,'2023-02-02 09:09:09'), + (175,'tt',"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,'xx',3,1.3,'2023-02-02 09:09:09'), + (176,'tt',"hammer","12oz carpenter's hammer",0.75,'xx',4,1.4,'2023-02-02 09:09:09'), + (177,'tt',"hammer","14oz carpenter's hammer",0.875,'xx',5,1.5,'2023-02-02 09:09:09'), + (178,'tt',"hammer","16oz carpenter's hammer",1.0,'xx',6,1.6,'2023-02-02 09:09:09'), + (179,'tt',"rocks","box of assorted rocks",5.3,'xx',7,1.7,'2023-02-02 09:09:09'), + (180,'tt',"jacket","water resistent black wind breaker",0.1,'xx',8,1.8,'2023-02-02 09:09:09'), + (181,'tt',"spare tire","24 inch spare tire",22.2,'xx',9,1.9,'2023-02-02 09:09:09'); diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/change_columns.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/change_columns.sql new file mode 100644 index 00000000000..a17f9a0a936 --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/change_columns.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 change add_column2 add_column int default 1 not null; +delete from products where id < 155; +insert into products +values (155,"scooter","Small 2-wheel scooter",3.14,1), + (156,"car battery","12V car battery",8.1,2), + (157,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,3), + (158,"hammer","12oz carpenter's hammer",0.75,4), + (159,"hammer","14oz carpenter's hammer",0.875,5), + (160,"hammer","16oz carpenter's hammer",1.0,6), + (161,"rocks","box of assorted rocks",5.3,7), + (162,"jacket","water resistent black wind breaker",0.1,8), + (163,"spare tire","24 inch spare tire",22.2,9); + diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/drop_columns.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/drop_columns.sql new file mode 100644 index 00000000000..9464e02e1d3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/drop_columns.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_column1,drop column add_column3; +insert into products +values (146,"scooter","Small 2-wheel scooter",3.14,1), + (147,"car battery","12V car battery",8.1,2), + (148,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,3), + (149,"hammer","12oz carpenter's hammer",0.75,4), + (150,"hammer","14oz carpenter's hammer",0.875,5), + (151,"hammer","16oz carpenter's hammer",1.0,6), + (152,"rocks","box of assorted rocks",5.3,7), + (153,"jacket","water resistent black wind breaker",0.1,8), + (154,"spare tire","24 inch spare tire",22.2,9); +update products set name = 'dailai' where id > 143; diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/drop_columns_validate_schema.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/drop_columns_validate_schema.sql new file mode 100644 index 00000000000..262006bd81e --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/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; \ No newline at end of file diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/modify_columns.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/modify_columns.sql new file mode 100644 index 00000000000..ab64c47567b --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/modify_columns.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 modify name longtext null; +delete from products where id < 155; +insert into products +values (164,"scooter","Small 2-wheel scooter",3.14,1), + (165,"car battery","12V car battery",8.1,2), + (166,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8,3), + (167,"hammer","12oz carpenter's hammer",0.75,4), + (168,"hammer","14oz carpenter's hammer",0.875,5), + (169,"hammer","16oz carpenter's hammer",1.0,6), + (170,"rocks","box of assorted rocks",5.3,7), + (171,"jacket","water resistent black wind breaker",0.1,8), + (172,"spare tire","24 inch spare tire",22.2,9); + diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/shop.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/shop.sql new file mode 100644 index 00000000000..b867cd24c3c --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/ddl/shop.sql @@ -0,0 +1,80 @@ +-- +-- 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 +); + +drop table if exists orders; + +CREATE TABLE orders ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + customer_id BIGINT NOT NULL, + order_date DATETIME NOT NULL, + total_amount DECIMAL ( 10, 2 ) NOT NULL, + STATUS VARCHAR ( 50 ) DEFAULT 'pending', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +); + +drop table if exists customers; + +CREATE TABLE customers ( + id BIGINT PRIMARY KEY, + NAME VARCHAR ( 255 ) NOT NULL, + email VARCHAR ( 255 ) NOT NULL, + phone VARCHAR ( 50 ), + address TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +); + +ALTER TABLE products AUTO_INCREMENT = 101; + +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); + +INSERT INTO orders ( id, customer_id, order_date, total_amount, STATUS ) +VALUES + ( 1, 1, '2024-01-01 10:00:00', 299.99, 'completed' ), + ( 2, 2, '2024-01-02 11:00:00', 199.99, 'completed' ), + ( 3, 3, '2024-01-03 12:00:00', 399.99, 'processing' ); + +INSERT INTO customers ( id, NAME, email, phone, address ) +VALUES + ( 1, 'John Doe', 'john@example.com', '123-456-7890', '123 Main St' ), + ( 2, 'Jane Smith', 'jane@example.com', '234-567-8901', '456 Oak Ave' ), + ( 3, 'Bob Johnson', 'bob@example.com', '345-678-9012', '789 Pine Rd' ); diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/server-gtids/my.cnf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/server-gtids/my.cnf new file mode 100644 index 00000000000..a390897885d --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/server-gtids/my.cnf @@ -0,0 +1,65 @@ +# +# 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. +# + +# For advice on how to change settings please see +# http://dev.mysql.com/doc/refman/5.7/en/server-configuration-defaults.html + +[mysqld] +# +# Remove leading # and set to the amount of RAM for the most important data +# cache in MySQL. Start at 70% of total RAM for dedicated server, else 10%. +# innodb_buffer_pool_size = 128M +# +# Remove leading # to turn on a very important data integrity option: logging +# changes to the binary log between backups. +# log_bin +# +# Remove leading # to set options mainly useful for reporting servers. +# The server defaults are faster for transactions and fast SELECTs. +# Adjust sizes as needed, experiment to find the optimal values. +# join_buffer_size = 128M +# sort_buffer_size = 2M +# read_rnd_buffer_size = 2M +skip-host-cache +skip-name-resolve +#datadir=/var/lib/mysql +#socket=/var/lib/mysql/mysql.sock +secure-file-priv=/var/lib/mysql +user=mysql + +# Disabling symbolic-links is recommended to prevent assorted security risks +symbolic-links=0 + +#log-error=/var/log/mysqld.log +#pid-file=/var/run/mysqld/mysqld.pid + +# ---------------------------------------------- +# Enable the binlog for replication & CDC +# ---------------------------------------------- + +# Enable binary replication log and set the prefix, expiration, and log format. +# The prefix is arbitrary, expiration can be short for integration tests but would +# be longer on a production system. Row-level info is required for ingest to work. +# Server ID is required, but this will vary on production systems +server-id = 223344 +log_bin = mysql-bin +expire_logs_days = 1 +binlog_format = row + +# enable gtid mode +gtid_mode = on +enforce_gtid_consistency = on \ No newline at end of file diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/setup.sql b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/setup.sql new file mode 100644 index 00000000000..495bda2c7a1 --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/docker/setup.sql @@ -0,0 +1,29 @@ +-- +-- 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. +-- + +-- In production you would almost certainly limit the replication user must be on the follower (slave) machine, +-- to prevent other clients accessing the log from other machines. For example, 'replicator'@'follower.acme.com'. +-- However, in this database we'll grant 2 users different privileges: +-- +-- 1) 'mysqluser' - all privileges +-- 2) 'st_user_source' - all privileges required by the snapshot reader AND binlog reader (used for testing) +-- +GRANT ALL PRIVILEGES ON *.* TO 'mysqluser'@'%'; + +CREATE USER 'st_user_source' IDENTIFIED BY 'mysqlpw'; +GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, DROP, LOCK TABLES ON *.* TO 'st_user_source'@'%'; +-- ---------------------------------------------------------------------------------------------------------------- diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/mysqlcdc_to_sql_transform_with_schema_change.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/mysqlcdc_to_sql_transform_with_schema_change.conf new file mode 100644 index 00000000000..413e161ca6f --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/mysqlcdc_to_sql_transform_with_schema_change.conf @@ -0,0 +1,61 @@ +# +# 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. +# + +###### +###### This config file is a demonstration of streaming processing in seatunnel config +###### + +env { + parallelism = 1 + job.mode = "STREAMING" +} + +source{ + MySQL-CDC { + plugin_output = "fake" + server-id = 5652-5657 + username = "st_user_source" + password = "mysqlpw" + table-names = ["shop.products"] + base-url = "jdbc:mysql://mysql_cdc_e2e:3306/shop" + table-names-config = [{"table": "shop.products", "primaryKeys": ["id"]}] + schema-changes.enabled = true + } + +} + +transform { + Sql { + plugin_input = "fake" + plugin_output = "fake1" + query = "select * from fake" + } +} + +sink { + Elasticsearch { + hosts = ["https://elasticsearch:9200"] + username = "elastic" + password = "elasticsearch" + tls_verify_certificate = false + tls_verify_hostname = false + index = "schema_change_index" + index_type = "_doc" + "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST" + "data_save_mode"="APPEND_DATA" + } +} diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/mysqlcdc_to_sql_transform_with_schema_change.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/mysqlcdc_to_sql_transform_with_schema_change.conf new file mode 100644 index 00000000000..6a20e65f0a3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/mysqlcdc_to_sql_transform_with_schema_change.conf @@ -0,0 +1,67 @@ +# +# 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. +# + +###### +###### This config file is a demonstration of streaming processing in seatunnel config +###### + +env { + # You can set engine configuration here + job.mode = "STREAMING" + checkpoint.interval = 2000 +} + +source { + MySQL-CDC { + username = "st_user_source" + password = "mysqlpw" + table-names = ["shop.products", "shop.orders", "shop.customers"] + base-url = "jdbc:mysql://mysql_cdc_e2e:3306/shop" + + schema-changes.enabled = true + } +} +sink { + StarRocks { + # docker allin1 environment can use port 8080 8040 instead of port FE 8030 + nodeUrls = ["starrocks_cdc_e2e:8040"] + username = "root" + password = "" + database = "shop" + table = "${table_name}" + base-url = "jdbc:mysql://starrocks_cdc_e2e:9030/shop" + max_retries = 3 + enable_upsert_delete = true + schema_save_mode="RECREATE_SCHEMA" + data_save_mode="DROP_DATA" + save_mode_create_template = """ + CREATE TABLE IF NOT EXISTS shop.`${table_name}` ( + ${rowtype_primary_key}, + ${rowtype_fields} + ) ENGINE=OLAP + PRIMARY KEY (${rowtype_primary_key}) + DISTRIBUTED BY HASH (${rowtype_primary_key}) + PROPERTIES ( + "replication_num" = "1", + "in_memory" = "false", + "enable_persistent_index" = "true", + "replicated_storage" = "true", + "compression" = "LZ4" + ) + """ + } +} diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java index b1deb453d6d..eb8b3515508 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractMultiCatalogFlatMapTransform.java @@ -19,6 +19,7 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.transform.SeaTunnelFlatMapTransform; @@ -43,4 +44,10 @@ public List flatMap(SeaTunnelRow row) { return ((SeaTunnelFlatMapTransform) transformMap.get(row.getTableId())) .flatMap(row); } + + @Override + public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) { + + return transformMap.get(event.tablePath().getFullName()).mapSchemaChangeEvent(event); + } } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractSeaTunnelTransform.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractSeaTunnelTransform.java index 01ce7eaf0a2..70d4c3f8454 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractSeaTunnelTransform.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/common/AbstractSeaTunnelTransform.java @@ -59,6 +59,12 @@ public CatalogTable getProducedCatalogTable() { return outputCatalogTable; } + public void restProducedCatalogTable() { + synchronized (this) { + outputCatalogTable = transformCatalogTable(); + } + } + @Override public List getProducedCatalogTables() { return Collections.singletonList(getProducedCatalogTable()); diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLEngine.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLEngine.java index 62c25be374b..f46f158bf18 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLEngine.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLEngine.java @@ -33,5 +33,9 @@ void init( List transformBySQL(SeaTunnelRow inputRow, SeaTunnelRowType outputRowType); + String getChangeColumnName(String columnName); + default void close() {} + + void resetAllColumnsCount(); } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java index 230541a7bf2..4b86ec6ad45 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/SQLTransform.java @@ -17,6 +17,8 @@ package org.apache.seatunnel.transform.sql; +import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; + import org.apache.seatunnel.api.common.CommonOptions; import org.apache.seatunnel.api.configuration.Option; import org.apache.seatunnel.api.configuration.Options; @@ -27,6 +29,15 @@ import org.apache.seatunnel.api.table.catalog.PhysicalColumn; import org.apache.seatunnel.api.table.catalog.TableIdentifier; import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent; +import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; +import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher; +import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandler; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; @@ -46,6 +57,7 @@ @Slf4j public class SQLTransform extends AbstractCatalogSupportFlatMapTransform { public static final String PLUGIN_NAME = "Sql"; + private final TableSchemaChangeEventHandler tableSchemaChangeEventHandler; public static final Option KEY_QUERY = Options.key("query").stringType().noDefaultValue().withDescription("The query SQL"); @@ -68,6 +80,7 @@ public class SQLTransform extends AbstractCatalogSupportFlatMapTransform { public SQLTransform(@NonNull ReadonlyConfig config, @NonNull CatalogTable catalogTable) { super(catalogTable); + this.tableSchemaChangeEventHandler = new TableSchemaChangeEventDispatcher(); this.query = config.get(KEY_QUERY); if (config.getOptional(KEY_ENGINE).isPresent()) { this.engineType = EngineType.valueOf(config.get(KEY_ENGINE).toUpperCase()); @@ -112,20 +125,23 @@ protected List transformRow(SeaTunnelRow inputRow) { @Override protected TableSchema transformTableSchema() { + return convertTableSchema(inputCatalogTable.getTableSchema()); + } + + protected TableSchema convertTableSchema(TableSchema tableSchema) { tryOpen(); List inputColumnsMapping = new ArrayList<>(); outRowType = sqlEngine.typeMapping(inputColumnsMapping); List outputColumns = Arrays.asList(outRowType.getFieldNames()); TableSchema.Builder builder = TableSchema.builder(); - if (inputCatalogTable.getTableSchema().getPrimaryKey() != null - && outputColumns.containsAll( - inputCatalogTable.getTableSchema().getPrimaryKey().getColumnNames())) { + if (tableSchema.getPrimaryKey() != null + && outputColumns.containsAll(tableSchema.getPrimaryKey().getColumnNames())) { builder.primaryKey(inputCatalogTable.getTableSchema().getPrimaryKey().copy()); } List outputConstraintKeys = - inputCatalogTable.getTableSchema().getConstraintKeys().stream() + tableSchema.getConstraintKeys().stream() .filter( key -> { List constraintColumnNames = @@ -148,7 +164,7 @@ protected TableSchema transformTableSchema() { Column simpleColumn = null; String inputColumnName = inputColumnsMapping.get(i); if (inputColumnName != null) { - for (Column inputColumn : inputCatalogTable.getTableSchema().getColumns()) { + for (Column inputColumn : tableSchema.getColumns()) { if (inputColumnName.equals(inputColumn.getName())) { simpleColumn = inputColumn; break; @@ -176,6 +192,134 @@ protected TableSchema transformTableSchema() { return builder.columns(columns).build(); } + @Override + public SchemaChangeEvent mapSchemaChangeEvent(SchemaChangeEvent event) { + + TableSchema newTableSchema = + tableSchemaChangeEventHandler + .reset(inputCatalogTable.getTableSchema()) + .apply(event); + this.inputCatalogTable = + CatalogTable.of( + inputCatalogTable.getTableId(), + newTableSchema, + inputCatalogTable.getOptions(), + inputCatalogTable.getPartitionKeys(), + inputCatalogTable.getComment()); + sqlEngine.init( + inputTableName, + inputCatalogTable.getTableId().getTableName(), + inputCatalogTable.getSeaTunnelRowType(), + query); + sqlEngine.resetAllColumnsCount(); + restProducedCatalogTable(); + + if (event instanceof AlterTableColumnsEvent) { + AlterTableColumnsEvent alterTableColumnsEvent = (AlterTableColumnsEvent) event; + AlterTableColumnsEvent newEvent = + new AlterTableColumnsEvent( + event.tableIdentifier(), + alterTableColumnsEvent.getEvents().stream() + .map(this::convertName) + .collect(Collectors.toList())); + + newEvent.setJobId(event.getJobId()); + newEvent.setStatement(((AlterTableColumnsEvent) event).getStatement()); + newEvent.setSourceDialectName(((AlterTableColumnsEvent) event).getSourceDialectName()); + if (event.getChangeAfter() != null) { + newEvent.setChangeAfter( + CatalogTable.of( + event.getChangeAfter().getTableId(), event.getChangeAfter())); + } + return newEvent; + } + if (event instanceof AlterTableColumnEvent) { + return convertName((AlterTableColumnEvent) event); + } + return event; + } + + @VisibleForTesting + public AlterTableColumnEvent convertName(AlterTableColumnEvent event) { + AlterTableColumnEvent newEvent = event; + switch (event.getEventType()) { + case SCHEMA_CHANGE_ADD_COLUMN: + AlterTableAddColumnEvent addColumnEvent = (AlterTableAddColumnEvent) event; + newEvent = + new AlterTableAddColumnEvent( + event.tableIdentifier(), + addColumnEvent.getColumn(), + addColumnEvent.isFirst(), + addColumnEvent.getAfterColumn()); + break; + case SCHEMA_CHANGE_DROP_COLUMN: + AlterTableDropColumnEvent dropColumnEvent = (AlterTableDropColumnEvent) event; + newEvent = + new AlterTableDropColumnEvent( + event.tableIdentifier(), convertName(dropColumnEvent.getColumn())); + break; + case SCHEMA_CHANGE_MODIFY_COLUMN: + AlterTableModifyColumnEvent modifyColumnEvent = (AlterTableModifyColumnEvent) event; + newEvent = + new AlterTableModifyColumnEvent( + event.tableIdentifier(), + convertName(modifyColumnEvent.getColumn()), + modifyColumnEvent.isFirst(), + convertName(modifyColumnEvent.getAfterColumn())); + break; + case SCHEMA_CHANGE_CHANGE_COLUMN: + AlterTableChangeColumnEvent changeColumnEvent = (AlterTableChangeColumnEvent) event; + boolean nameChanged = + !changeColumnEvent + .getOldColumn() + .equals(changeColumnEvent.getColumn().getName()); + if (nameChanged) { + log.warn( + "FieldRenameTransform does not support changing column name, " + + "old column name: {}, new column name: {}", + changeColumnEvent.getOldColumn(), + changeColumnEvent.getColumn().getName()); + return changeColumnEvent; + } + + newEvent = + new AlterTableChangeColumnEvent( + event.tableIdentifier(), + convertName(changeColumnEvent.getOldColumn()), + convertName(changeColumnEvent.getColumn()), + changeColumnEvent.isFirst(), + convertName(changeColumnEvent.getAfterColumn())); + break; + default: + log.warn("Unsupported event: {}", event); + return event; + } + + newEvent.setJobId(event.getJobId()); + newEvent.setStatement(event.getStatement()); + newEvent.setSourceDialectName(event.getSourceDialectName()); + if (event.getChangeAfter() != null) { + CatalogTable newChangeAfter = + CatalogTable.of( + event.getChangeAfter().getTableId(), + convertTableSchema(event.getChangeAfter().getTableSchema()), + event.getChangeAfter().getOptions(), + event.getChangeAfter().getPartitionKeys(), + event.getChangeAfter().getComment()); + newEvent.setChangeAfter(newChangeAfter); + } + return newEvent; + } + + @VisibleForTesting + public String convertName(String name) { + return sqlEngine.getChangeColumnName(name); + } + + private Column convertName(Column column) { + return column.rename(convertName(column.getName())); + } + @Override protected TableIdentifier transformTableIdentifier() { return inputCatalogTable.getTableId().copy(); diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java index 9ce552c289a..309d6f63a7d 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLEngine.java @@ -48,13 +48,16 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.ServiceLoader; import java.util.stream.Collectors; public class ZetaSQLEngine implements SQLEngine { private static final Logger log = LoggerFactory.getLogger(ZetaSQLEngine.class); public static final String ESCAPE_IDENTIFIER = "`"; + public static Map columnNameMapping = new HashMap<>(); private String inputTableName; @Nullable private String catalogTableName; @@ -185,23 +188,36 @@ public SeaTunnelRowType typeMapping(List inputColumnsMapping) { for (SelectItem selectItem : selectItems) { if (selectItem.getExpression() instanceof AllColumns) { for (int i = 0; i < inputRowType.getFieldNames().length; i++) { - fieldNames[idx] = cleanEscape(inputRowType.getFieldName(i)); + String fieldName = cleanEscape(inputRowType.getFieldName(i)); + fieldNames[idx] = fieldName; seaTunnelDataTypes[idx] = inputRowType.getFieldType(i); if (inputColumnsMapping != null) { inputColumnsMapping.set(idx, inputRowType.getFieldName(i)); } + columnNameMapping.put(fieldName, fieldName); idx++; } } else { Expression expression = selectItem.getExpression(); if (selectItem.getAlias() != null) { - String aliasName = selectItem.getAlias().getName(); - fieldNames[idx] = cleanEscape(aliasName); + String aliasName = cleanEscape(selectItem.getAlias().getName()); + String fieldName; + fieldNames[idx] = aliasName; + if (expression instanceof Column) { + fieldName = cleanEscape(((Column) expression).getColumnName()); + } else { + fieldName = cleanEscape(expression.toString()); + } + columnNameMapping.put(fieldName, aliasName); } else { if (expression instanceof Column) { - fieldNames[idx] = cleanEscape(((Column) expression).getColumnName()); + String fieldName = cleanEscape(((Column) expression).getColumnName()); + fieldNames[idx] = fieldName; + columnNameMapping.put(fieldName, fieldName); } else { - fieldNames[idx] = cleanEscape(expression.toString()); + String fieldName = cleanEscape(expression.toString()); + fieldNames[idx] = fieldName; + columnNameMapping.put(fieldName, fieldName); } } @@ -226,6 +242,14 @@ public SeaTunnelRowType typeMapping(List inputColumnsMapping) { return outRowType; } + @Override + public String getChangeColumnName(String columnName) { + if (columnNameMapping.containsKey(columnName)) { + return columnNameMapping.get(columnName); + } + return columnName; + } + private static String cleanEscape(String columnName) { if (columnName.startsWith(ESCAPE_IDENTIFIER) && columnName.endsWith(ESCAPE_IDENTIFIER)) { columnName = columnName.substring(1, columnName.length() - 1); @@ -303,4 +327,9 @@ private int countColumnsSize(List> selectItems) { - allColumnsCnt; return allColumnsCount; } + + @Override + public void resetAllColumnsCount() { + allColumnsCount = null; + } } diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java index 2d48f7b28a1..9782cc8e815 100644 --- a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLTransformTest.java @@ -17,12 +17,21 @@ package org.apache.seatunnel.transform.sql; +import org.apache.seatunnel.shade.com.google.common.collect.Lists; + import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.catalog.Column; +import org.apache.seatunnel.api.table.catalog.ConstraintKey; import org.apache.seatunnel.api.table.catalog.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.PrimaryKey; import org.apache.seatunnel.api.table.catalog.TableIdentifier; import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.LocalTimeType; import org.apache.seatunnel.api.table.type.MapType; @@ -34,10 +43,12 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Objects; +import java.util.stream.Collectors; public class SQLTransformTest { @@ -282,4 +293,219 @@ public void testEscapeIdentifier() { BasicType.STRING_TYPE, tableSchema.getColumns().get(1).getDataType()); Assertions.assertEquals("a", result.get(0).getField(1)); } + + @Test + public void testSchemaChange() { + CatalogTable catalogTable = + CatalogTable.of( + TableIdentifier.of(TEST_NAME, TEST_NAME, null, TEST_NAME), + TableSchema.builder() + .column( + PhysicalColumn.of( + "id", + BasicType.LONG_TYPE, + null, + null, + false, + null, + null)) + .column( + PhysicalColumn.of( + "name", + BasicType.STRING_TYPE, + null, + null, + true, + null, + null)) + .column( + PhysicalColumn.of( + "age", + BasicType.LONG_TYPE, + null, + null, + true, + null, + null)) + .primaryKey(PrimaryKey.of("pk1", Arrays.asList("id"))) + .constraintKey( + ConstraintKey.of( + ConstraintKey.ConstraintType.UNIQUE_KEY, + "uk1", + Arrays.asList( + ConstraintKey.ConstraintKeyColumn.of( + "name", + ConstraintKey.ColumnSortType.ASC), + ConstraintKey.ConstraintKeyColumn.of( + "age", + ConstraintKey.ColumnSortType.ASC)))) + .build(), + Collections.emptyMap(), + Collections.singletonList("name"), + null); + + ReadonlyConfig config = + ReadonlyConfig.fromMap( + new HashMap() { + { + put("query", "select * from dual"); + } + }); + List result; + SQLTransform transform = new SQLTransform(config, catalogTable); + result = + transform.transformRow( + new SeaTunnelRow( + new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)})); + List columnNames; + List columnType; + List assertNames; + List assertTypes; + Object[] columnValues; + Object[] assertValue; + + columnNames = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + columnType = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map((e) -> e.getDataType().getSqlType().name()) + .collect(Collectors.toList()); + assertNames = Lists.newArrayList("id", "name", "age"); + assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT"); + + columnValues = result.get(0).getFields(); + assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)}; + Assertions.assertIterableEquals(columnNames, assertNames); + Assertions.assertIterableEquals(columnType, assertTypes); + Assertions.assertArrayEquals(columnValues, assertValue); + + // test add column + AlterTableAddColumnEvent addColumnEvent = + AlterTableAddColumnEvent.add( + catalogTable.getTableId(), + PhysicalColumn.of("f4", BasicType.LONG_TYPE, null, null, true, null, null)); + transform.mapSchemaChangeEvent(addColumnEvent); + + result = + transform.transformRow( + new SeaTunnelRow( + new Object[] { + Integer.valueOf(1), + "Cosmos", + Integer.valueOf(30), + Integer.valueOf(14) + })); + columnNames = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + columnType = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map((e) -> e.getDataType().getSqlType().name()) + .collect(Collectors.toList()); + assertNames = Lists.newArrayList("id", "name", "age", "f4"); + assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "BIGINT"); + + columnValues = result.get(0).getFields(); + assertValue = + new Object[] { + Integer.valueOf(1), "Cosmos", Integer.valueOf(30), Integer.valueOf(14) + }; + Assertions.assertIterableEquals(columnNames, assertNames); + Assertions.assertIterableEquals(columnType, assertTypes); + Assertions.assertArrayEquals(columnValues, assertValue); + + // test modify column + AlterTableModifyColumnEvent modifyColumnEvent = + AlterTableModifyColumnEvent.modify( + catalogTable.getTableId(), + PhysicalColumn.of( + "f4", BasicType.STRING_TYPE, null, null, true, null, null)); + transform.mapSchemaChangeEvent(modifyColumnEvent); + result = + transform.transformRow( + new SeaTunnelRow( + new Object[] { + Integer.valueOf(1), "Cosmos", Integer.valueOf(30), "Cosmos" + })); + columnNames = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + columnType = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map((e) -> e.getDataType().getSqlType().name()) + .collect(Collectors.toList()); + assertNames = Lists.newArrayList("id", "name", "age", "f4"); + assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "STRING"); + + columnValues = result.get(0).getFields(); + assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30), "Cosmos"}; + Assertions.assertIterableEquals(columnNames, assertNames); + Assertions.assertIterableEquals(columnType, assertTypes); + Assertions.assertArrayEquals(columnValues, assertValue); + + // test change column + AlterTableChangeColumnEvent changeColumnEvent = + AlterTableChangeColumnEvent.change( + catalogTable.getTableId(), + "f4", + PhysicalColumn.of("f5", BasicType.INT_TYPE, null, null, true, null, null)); + transform.mapSchemaChangeEvent(changeColumnEvent); + result = + transform.transformRow( + new SeaTunnelRow( + new Object[] { + Integer.valueOf(1), + "Cosmos", + Integer.valueOf(30), + Integer.valueOf(14) + })); + columnNames = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + columnType = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map((e) -> e.getDataType().getSqlType().name()) + .collect(Collectors.toList()); + assertNames = Lists.newArrayList("id", "name", "age", "f5"); + assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT", "INT"); + + columnValues = result.get(0).getFields(); + assertValue = + new Object[] { + Integer.valueOf(1), "Cosmos", Integer.valueOf(30), Integer.valueOf(14) + }; + Assertions.assertIterableEquals(columnNames, assertNames); + Assertions.assertIterableEquals(columnType, assertTypes); + Assertions.assertArrayEquals(columnValues, assertValue); + + // test drop column + AlterTableDropColumnEvent dropColumnEvent = + new AlterTableDropColumnEvent(catalogTable.getTableId(), "f5"); + transform.mapSchemaChangeEvent(dropColumnEvent); + result = + transform.transformRow( + new SeaTunnelRow( + new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)})); + columnNames = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map(Column::getName) + .collect(Collectors.toList()); + columnType = + transform.getProducedCatalogTable().getTableSchema().getColumns().stream() + .map((e) -> e.getDataType().getSqlType().name()) + .collect(Collectors.toList()); + assertNames = Lists.newArrayList("id", "name", "age"); + assertTypes = Lists.newArrayList("BIGINT", "STRING", "BIGINT"); + + columnValues = result.get(0).getFields(); + assertValue = new Object[] {Integer.valueOf(1), "Cosmos", Integer.valueOf(30)}; + Assertions.assertIterableEquals(columnNames, assertNames); + Assertions.assertIterableEquals(columnType, assertTypes); + Assertions.assertArrayEquals(columnValues, assertValue); + } }