This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12373-48160ec297b7daacf7821e47c426feda32c191d8 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 0064881f158af8dff33b1c9c7fbc1f4660361b66 Author: Steven Zhao <[email protected]> AuthorDate: Sun Oct 11 09:45:53 2026 +0000 [Test][E2E][MySQL-CDC] Cover the SET/ENUM DDL a same-dialect MySQL sink executes (#12373) --- .../cdc/mysql/MysqlCDCWithSchemaChangeIT.java | 95 ++++++++++++++++++++++ .../test/resources/ddl/add_set_enum_columns.sql | 40 +++++++++ .../test/resources/ddl/set_enum_schema_change.sql | 52 ++++++++++++ ...qlcdc_to_mysql_with_set_enum_schema_change.conf | 54 ++++++++++++ 4 files changed, 241 insertions(+) diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java index d4c16d338d..ad1fd293ee 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java @@ -134,6 +134,21 @@ public class MysqlCDCWithSchemaChangeIT extends TestSuiteBase implements TestRes private static final String INCREMENTAL_READ_MARKER = "Start incremental read task for incremental split"; + /** + * Source table of the SET / ENUM DDL case. It carries SET / ENUM columns before the job starts, + * so the snapshot path exercises them, and is altered again while the job runs. + */ + private static final String SET_ENUM_SOURCE_TABLE = "products_with_set_enum"; + /** Sink table the SET / ENUM DDL case evolves. */ + private static final String SET_ENUM_SINK_TABLE = "mysql_cdc_e2e_sink_table_with_set_enum"; + /** Job config for the SET / ENUM DDL case. */ + private static final String SET_ENUM_SCHEMA_CHANGE_JOB_CONFIG = + "/mysqlcdc_to_mysql_with_set_enum_schema_change.conf"; + /** SQL template that resets the SET / ENUM source and sink tables to their initial shape. */ + private static final String SET_ENUM_SCHEMA_CHANGE_INIT_TEMPLATE = "set_enum_schema_change"; + /** SQL template that adds and modifies SET / ENUM columns while the job is running. */ + private static final String SET_ENUM_SCHEMA_CHANGE_ALTER_TEMPLATE = "add_set_enum_columns"; + private static final MySqlContainer MYSQL_CONTAINER = createMySqlContainer(MySqlVersion.V8_0); private final UniqueDatabase shopDatabase = @@ -424,6 +439,52 @@ public class MysqlCDCWithSchemaChangeIT extends TestSuiteBase implements TestRes "drop.column was excluded, so the sink must keep the column the source dropped"); } + /** + * Covers the DDL a same-dialect MySQL sink builds for {@code SET} / {@code ENUM} columns. + * + * <p>When the source dialect matches the sink dialect, schema-change statements are built from + * the column's source type expression rather than from the reconverted type, so an expression + * that lost the option list ({@code SET(5)} instead of {@code SET('a','b','c')}) is only + * rejected once MySQL executes it. This case therefore alters the source table while the job is + * running and compares the resulting column definitions on both sides. + */ + @Order(5) + @TestTemplate + public void testMysqlCdcWithSetAndEnumColumnSchemaChange(TestContainer container) + throws IOException, InterruptedException { + String jobId = String.valueOf(JobIdGenerator.newJobId()); + shopDatabase.setTemplateName(SET_ENUM_SCHEMA_CHANGE_INIT_TEMPLATE).createAndInitialize(); + + CompletableFuture.runAsync( + () -> { + try { + container.executeJob(SET_ENUM_SCHEMA_CHANGE_JOB_CONFIG, jobId); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + }); + + // SET / ENUM columns that exist before the job starts must replicate as STRING values + assertTableStructureAndData(MYSQL_DATABASE, SET_ENUM_SOURCE_TABLE, SET_ENUM_SINK_TABLE); + + // add SET / ENUM columns and widen an existing SET column while the job is running + waitForIncrementalRead(container, MYSQL_DATABASE + "." + SET_ENUM_SOURCE_TABLE); + shopDatabase.setTemplateName(SET_ENUM_SCHEMA_CHANGE_ALTER_TEMPLATE).createAndInitialize(); + assertTableStructureAndData( + MYSQL_DATABASE, + SET_ENUM_SOURCE_TABLE, + MYSQL_DATABASE, + SET_ENUM_SINK_TABLE, + SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS); + + // explicit per-column assertion, so a regression reports SET(5) instead of a DESCRIBE diff + assertColumnTypeEquals(SET_ENUM_SOURCE_TABLE, SET_ENUM_SINK_TABLE, "c_set"); + assertColumnTypeEquals(SET_ENUM_SOURCE_TABLE, SET_ENUM_SINK_TABLE, "c_enum"); + assertColumnTypeEquals(SET_ENUM_SOURCE_TABLE, SET_ENUM_SINK_TABLE, "c_set_added"); + assertColumnTypeEquals(SET_ENUM_SOURCE_TABLE, SET_ENUM_SINK_TABLE, "c_enum_added"); + } + // Asserts the sink row with the given id exists and its name matches expectedName. private void assertSinkNameEquals(List<List<Object>> sinkRows, int id, Object expectedName) { Optional<List<Object>> row = @@ -853,4 +914,38 @@ public class MysqlCDCWithSchemaChangeIT extends TestSuiteBase implements TestRes throw new RuntimeException(e); } } + + /** + * Asserts that the source and the sink define the column identically. Reading {@code + * information_schema.COLUMNS.COLUMN_TYPE} keeps the failure message focused on the + * reconstructed type expression, which is what the sink writes into the statements it executes. + */ + private void assertColumnTypeEquals(String sourceTable, String sinkTable, String columnName) { + Assertions.assertEquals( + queryColumnType(MYSQL_DATABASE, sourceTable, columnName), + queryColumnType(MYSQL_DATABASE, sinkTable, columnName), + "Column definition of " + columnName + " differs between source and sink"); + } + + /** Reads the full type expression of one column from {@code information_schema}. */ + private String queryColumnType(String database, String table, String columnName) { + List<List<Object>> rows = + query( + String.format( + "SELECT COLUMN_TYPE FROM information_schema.COLUMNS" + + " WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s'" + + " AND COLUMN_NAME = '%s'", + database, table, columnName)); + Assertions.assertEquals( + 1, + rows.size(), + "Cannot read " + + database + + "." + + table + + "." + + columnName + + " from information_schema"); + return String.valueOf(rows.get(0).get(0)); + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_set_enum_columns.sql b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_set_enum_columns.sql new file mode 100644 index 0000000000..b271b91bc9 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_set_enum_columns.sql @@ -0,0 +1,40 @@ +-- +-- 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; + +-- ADD / MODIFY of SET and ENUM columns reach the sink as the DDL the job builds from the column's +-- source type expression, so these statements are only valid when that expression keeps the option +-- list (SET('a','b')) instead of the Debezium bookkeeping length (SET(5)). +-- +-- The option lists avoid the word "unsigned" for the same reason as the fixture: a literal such as +-- 'NO_UNSIGNED_SUBTRACTION' is mis-read as the UNSIGNED attribute until #12333 lands, which would +-- make this case depend on a second, unrelated PR. +alter table products_with_set_enum + add column c_set_added set('a','b') null, + add column c_enum_added enum('x','y') null, + modify column c_set set('REAL_AS_FLOAT','PIPES_AS_CONCAT','ANSI_QUOTES') null; + +insert into products_with_set_enum (id, name, weight, c_set, c_enum, c_set_added, c_enum_added) +values (201, 'added after the schema change', 0.5, 'ANSI_QUOTES', 'signed', 'a,b', 'x'); + +update products_with_set_enum set name = 'scooter updated' where id = 101; +delete from products_with_set_enum where id = 102; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/set_enum_schema_change.sql b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/set_enum_schema_change.sql new file mode 100644 index 0000000000..28fd099721 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/set_enum_schema_change.sql @@ -0,0 +1,52 @@ +-- +-- 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_with_set_enum; +CREATE TABLE products_with_set_enum ( + id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(255) NOT NULL DEFAULT 'SeaTunnel', + weight FLOAT, + c_set set('REAL_AS_FLOAT','PIPES_AS_CONCAT') DEFAULT NULL, + c_enum enum('signed','sealed') DEFAULT NULL +); + +insert into products_with_set_enum (id, name, weight, c_set, c_enum) +values (101, 'scooter', 3.14, 'REAL_AS_FLOAT', 'signed'), + (102, 'car battery', 8.1, 'REAL_AS_FLOAT,PIPES_AS_CONCAT', 'sealed'); + +-- Kept in step with the source table: the other cases in this suite pre-create their sink tables +-- so that the DESCRIBE comparison can cover the whole table, and an auto-created sink table is +-- rendered without AUTO_INCREMENT. +-- +-- The option lists deliberately avoid the word "unsigned": a literal such as +-- 'NO_UNSIGNED_SUBTRACTION' trips the catalog-side substring check that #12333 fixes, which would +-- make this case depend on a second, unrelated PR just to reach the snapshot. Catalog coverage for +-- that shape belongs to connector-jdbc (see #12333). +drop table if exists mysql_cdc_e2e_sink_table_with_set_enum; +CREATE TABLE mysql_cdc_e2e_sink_table_with_set_enum ( + id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(255) NOT NULL DEFAULT 'SeaTunnel', + weight FLOAT, + c_set set('REAL_AS_FLOAT','PIPES_AS_CONCAT') DEFAULT NULL, + c_enum enum('signed','sealed') DEFAULT NULL +); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_set_enum_schema_change.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_set_enum_schema_change.conf new file mode 100644 index 0000000000..2ebd27630b --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_set_enum_schema_change.conf @@ -0,0 +1,54 @@ +# +# 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 validates the DDL a same-dialect MySQL sink generates for SET / ENUM columns. +###### The sink is MySQL, so schema-change statements are built from the column's source type +###### expression instead of the reconverted type, which is the path a malformed expression such +###### as SET(5) used to break. +###### + +env { + # You can set engine configuration here + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + MySQL-CDC { + server-id = 5710-5715 + username = "st_user_source" + password = "mysqlpw" + table-names = ["shop.products_with_set_enum"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/shop" + + schema-changes.enabled = true + } +} + +sink { + jdbc { + url = "jdbc:mysql://mysql_cdc_e2e:3306/shop" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = shop + table = mysql_cdc_e2e_sink_table_with_set_enum + primary_keys = ["id"] + } +}
