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-12355-c6ddef67e4f7e799e012ea2bf73ab9cc869f82f9 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit d91e26a3368f59a37bb6a285b36cde071728e726 Author: Steven Zhao <[email protected]> AuthorDate: Fri Oct 9 09:25:47 2026 +0000 [Fix][Connector-V2][MySQL-CDC] Keep the SET/ENUM option list when rebuilding the source type (#12355) --- .../parser/CustomAlterTableParserListener.java | 18 +++ .../seatunnel/cdc/mysql/utils/MySqlTypeUtils.java | 58 ++++++++++ .../parser/CustomMySqlAntlrDdlParserTest.java | 126 +++++++++++++++++++++ 3 files changed, 202 insertions(+) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java index 60c391f7be..707338db6a 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomAlterTableParserListener.java @@ -243,4 +243,22 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener public org.apache.seatunnel.api.table.catalog.Column toSeatunnelColumn(Column column) { return MySqlTypeUtils.convertToSeaTunnelColumn(column, dbzConnectorConfig); } + + /** + * {@code SET} and {@code ENUM} do not take a length: Debezium keeps their option list in {@link + * Column#enumValues()} and the value of {@link Column#length()} is only a bookkeeping count + * ({@code 2 * options - 1}). Appending it would produce a type such as {@code SET(5)}, which is + * not valid DDL, and that value reaches generated auto-create statements through {@code + * org.apache.seatunnel.api.table.catalog.Column#getSourceType()} (see <a + * href="https://github.com/apache/seatunnel/issues/12354";>issue #12354</a>). Render the option + * list instead, the way the base implementation renders length and scale for other types. + */ + @Override + public String getSourceColumnTypeWithLengthScale(Column column) { + List<String> enumValues = column.enumValues(); + if (enumValues != null && !enumValues.isEmpty()) { + return String.format("%s(%s)", column.typeName(), String.join(",", enumValues)); + } + return SeatunnelDDLParser.super.getSourceColumnTypeWithLengthScale(column); + } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java index 54f352ab73..9a571ec36a 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlTypeUtils.java @@ -31,6 +31,7 @@ import io.debezium.relational.Column; import io.debezium.relational.RelationalDatabaseConnectorConfig; import lombok.extern.slf4j.Slf4j; +import java.util.List; import java.util.Optional; /** Utilities for converting from MySQL types to SeaTunnel types. */ @@ -152,6 +153,20 @@ public class MySqlTypeUtils { builder.columnType(String.format("TINYINT(%s)", column.length())); } break; + case MySqlTypeConverter.MYSQL_ENUM: + case MySqlTypeConverter.MYSQL_SET: + // Debezium reports a bookkeeping length for these types (options * 2 - 1 for SET, + // always 1 for ENUM) rather than the length of the stored value. Forwarding it + // makes MySqlTypeConverter produce a column length of 1 for an ENUM and options * + // 2 - 1 for a SET, which sinks that rebuild the type from the column length (any + // non-MySQL sink, via JdbcDialect#applySchemaChange -> reconvert) turn into an + // undersized column. Derive the real length from the option list instead; it is + // available on the DDL path and, for completeness, whenever the column carries it. + long optionListLength = maxOptionListLength(column); + if (optionListLength > 0) { + builder.length(optionListLength).precision(optionListLength); + } + break; default: break; } @@ -162,4 +177,47 @@ public class MySqlTypeUtils { return (intTypeNarrowing ? MySqlTypeConverter.DEFAULT_INSTANCE : NO_INT_NARROWING_CONVERTER) .convert(builder.build()); } + + /** + * Computes the longest value a {@code SET} / {@code ENUM} column can store, in characters: the + * longest member for an {@code ENUM}, and the commas plus the sum of all members for a {@code + * SET}. + * + * @param column Debezium column, whose option list is populated on the DDL-parsing path + * @return the derived length, or {@code -1} when no option list is available + */ + private static long maxOptionListLength(Column column) { + List<String> enumValues = column.enumValues(); + if (enumValues == null || enumValues.isEmpty()) { + return -1L; + } + boolean isSet = MySqlTypeConverter.MYSQL_SET.equalsIgnoreCase(column.typeName()); + long totalLength = 0L; + long maxLength = 0L; + for (String enumValue : enumValues) { + long length = unquotedValueLength(enumValue); + totalLength += length; + maxLength = Math.max(maxLength, length); + } + return isSet ? totalLength + enumValues.size() - 1L : maxLength; + } + + /** + * Returns the number of characters of a member of an option list as it appears in the DDL, so + * the surrounding quotes and any doubled quote escape are removed before measuring. + * + * @param enumValue raw option text collected by the DDL parser + * @return the stored length of that member + */ + private static long unquotedValueLength(String enumValue) { + if (enumValue == null) { + return 0L; + } + if (enumValue.length() >= 2 + && enumValue.charAt(0) == '\'' + && enumValue.charAt(enumValue.length() - 1) == '\'') { + return enumValue.substring(1, enumValue.length() - 1).replace("''", "'").length(); + } + return enumValue.length(); + } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java index d8caa91bd9..515d8706fd 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java @@ -17,9 +17,14 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.parser; +import org.apache.seatunnel.api.table.catalog.Column; +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.AlterTableCommentEvent; import org.apache.seatunnel.api.table.schema.event.AlterTableEvent; +import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent; +import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfig; import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfigFactory; @@ -91,6 +96,127 @@ public class CustomMySqlAntlrDdlParserTest { Assertions.assertEquals("Product catalog table", event.getNewComment()); } + /** + * Regression test for <a href="https://github.com/apache/seatunnel/issues/12354";>#12354</a>. + * + * <p>Debezium reports the bare type name for {@code SET} / {@code ENUM} and keeps the option + * list on the column, but the base {@code getSourceColumnTypeWithLengthScale} renders the + * bookkeeping length as {@code SET(5)} / {@code ENUM(1)}. That string is what {@code + * org.apache.seatunnel.api.table.catalog.Column#getSourceType()} exposes, and {@code + * MysqlCreateTableSqlBuilder} emits it verbatim into generated auto-create DDL, which MySQL + * rejects. The option list therefore has to survive as part of the rebuilt type expression, and + * the column length must not stay on the bookkeeping value either, since a sink that rebuilds + * the type from the length would create an undersized column. + */ + @Test + void testParseAlterTableAddSetAndEnumColumnKeepsOptionList() { + MySqlSourceConfigFactory factory = new MySqlSourceConfigFactory(); + factory.hostname("localhost"); + factory.username("test"); + factory.password("test"); + CustomMySqlAntlrDdlParser parser = + new CustomMySqlAntlrDdlParser(factory.create(0).getDbzConnectorConfig()); + parser.setCurrentDatabase("test_db"); + parser.parse( + "ALTER TABLE products ADD COLUMN c_set SET('a','b','c') NULL, " + + "ADD COLUMN c_enum ENUM('x','y') NULL, " + + "ADD COLUMN c_enum_multi ENUM('active','inactive') NULL, " + + "ADD COLUMN c_set_single SET('only') NULL, " + + "ADD COLUMN c_set_escaped SET('a,b','it''s') NULL, " + + "ADD COLUMN c_enum_charset ENUM('x','y') CHARACTER SET utf8mb4 NULL", + new Tables()); + + List<AlterTableColumnEvent> events = parser.getAndClearParsedColumnEvents(); + Assertions.assertEquals(6, events.size()); + + Column setColumn = ((AlterTableAddColumnEvent) events.get(0)).getColumn(); + Assertions.assertEquals("c_set", setColumn.getName()); + Assertions.assertEquals("SET('a','b','c')", setColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, setColumn.getDataType()); + // The longest value of this SET is "a,b,c": three members plus two separators. + Assertions.assertEquals(5L, setColumn.getColumnLength()); + + Column enumColumn = ((AlterTableAddColumnEvent) events.get(1)).getColumn(); + Assertions.assertEquals("c_enum", enumColumn.getName()); + Assertions.assertEquals("ENUM('x','y')", enumColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, enumColumn.getDataType()); + Assertions.assertEquals(1L, enumColumn.getColumnLength()); + + // Multi-character members are the case where the real length and Debezium's bookkeeping + // count (options * 2 - 1 for SET, 1 for ENUM) differ, so the length assertion below only + // passes for the right reason with members like these. + Column multiEnumColumn = ((AlterTableAddColumnEvent) events.get(2)).getColumn(); + Assertions.assertEquals("c_enum_multi", multiEnumColumn.getName()); + Assertions.assertEquals("ENUM('active','inactive')", multiEnumColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, multiEnumColumn.getDataType()); + Assertions.assertEquals(8L, multiEnumColumn.getColumnLength()); + + // A single-option SET would also look plausible with a wrong fallback: its bookkeeping + // length is 1, not the four characters of its only member. + Column singleSetColumn = ((AlterTableAddColumnEvent) events.get(3)).getColumn(); + Assertions.assertEquals("c_set_single", singleSetColumn.getName()); + Assertions.assertEquals("SET('only')", singleSetColumn.getSourceType()); + Assertions.assertEquals(4L, singleSetColumn.getColumnLength()); + + // The option list is re-emitted verbatim, so an embedded comma and an escaped quote have to + // round-trip exactly; the stored length counts the unescaped members plus the separator. + Column escapedSetColumn = ((AlterTableAddColumnEvent) events.get(4)).getColumn(); + Assertions.assertEquals("c_set_escaped", escapedSetColumn.getName()); + Assertions.assertEquals("SET('a,b','it''s')", escapedSetColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, escapedSetColumn.getDataType()); + Assertions.assertEquals(8L, escapedSetColumn.getColumnLength()); + + // The rebuilt expression carries the option list only: a CHARACTER SET clause on the + // source column is not part of it. Pinned here so the behaviour is explicit rather than + // accidental; the sink then applies its own default charset for the added column. + Column charsetEnumColumn = ((AlterTableAddColumnEvent) events.get(5)).getColumn(); + Assertions.assertEquals("c_enum_charset", charsetEnumColumn.getName()); + Assertions.assertEquals("ENUM('x','y')", charsetEnumColumn.getSourceType()); + Assertions.assertEquals(1L, charsetEnumColumn.getColumnLength()); + } + + /** + * Same regression as {@link #testParseAlterTableAddSetAndEnumColumnKeepsOptionList()}, reached + * through the other two DDL forms that share {@code toSeatunnelColumnWithFullTypeInfo}: {@code + * MODIFY COLUMN} and {@code CHANGE COLUMN}. {@code CHANGE} additionally renames the column, so + * the rebuilt type has to survive that too. A length-bearing type is kept in the same statement + * to pin the fallback to the base implementation. + */ + @Test + void testParseAlterTableModifyAndChangeSetAndEnumColumnKeepOptionList() { + MySqlSourceConfigFactory factory = new MySqlSourceConfigFactory(); + factory.hostname("localhost"); + factory.username("test"); + factory.password("test"); + CustomMySqlAntlrDdlParser parser = + new CustomMySqlAntlrDdlParser(factory.create(0).getDbzConnectorConfig()); + parser.setCurrentDatabase("test_db"); + parser.parse( + "ALTER TABLE products MODIFY COLUMN c_set SET('a','b','c') NULL, " + + "CHANGE COLUMN c_enum_src c_enum ENUM('x','y') NULL, " + + "MODIFY COLUMN c_varchar VARCHAR(64) NULL", + new Tables()); + + List<AlterTableColumnEvent> events = parser.getAndClearParsedColumnEvents(); + Assertions.assertEquals(3, events.size()); + + Column modifiedColumn = ((AlterTableModifyColumnEvent) events.get(0)).getColumn(); + Assertions.assertEquals("c_set", modifiedColumn.getName()); + Assertions.assertEquals("SET('a','b','c')", modifiedColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, modifiedColumn.getDataType()); + + AlterTableChangeColumnEvent changeColumnEvent = (AlterTableChangeColumnEvent) events.get(1); + Assertions.assertEquals("c_enum_src", changeColumnEvent.getOldColumn()); + Column renamedColumn = changeColumnEvent.getColumn(); + Assertions.assertEquals("c_enum", renamedColumn.getName()); + Assertions.assertEquals("ENUM('x','y')", renamedColumn.getSourceType()); + Assertions.assertEquals(BasicType.STRING_TYPE, renamedColumn.getDataType()); + + Column varcharColumn = ((AlterTableModifyColumnEvent) events.get(2)).getColumn(); + Assertions.assertEquals("c_varchar", varcharColumn.getName()); + Assertions.assertEquals("VARCHAR(64)", varcharColumn.getSourceType()); + } + @Test void testParsePropagatesDdlErrors() { CustomMySqlAntlrDdlParser parser = new ParserWithTreeWalkError();
