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-12608-3556e91cc9220f5e4c0537d960ff6c57d612e072 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7a1d727b8f95a9dfd7a9b7b0c6d93f5622456f0d Author: Pravin Jaywantrao Maske <[email protected]> AuthorDate: Thu Oct 8 14:47:35 2026 +0000 [Fix][Connector-V2][MySQL-CDC] Keep NULL in nullable unique key columns of tables without a primary key (#12608) --- docs/en/connectors/source/MySQL-CDC.md | 2 + .../introduction/concepts/incompatible-changes.md | 8 + docs/zh/connectors/source/MySQL-CDC.md | 2 + .../introduction/concepts/incompatible-changes.md | 8 + .../connector/mysql/antlr/MySqlAntlrDdlParser.java | 557 +++++++++++++++++++++ .../seatunnel/cdc/mysql/utils/MySqlSchemaTest.java | 163 ++++++ .../cdc/mysql/AbstractMysqlCDCITBase.java | 149 ++++++ .../ddl/nullable_unique_key_null_value.sql | 135 +++++ ..._mysql_with_nullable_unique_key_null_value.conf | 46 ++ 9 files changed, 1070 insertions(+) diff --git a/docs/en/connectors/source/MySQL-CDC.md b/docs/en/connectors/source/MySQL-CDC.md index 8c73533b18..12c7cb75eb 100644 --- a/docs/en/connectors/source/MySQL-CDC.md +++ b/docs/en/connectors/source/MySQL-CDC.md @@ -513,6 +513,8 @@ source { Without a usable primary key (configured or physical) the connector cannot safely apply UPDATE/DELETE events. Use this mode only for append-only workloads or when downstream sink behavior does not depend on row identity. +NULL values in a nullable unique key column, or in a nullable column configured in `table-names-config.primaryKeys`, are emitted as NULL in the snapshot and binlog phases. A NULL cannot identify a row, so choose a NOT NULL column as the key when UPDATE/DELETE events must be applied by key. + ### Start From a Specific Binlog Offset Use `startup.mode = "specific"` when the first record must be read from a known binlog file and position. diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 340ebb8492..875e8b20d1 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -5,6 +5,14 @@ You need to check this document before you upgrade to related version. ## dev +### MySQL-CDC: NULL in a nullable unique key column is emitted as NULL + +- **Behavior change: MySQL-CDC emits NULL instead of the type default for nullable unique key columns of tables without a primary key** + - **Affected component**: `seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql` + - **Description**: For a table without a primary key, the MySQL DDL parser (Debezium 1.9.8) promoted the first unique key to the primary key and marked its columns NOT NULL, although the database allows NULL in them. NULL values in these columns were emitted as the type default (for example `0`, `''` or the epoch) in the snapshot and binlog phases, including after a binlog DDL such as `ALTER TABLE ... ADD UNIQUE KEY` or `CREATE UNIQUE INDEX`. These columns now keep their declared nulla [...] + - **Impact**: Sinks now receive NULL where they previously received the type default. A sink column declared `NOT NULL`, or a sink `primary_keys` that includes such a column, can reject these rows. + - **Migration Guide**: Make the affected sink columns nullable, and do not use a nullable column as the sink primary key. A job restored from a checkpoint that already holds the schema of such a table (saved after a binlog DDL on it) keeps the old behavior for that table until it is started without that checkpoint. + ### DuckDB BIT and ENUM automatic DDL - Scalar `BIT` and `ENUM` columns with no catalog length now retain an unspecified STRING length diff --git a/docs/zh/connectors/source/MySQL-CDC.md b/docs/zh/connectors/source/MySQL-CDC.md index 8a7b26c905..52efb9a442 100644 --- a/docs/zh/connectors/source/MySQL-CDC.md +++ b/docs/zh/connectors/source/MySQL-CDC.md @@ -506,6 +506,8 @@ source { 上述示例演示的是"逻辑主键"场景：源表本身没有物理主键，但通过 `table-names-config.primaryKeys` 显式声明了一列作为稳定行标识，并启用 `exactly_once = true`，让快照阶段与 binlog 阶段都使用同一逻辑主键。只有当被声明的列在源数据中确实保持唯一时，UPDATE/DELETE 才能被正确路由；如果源数据中存在重复值，行为将不再可靠。 +可空唯一键列中的 NULL 值，以及 `table-names-config.primaryKeys` 中配置的可空列中的 NULL 值，在快照阶段和 binlog 阶段都会按 NULL 输出。NULL 无法标识一行数据，因此当需要按键应用 UPDATE/DELETE 事件时，请选择 NOT NULL 的列作为键。 + ### 从指定 Binlog 位置启动 当需要从明确的 binlog 文件和位置开始读取时，可以使用 `startup.mode = "specific"`。 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 7facd5f3c2..d3e75a6aa3 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -4,6 +4,14 @@ ## dev +### MySQL-CDC：可空唯一键列中的 NULL 按 NULL 输出 + +- **行为变更：对于没有主键的表，MySQL-CDC 对可空唯一键列输出 NULL，而不是类型默认值** + - **影响组件**：`seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql` + - **描述**：对于没有主键的表，MySQL DDL 解析器（Debezium 1.9.8）会把第一个唯一键提升为主键，并把其中的列标记为 NOT NULL，尽管数据库允许这些列为 NULL。这些列中的 NULL 值在快照阶段和 binlog 阶段会被输出为类型默认值（例如 `0`、`''` 或纪元时间），在 binlog 中出现 `ALTER TABLE ... ADD UNIQUE KEY` 或 `CREATE UNIQUE INDEX` 等 DDL 之后也是如此。现在这些列保持声明时的可空性，NULL 按 NULL 输出。真正的主键列不受影响。 + - **影响**：Sink 现在会在原来收到类型默认值的位置收到 NULL。如果 Sink 列声明为 `NOT NULL`，或 Sink 的 `primary_keys` 包含这样的列，这些行可能会被拒绝。 + - **迁移指南**：将受影响的 Sink 列改为可空，并且不要使用可空列作为 Sink 主键。如果作业从一个已经保存了此类表结构的 checkpoint 恢复（该表在 binlog 中出现过 DDL 之后保存），该表会保持旧行为，直到作业不使用该 checkpoint 重新启动。 + ### DuckDB BIT 和 ENUM 自动建表 - Catalog 未提供长度时，标量 `BIT` 和 `ENUM` 列现在保留未指定的 STRING 长度，不再使用原来的 1/255 回退值。 diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java new file mode 100644 index 0000000000..68f163073a --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/io/debezium/connector/mysql/antlr/MySqlAntlrDdlParser.java @@ -0,0 +1,557 @@ +/* + * 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 io.debezium.connector.mysql.antlr; + +import org.antlr.v4.runtime.CharStream; +import org.antlr.v4.runtime.CommonTokenStream; +import org.antlr.v4.runtime.tree.ParseTree; + +import com.mysql.cj.CharsetMapping; +import io.debezium.antlr.AntlrDdlParser; +import io.debezium.antlr.AntlrDdlParserListener; +import io.debezium.antlr.DataTypeResolver; +import io.debezium.antlr.DataTypeResolver.DataTypeEntry; +import io.debezium.connector.mysql.MySqlSystemVariables; +import io.debezium.connector.mysql.MySqlValueConverters; +import io.debezium.connector.mysql.antlr.listener.MySqlAntlrDdlParserListener; +import io.debezium.ddl.parser.mysql.generated.MySqlLexer; +import io.debezium.ddl.parser.mysql.generated.MySqlParser; +import io.debezium.ddl.parser.mysql.generated.MySqlParser.CharsetNameContext; +import io.debezium.ddl.parser.mysql.generated.MySqlParser.CollationNameContext; +import io.debezium.relational.Column; +import io.debezium.relational.ColumnEditor; +import io.debezium.relational.SystemVariables; +import io.debezium.relational.TableEditor; +import io.debezium.relational.TableId; +import io.debezium.relational.Tables.TableFilter; + +import java.sql.Types; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.stream.Collectors; + +/** + * An ANTLR based parser for MySQL DDL statements. + * + * <p>Copied from Debezium project(1.9.8.Final) to keep the columns of a nullable unique key + * optional when it is promoted to the primary key of a table without one. Otherwise NULL values in + * these columns are emitted as the type default (for example 0). Debezium 2.x avoids it by not + * promoting such a unique key at all ({@code parseUniqueIndexColumnNames}). + * + * <p>{@link #parsePrimaryIndexColumnNames} is the only modified method: only a real primary key + * marks its columns NOT NULL. Re-apply this change, or drop this copy, when upgrading Debezium. + * + * @author Roman Kuchár <[email protected]>. + */ +public class MySqlAntlrDdlParser extends AntlrDdlParser<MySqlLexer, MySqlParser> { + + private final ConcurrentMap<String, String> charsetNameForDatabase = new ConcurrentHashMap<>(); + private final MySqlValueConverters converters; + private final TableFilter tableFilter; + + public MySqlAntlrDdlParser() { + this(null, TableFilter.includeAll()); + } + + public MySqlAntlrDdlParser(MySqlValueConverters converters) { + this(converters, TableFilter.includeAll()); + } + + public MySqlAntlrDdlParser(MySqlValueConverters converters, TableFilter tableFilter) { + this(true, false, false, converters, tableFilter); + } + + public MySqlAntlrDdlParser( + boolean throwErrorsFromTreeWalk, + boolean includeViews, + boolean includeComments, + MySqlValueConverters converters, + TableFilter tableFilter) { + super(throwErrorsFromTreeWalk, includeViews, includeComments); + systemVariables = new MySqlSystemVariables(); + this.converters = converters; + this.tableFilter = tableFilter; + } + + @Override + protected ParseTree parseTree(MySqlParser parser) { + return parser.root(); + } + + @Override + protected AntlrDdlParserListener createParseTreeWalkerListener() { + return new MySqlAntlrDdlParserListener(this); + } + + @Override + protected MySqlLexer createNewLexerInstance(CharStream charStreams) { + return new MySqlLexer(charStreams); + } + + @Override + protected MySqlParser createNewParserInstance(CommonTokenStream commonTokenStream) { + return new MySqlParser(commonTokenStream); + } + + @Override + protected SystemVariables createNewSystemVariablesInstance() { + return new MySqlSystemVariables(); + } + + @Override + protected boolean isGrammarInUpperCase() { + return true; + } + + @Override + protected DataTypeResolver initializeDataTypeResolver() { + DataTypeResolver.Builder dataTypeResolverBuilder = new DataTypeResolver.Builder(); + + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.StringDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.CHAR, MySqlParser.CHAR), + new DataTypeEntry(Types.VARCHAR, MySqlParser.CHAR, MySqlParser.VARYING), + new DataTypeEntry(Types.VARCHAR, MySqlParser.VARCHAR), + new DataTypeEntry(Types.VARCHAR, MySqlParser.TINYTEXT), + new DataTypeEntry(Types.VARCHAR, MySqlParser.TEXT), + new DataTypeEntry(Types.VARCHAR, MySqlParser.MEDIUMTEXT), + new DataTypeEntry(Types.VARCHAR, MySqlParser.LONGTEXT), + new DataTypeEntry(Types.NCHAR, MySqlParser.NCHAR), + new DataTypeEntry(Types.NVARCHAR, MySqlParser.NCHAR, MySqlParser.VARYING), + new DataTypeEntry(Types.NVARCHAR, MySqlParser.NVARCHAR), + new DataTypeEntry(Types.CHAR, MySqlParser.CHAR, MySqlParser.BINARY), + new DataTypeEntry(Types.VARCHAR, MySqlParser.VARCHAR, MySqlParser.BINARY), + new DataTypeEntry(Types.VARCHAR, MySqlParser.TINYTEXT, MySqlParser.BINARY), + new DataTypeEntry(Types.VARCHAR, MySqlParser.TEXT, MySqlParser.BINARY), + new DataTypeEntry( + Types.VARCHAR, MySqlParser.MEDIUMTEXT, MySqlParser.BINARY), + new DataTypeEntry(Types.VARCHAR, MySqlParser.LONGTEXT, MySqlParser.BINARY), + new DataTypeEntry(Types.NCHAR, MySqlParser.NCHAR, MySqlParser.BINARY), + new DataTypeEntry(Types.NVARCHAR, MySqlParser.NVARCHAR, MySqlParser.BINARY), + new DataTypeEntry(Types.CHAR, MySqlParser.CHARACTER), + new DataTypeEntry( + Types.VARCHAR, MySqlParser.CHARACTER, MySqlParser.VARYING))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.NationalStringDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.NVARCHAR, MySqlParser.NATIONAL, MySqlParser.VARCHAR) + .setSuffixTokens(MySqlParser.BINARY), + new DataTypeEntry(Types.NCHAR, MySqlParser.NATIONAL, MySqlParser.CHARACTER) + .setSuffixTokens(MySqlParser.BINARY), + new DataTypeEntry(Types.NVARCHAR, MySqlParser.NCHAR, MySqlParser.VARCHAR) + .setSuffixTokens(MySqlParser.BINARY))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.NationalVaryingStringDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry( + Types.NVARCHAR, + MySqlParser.NATIONAL, + MySqlParser.CHAR, + MySqlParser.VARYING), + new DataTypeEntry( + Types.NVARCHAR, + MySqlParser.NATIONAL, + MySqlParser.CHARACTER, + MySqlParser.VARYING))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.DimensionDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.SMALLINT, MySqlParser.TINYINT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.SMALLINT, MySqlParser.INT1) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.SMALLINT, MySqlParser.SMALLINT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.SMALLINT, MySqlParser.INT2) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.MEDIUMINT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.INT3) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.MIDDLEINT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.INT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.INTEGER) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.INTEGER, MySqlParser.INT4) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.BIGINT, MySqlParser.BIGINT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.BIGINT, MySqlParser.INT8) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.REAL, MySqlParser.REAL) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.DOUBLE, MySqlParser.DOUBLE) + .setSuffixTokens( + MySqlParser.PRECISION, + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.DOUBLE, MySqlParser.FLOAT8) + .setSuffixTokens( + MySqlParser.PRECISION, + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.FLOAT, MySqlParser.FLOAT) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.FLOAT, MySqlParser.FLOAT4) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL), + new DataTypeEntry(Types.DECIMAL, MySqlParser.DECIMAL) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL) + .setDefaultLengthScaleDimension(10, 0), + new DataTypeEntry(Types.DECIMAL, MySqlParser.DEC) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL) + .setDefaultLengthScaleDimension(10, 0), + new DataTypeEntry(Types.DECIMAL, MySqlParser.FIXED) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL) + .setDefaultLengthScaleDimension(10, 0), + new DataTypeEntry(Types.NUMERIC, MySqlParser.NUMERIC) + .setSuffixTokens( + MySqlParser.SIGNED, + MySqlParser.UNSIGNED, + MySqlParser.ZEROFILL) + .setDefaultLengthScaleDimension(10, 0), + new DataTypeEntry(Types.BIT, MySqlParser.BIT), + new DataTypeEntry(Types.TIME, MySqlParser.TIME), + new DataTypeEntry(Types.TIMESTAMP_WITH_TIMEZONE, MySqlParser.TIMESTAMP), + new DataTypeEntry(Types.TIMESTAMP, MySqlParser.DATETIME), + new DataTypeEntry(Types.BINARY, MySqlParser.BINARY), + new DataTypeEntry(Types.VARBINARY, MySqlParser.VARBINARY), + new DataTypeEntry(Types.BLOB, MySqlParser.BLOB), + new DataTypeEntry(Types.INTEGER, MySqlParser.YEAR))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.SimpleDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.DATE, MySqlParser.DATE), + new DataTypeEntry(Types.BLOB, MySqlParser.TINYBLOB), + new DataTypeEntry(Types.BLOB, MySqlParser.MEDIUMBLOB), + new DataTypeEntry(Types.BLOB, MySqlParser.LONGBLOB), + new DataTypeEntry(Types.BOOLEAN, MySqlParser.BOOL), + new DataTypeEntry(Types.BOOLEAN, MySqlParser.BOOLEAN), + new DataTypeEntry(Types.BIGINT, MySqlParser.SERIAL))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.CollectionDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.CHAR, MySqlParser.ENUM) + .setSuffixTokens(MySqlParser.BINARY), + new DataTypeEntry(Types.CHAR, MySqlParser.SET) + .setSuffixTokens(MySqlParser.BINARY))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.SpatialDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.OTHER, MySqlParser.GEOMETRYCOLLECTION), + new DataTypeEntry(Types.OTHER, MySqlParser.GEOMCOLLECTION), + new DataTypeEntry(Types.OTHER, MySqlParser.LINESTRING), + new DataTypeEntry(Types.OTHER, MySqlParser.MULTILINESTRING), + new DataTypeEntry(Types.OTHER, MySqlParser.MULTIPOINT), + new DataTypeEntry(Types.OTHER, MySqlParser.MULTIPOLYGON), + new DataTypeEntry(Types.OTHER, MySqlParser.POINT), + new DataTypeEntry(Types.OTHER, MySqlParser.POLYGON), + new DataTypeEntry(Types.OTHER, MySqlParser.JSON), + new DataTypeEntry(Types.OTHER, MySqlParser.GEOMETRY))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.LongVarbinaryDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.BLOB, MySqlParser.LONG) + .setSuffixTokens(MySqlParser.VARBINARY))); + dataTypeResolverBuilder.registerDataTypes( + MySqlParser.LongVarcharDataTypeContext.class.getCanonicalName(), + Arrays.asList( + new DataTypeEntry(Types.VARCHAR, MySqlParser.LONG) + .setSuffixTokens(MySqlParser.VARCHAR))); + + return dataTypeResolverBuilder.build(); + } + + /** + * Provides a map of default character sets by database/schema name. + * + * @return map of default character sets. + */ + public ConcurrentMap<String, String> charsetNameForDatabase() { + return charsetNameForDatabase; + } + + /** + * Parse a name from {@link MySqlParser.UidContext}. + * + * @param uidContext uid context + * @return name without quotes. + */ + public String parseName(MySqlParser.UidContext uidContext) { + return withoutQuotes(uidContext); + } + + /** + * Parse qualified table identification from {@link MySqlParser.FullIdContext}. {@link + * MySqlAntlrDdlParser#currentSchema()} will be used if definition of schema name is not part of + * the context. + * + * @param fullIdContext full id context. + * @return qualified {@link TableId}. + */ + public TableId parseQualifiedTableId(MySqlParser.FullIdContext fullIdContext) { + final char[] fullTableName = fullIdContext.getText().toCharArray(); + StringBuilder component = new StringBuilder(); + String dbName = null; + String tableName = null; + final char EMPTY = '\0'; + char lastQuote = EMPTY; + for (int i = 0; i < fullTableName.length; i++) { + char c = fullTableName[i]; + if (isQuote(c)) { + // Opening quote + if (lastQuote == EMPTY) { + lastQuote = c; + } + // Closing quote + else if (lastQuote == c) { + // escape of quote by doubling + if (i < fullTableName.length - 1 && fullTableName[i + 1] == c) { + component.append(c); + i++; + } else { + lastQuote = EMPTY; + } + } + // Quote that is part of name + else { + component.append(c); + } + } + // dot that is not in quotes, so name separator + else if (c == '.' && lastQuote == EMPTY) { + dbName = component.toString(); + component = new StringBuilder(); + } + // Any char is part of name including quoted dot + else { + component.append(c); + } + } + tableName = component.toString(); + return resolveTableId(dbName != null ? dbName : currentSchema(), tableName); + } + + /** + * Parse column names for primary index from {@link MySqlParser.IndexColumnNamesContext}. This + * method will updates column to be not optional and set primary key column names to table. + * + * <p>SeaTunnel change: Debezium also calls this method to promote the first unique key of a + * table without a primary key ({@code UNIQUE KEY}, {@code ALTER TABLE ... ADD UNIQUE KEY}, + * {@code CREATE UNIQUE INDEX}). Unlike a primary key, a unique key can hold NULLs, so its + * columns keep their declared nullability; marking them NOT NULL would make the value + * converters emit the type default instead of NULL. Only a real primary key marks its columns + * NOT NULL, as MySQL does. + * + * @param indexColumnNamesContext primary key index column names context. + * @param tableEditor editor for table where primary key index is parsed. + */ + public void parsePrimaryIndexColumnNames( + MySqlParser.IndexColumnNamesContext indexColumnNamesContext, TableEditor tableEditor) { + boolean realPrimaryKey = + indexColumnNamesContext.getParent() + instanceof MySqlParser.PrimaryKeyTableConstraintContext + || indexColumnNamesContext.getParent() + instanceof MySqlParser.AlterByAddPrimaryKeyContext; + List<String> pkColumnNames = + indexColumnNamesContext.indexColumnName().stream() + .map( + indexColumnNameContext -> { + // MySQL does not allow a primary key to have nullable columns, + // so let's make sure we model that correctly ... + String columnName; + if (indexColumnNameContext.uid() != null) { + columnName = parseName(indexColumnNameContext.uid()); + } else { + columnName = + withoutQuotes( + indexColumnNameContext + .STRING_LITERAL() + .getText()); + } + Column column = tableEditor.columnWithName(columnName); + if (realPrimaryKey && column != null && column.isOptional()) { + final ColumnEditor ce = column.edit().optional(false); + if (ce.hasDefaultValue() + && !ce.defaultValueExpression().isPresent()) { + ce.unsetDefaultValueExpression(); + } + tableEditor.addColumn(ce.create()); + } + return column != null ? column.name() : columnName; + }) + .collect(Collectors.toList()); + + tableEditor.setPrimaryKeyNames(pkColumnNames); + } + + /** + * Get the name of the character set for the current database, via the "character_set_database" + * system property. + * + * @return the name of the character set for the current database, or null if not known ... + */ + public String currentDatabaseCharset() { + String charsetName = + systemVariables.getVariable(MySqlSystemVariables.CHARSET_NAME_DATABASE); + if (charsetName == null || "DEFAULT".equalsIgnoreCase(charsetName)) { + charsetName = systemVariables.getVariable(MySqlSystemVariables.CHARSET_NAME_SERVER); + } + return charsetName; + } + + /** + * Get the name of the character set for the give table name. + * + * @return the name of the character set for the given table, or null if not known ... + */ + public String charsetForTable(TableId tableId) { + final String defaultDatabaseCharset = + tableId.catalog() != null ? charsetNameForDatabase().get(tableId.catalog()) : null; + return defaultDatabaseCharset != null ? defaultDatabaseCharset : currentDatabaseCharset(); + } + + /** + * Runs a function if all given object are not null. + * + * @param function function to run; may not be null + * @param nullableObjects object to be tested, if they are null. + */ + public void runIfNotNull(Runnable function, Object... nullableObjects) { + for (Object nullableObject : nullableObjects) { + if (nullableObject == null) { + return; + } + } + function.run(); + } + + /** + * Extracts the enumeration values properly parsed and escaped. + * + * @param enumValues the raw enumeration values from the parsed column definition + * @return the list of options allowed for the {@code ENUM} or {@code SET}; never null. + */ + public static List<String> extractEnumAndSetOptions(List<String> enumValues) { + return enumValues.stream() + .map(MySqlAntlrDdlParser::withoutQuotes) + .map(MySqlAntlrDdlParser::escapeOption) + .collect(Collectors.toList()); + } + + public static String escapeOption(String option) { + // Replace comma to backslash followed by comma (this escape sequence implies comma is part + // of the option) + // Replace backlash+single-quote to a single-quote. + // Replace double single-quote to a single-quote. + return option.replaceAll(",", "\\\\,").replaceAll("\\\\'", "'").replaceAll("''", "'"); + } + + public MySqlValueConverters getConverters() { + return converters; + } + + public TableFilter getTableFilter() { + return tableFilter; + } + + /** + * Obtains the charset name either form charset if present or from collation. + * + * @param charsetNode + * @param collationNode + * @return character set + */ + public String extractCharset( + CharsetNameContext charsetNode, CollationNameContext collationNode) { + String charsetName = null; + if (charsetNode != null && charsetNode.getText() != null) { + charsetName = withoutQuotes(charsetNode.getText()); + } else if (collationNode != null && collationNode.getText() != null) { + final String collationName = withoutQuotes(collationNode.getText()).toLowerCase(); + for (int index = 0; index < CharsetMapping.MAP_SIZE; index++) { + if (collationName.equals( + CharsetMapping.getStaticCollationNameForCollationIndex(index))) { + charsetName = CharsetMapping.getStaticMysqlCharsetNameForCollationIndex(index); + break; + } + } + } + return charsetName; + } +} diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlSchemaTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlSchemaTest.java index 7c9e0152e4..6729d9a0fd 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlSchemaTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlSchemaTest.java @@ -25,14 +25,22 @@ import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.api.table.catalog.TableSchema; import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceTableConfig; +import org.apache.seatunnel.connectors.cdc.base.utils.CatalogTableUtils; import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfig; import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfigFactory; +import org.apache.kafka.connect.data.Struct; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import io.debezium.config.Configuration; +import io.debezium.connector.mysql.MySqlConnectorConfig; +import io.debezium.connector.mysql.MySqlDatabaseSchema; +import io.debezium.connector.mysql.MySqlOffsetContext; +import io.debezium.connector.mysql.MySqlPartition; import io.debezium.jdbc.JdbcConfiguration; import io.debezium.jdbc.JdbcConnection; import io.debezium.relational.Table; @@ -43,6 +51,7 @@ import lombok.Getter; import java.sql.ResultSet; import java.sql.SQLException; +import java.time.Instant; import java.util.Arrays; import java.util.Collections; import java.util.Iterator; @@ -52,6 +61,7 @@ import static org.mockito.Mockito.when; public class MySqlSchemaTest { private static final String QUOTED_CHARACTER = "`"; + private static final TableId NO_PK = TableId.parse("db1.no_pk"); @Test public void testReadSchemaFallbackDescTable() { @@ -141,6 +151,159 @@ public class MySqlSchemaTest { Assertions.assertEquals("DATETIME", table.columnWithName("ts").typeName()); } + @Test + public void testNullableUniqueKeyColumnStaysOptionalWithoutPrimaryKey() { + Table table = + parseTable( + "CREATE TABLE `no_pk` (\n" + + " `id` int NOT NULL,\n" + + " `code` int DEFAULT NULL,\n" + + " UNIQUE KEY `uk_code` (`code`)\n" + + ")", + column("id", false), + column("code", true)); + + // Debezium promotes the unique key to the primary key and marks `code` NOT NULL + Assertions.assertEquals(Arrays.asList("code"), table.primaryKeyColumnNames()); + Assertions.assertTrue(table.columnWithName("code").isOptional()); + Assertions.assertFalse(table.columnWithName("id").isOptional()); + } + + @Test + public void testNullableColumnOfCompositeUniqueKeyStaysOptional() { + Table table = + parseTable( + "CREATE TABLE `no_pk` (\n" + + " `id` int NOT NULL,\n" + + " `a` int NOT NULL,\n" + + " `b` int DEFAULT NULL,\n" + + " UNIQUE KEY `uk_ab` (`a`, `b`)\n" + + ")", + column("id", false), + column("a", false), + column("b", true)); + + Assertions.assertFalse(table.columnWithName("a").isOptional()); + Assertions.assertTrue(table.columnWithName("b").isOptional()); + } + + @Test + public void testNullableColumnConfiguredAsPrimaryKeyEmitsNull() { + // table-names-config: primaryKeys = ["code"], marks the catalog column NOT NULL + JdbcSourceTableConfig config = new JdbcSourceTableConfig(); + config.setTable("db1.no_pk"); + config.setPrimaryKeys(Collections.singletonList("code")); + CatalogTable catalogTable = + CatalogTableUtils.mergeCatalogTableConfig( + catalogTable(column("id", false), column("code", true)), config); + Table table = + parseTable( + "CREATE TABLE `no_pk` (\n" + + " `id` int NOT NULL,\n" + + " `code` int DEFAULT NULL,\n" + + " UNIQUE KEY `uk_code` (`code`)\n" + + ")", + catalogTable); + + Assertions.assertEquals(Arrays.asList("code"), table.primaryKeyColumnNames()); + Assertions.assertTrue(table.columnWithName("code").isOptional()); + MySqlDatabaseSchema schema = createDatabaseSchema(); + schema.refresh(table); + Struct value = schema.schemaFor(NO_PK).valueFromColumnData(new Object[] {2, null}); + Assertions.assertNull(value.get("code")); + } + + @Test + public void testNullableUniqueKeyColumnStaysOptionalInStreamingDdl() { + MySqlDatabaseSchema schema = createDatabaseSchema(); + parseStreamingDdl( + schema, + "CREATE TABLE t1 (id int NOT NULL, code int DEFAULT NULL, name varchar(10))"); + parseStreamingDdl(schema, "ALTER TABLE t1 ADD UNIQUE KEY uk_code (code)"); + Table altered = schema.tableFor(TableId.parse("db1.t1")); + Assertions.assertEquals(Arrays.asList("code"), altered.primaryKeyColumnNames()); + Assertions.assertTrue(altered.columnWithName("code").isOptional()); + + parseStreamingDdl( + schema, + "CREATE TABLE t2 (id int NOT NULL, code int DEFAULT NULL, name varchar(10))"); + parseStreamingDdl(schema, "CREATE UNIQUE INDEX uk_code ON t2 (code)"); + Table indexed = schema.tableFor(TableId.parse("db1.t2")); + Assertions.assertEquals(Arrays.asList("code"), indexed.primaryKeyColumnNames()); + Assertions.assertTrue(indexed.columnWithName("code").isOptional()); + } + + @Test + public void testRealPrimaryKeyColumnIsNotOptionalInStreamingDdl() { + MySqlDatabaseSchema schema = createDatabaseSchema(); + parseStreamingDdl(schema, "CREATE TABLE t1 (id int, name varchar(10), PRIMARY KEY (id))"); + Assertions.assertFalse( + schema.tableFor(TableId.parse("db1.t1")).columnWithName("id").isOptional()); + + parseStreamingDdl(schema, "CREATE TABLE t2 (id int, name varchar(10))"); + parseStreamingDdl(schema, "ALTER TABLE t2 ADD PRIMARY KEY (id)"); + Assertions.assertFalse( + schema.tableFor(TableId.parse("db1.t2")).columnWithName("id").isOptional()); + } + + private static MySqlDatabaseSchema createDatabaseSchema() { + return MySqlConnectionUtils.createMySqlDatabaseSchema(streamingConfig(), false); + } + + private static void parseStreamingDdl(MySqlDatabaseSchema schema, String ddl) { + MySqlConnectorConfig config = streamingConfig(); + schema.parseStreamingDdl( + new MySqlPartition(config.getLogicalName()), + ddl, + "db1", + MySqlOffsetContext.initial(config), + Instant.now()); + } + + private static MySqlConnectorConfig streamingConfig() { + MySqlSourceConfigFactory factory = new MySqlSourceConfigFactory(); + factory.hostname("localhost"); + factory.username("test"); + factory.password("test"); + return factory.create(0).getDbzConnectorConfig(); + } + + private static Table parseTable(String createTableSql, PhysicalColumn... columns) { + return parseTable(createTableSql, catalogTable(columns)); + } + + private static Table parseTable(String createTableSql, CatalogTable catalogTable) { + MySqlSourceConfigFactory factory = new MySqlSourceConfigFactory(); + factory.hostname("localhost"); + factory.username("test"); + factory.password("test"); + MySqlSchema schema = + new MySqlSchema( + factory.create(0), false, Collections.singletonMap(NO_PK, catalogTable)); + return schema.getTableSchema( + new MockJdbcConnection( + createTableSql, Collections.<DescTableField>emptyIterator()), + NO_PK) + .getTable(); + } + + private static CatalogTable catalogTable(PhysicalColumn... columns) { + return CatalogTable.of( + TableIdentifier.of("test", TablePath.of(NO_PK.catalog(), NO_PK.table())), + TableSchema.builder().columns(Arrays.asList(columns)).build(), + Collections.emptyMap(), + Collections.emptyList(), + null); + } + + private static PhysicalColumn column(String name, boolean nullable) { + return PhysicalColumn.builder() + .name(name) + .dataType(BasicType.INT_TYPE) + .nullable(nullable) + .build(); + } + private static class MockJdbcConnection extends JdbcConnection { private String showCreateTableSQL; private Iterator<DescTableField> fields; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java index f3a470c48b..a3ed2bde9d 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java @@ -358,6 +358,155 @@ public abstract class AbstractMysqlCDCITBase extends TestSuiteBase implements Te }); } + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK}, + disabledReason = "Currently SPARK do not support cdc") + public void testMysqlCdcNullInNullableUniqueKeyWithoutPrimaryKey(TestContainer container) { + // Tables without a primary key whose unique key (single, and composite with one nullable + // column) holds NULLs: the NULLs must reach the sink as NULL, not as the type default 0. + // Controls: a table with a real primary key, and a table whose unique key is NOT NULL. + // uk_added_null gets its unique key in the binlog phase, parsed from the binlog DDL. + inventoryDatabase.setTemplateName("nullable_unique_key_null_value").createAndInitialize(); + + CompletableFuture.supplyAsync( + () -> { + try { + container.executeJob( + "/mysqlcdc_to_mysql_with_nullable_unique_key_null_value.conf"); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + return null; + }); + + // snapshot phase + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNullableUniqueKeyTablesSynced(2)); + + // binlog phase + executeSql( + "ALTER TABLE " + MYSQL_DATABASE + ".uk_added_null ADD UNIQUE KEY uk_code (code)"); + insertNullableUniqueKeyRows(4, 5); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNullableUniqueKeyTablesSynced(3)); + } + + @TestTemplate + @DisabledOnContainer( + value = {}, + type = {EngineType.SPARK, EngineType.FLINK}, + disabledReason = "Currently SPARK and FLINK do not support restore") + public void testMysqlCdcNullInNullableUniqueKeyAfterRestore(TestContainer container) + throws Exception { + // The schema of uk_added_null changes in the binlog phase, so the restored job rebuilds it + // from the table history saved in the savepoint: NULL must still reach the sink as NULL. + inventoryDatabase.setTemplateName("nullable_unique_key_null_value").createAndInitialize(); + String conf = "/mysqlcdc_to_mysql_with_nullable_unique_key_null_value.conf"; + String jobId = String.valueOf(JobIdGenerator.newJobId()); + CompletableFuture<Container.ExecResult> jobFuture = + CompletableFuture.supplyAsync( + () -> { + try { + return container.executeJob(conf, jobId); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNullableUniqueKeyTablesSynced(2)); + executeSql( + "ALTER TABLE " + MYSQL_DATABASE + ".uk_added_null ADD UNIQUE KEY uk_code (code)"); + insertNullableUniqueKeyRows(4, 5); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNullableUniqueKeyTablesSynced(3)); + + Assertions.assertEquals(0, container.savepointJob(jobId).getExitCode()); + await().atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + "SAVEPOINT_DONE", container.getJobStatus(jobId))); + Assertions.assertEquals(0, jobFuture.get().getExitCode()); + + CompletableFuture<Container.ExecResult> restoreFuture = + CompletableFuture.supplyAsync( + () -> { + try { + return container.restoreJob(conf, jobId); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + await().atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> Assertions.assertEquals("RUNNING", container.getJobStatus(jobId))); + insertNullableUniqueKeyRows(6, 7); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> assertNullableUniqueKeyTablesSynced(4)); + + container.stopJob(jobId); + await().atMost(120000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> Assertions.assertEquals("CANCELED", container.getJobStatus(jobId))); + Assertions.assertEquals(0, restoreFuture.get().getExitCode()); + } + + /** Inserts one row with NULL in the key column and one row with a value into each table. */ + private void insertNullableUniqueKeyRows(int nullId, int codedId) { + for (String table : new String[] {"uk_added_null", "uk_null_single", "pk_null_control"}) { + executeSql( + String.format( + "INSERT INTO %s.%s VALUES (%d, NULL, 'binlog-null'), (%d, %d, 'binlog-coded')", + MYSQL_DATABASE, table, nullId, codedId, codedId)); + } + executeSql( + String.format( + "INSERT INTO %s.uk_null_composite VALUES (%d, 3, NULL, 'binlog-null'), (%d, 3, %d, 'binlog-coded')", + MYSQL_DATABASE, nullId, codedId, codedId)); + executeSql( + String.format( + "INSERT INTO %s.uk_notnull_control VALUES (%d, %d, 'binlog-coded'), (%d, %d, 'binlog-coded')", + MYSQL_DATABASE, nullId, nullId, codedId, codedId)); + } + + private void assertNullableUniqueKeyTablesSynced(int expectedNullRows) { + String sinkDatabase = "mysql_cdc_null_sink"; + String[][] tables = { + {"uk_null_single", "id, code, name", "code"}, + {"uk_null_composite", "id, a, b, name", "b"}, + {"pk_null_control", "id, code, name", "code"}, + {"uk_notnull_control", "id, code, name", null}, + {"uk_added_null", "id, code, name", "code"} + }; + for (String[] table : tables) { + String rowQuery = "select " + table[1] + " from %s." + table[0] + " order by id"; + Assertions.assertIterableEquals( + query(String.format(rowQuery, MYSQL_DATABASE)), + query(String.format(rowQuery, sinkDatabase)), + table[0]); + if (table[2] != null) { + // NULL must stay SQL NULL in the sink, not become the type default 0 + String nullCount = + "select count(*) from %s." + table[0] + " where " + table[2] + " is null"; + String zeroCount = + "select count(*) from %s." + table[0] + " where " + table[2] + " = 0"; + Assertions.assertEquals( + expectedNullRows, + ((Number) query(String.format(nullCount, sinkDatabase)).get(0).get(0)) + .intValue(), + table[0] + " NULL rows in sink"); + Assertions.assertEquals( + 0, + ((Number) query(String.format(zeroCount, sinkDatabase)).get(0).get(0)) + .intValue(), + table[0] + " rows with 0 in sink"); + } + } + } + @TestTemplate @DisabledOnContainer( value = {}, diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key_null_value.sql b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key_null_value.sql new file mode 100644 index 0000000000..910f3bbdf3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key_null_value.sql @@ -0,0 +1,135 @@ +-- +-- 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. +-- + +-- Tables for the E2E test of NULL values in the nullable unique key of a table without a primary +-- key (single and composite key), with two controls: a table with a real primary key and a table +-- whose unique key is NOT NULL. uk_added_null gets its unique key in the binlog phase. The sink +-- tables have no unique key, so NULLs can be compared. + +CREATE DATABASE IF NOT EXISTS `mysql_cdc`; +CREATE DATABASE IF NOT EXISTS `mysql_cdc_null_sink`; + +use mysql_cdc; + +DROP TABLE IF EXISTS uk_null_single; +CREATE TABLE uk_null_single +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + UNIQUE KEY `uk_code` (`code`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS uk_null_composite; +CREATE TABLE uk_null_composite +( + `id` int NOT NULL, + `a` int NOT NULL, + `b` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + UNIQUE KEY `uk_ab` (`a`, `b`) +) ENGINE = InnoDB; + +INSERT INTO uk_null_single (id, code, name) +VALUES (1, 1, 'snapshot-coded'), (2, NULL, 'snapshot-null'), (3, NULL, 'snapshot-null'); + +INSERT INTO uk_null_composite (id, a, b, name) +VALUES (1, 1, 1, 'snapshot-coded'), (2, 1, NULL, 'snapshot-null'), (3, 2, NULL, 'snapshot-null'); + +DROP TABLE IF EXISTS pk_null_control; +CREATE TABLE pk_null_control +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`), + UNIQUE KEY `uk_code` (`code`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS uk_notnull_control; +CREATE TABLE uk_notnull_control +( + `id` int NOT NULL, + `code` int NOT NULL, + `name` varchar(32) DEFAULT NULL, + UNIQUE KEY `uk_code` (`code`) +) ENGINE = InnoDB; + +INSERT INTO pk_null_control (id, code, name) +VALUES (1, 1, 'snapshot-coded'), (2, NULL, 'snapshot-null'), (3, NULL, 'snapshot-null'); + +INSERT INTO uk_notnull_control (id, code, name) +VALUES (1, 1, 'snapshot-coded'), (2, 0, 'snapshot-zero'), (3, 3, 'snapshot-coded'); + +DROP TABLE IF EXISTS uk_added_null; +CREATE TABLE uk_added_null +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL +) ENGINE = InnoDB; + +INSERT INTO uk_added_null (id, code, name) +VALUES (1, 1, 'snapshot-coded'), (2, NULL, 'snapshot-null'), (3, NULL, 'snapshot-null'); + +use mysql_cdc_null_sink; + +DROP TABLE IF EXISTS uk_null_single; +CREATE TABLE uk_null_single +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS uk_null_composite; +CREATE TABLE uk_null_composite +( + `id` int NOT NULL, + `a` int NOT NULL, + `b` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS pk_null_control; +CREATE TABLE pk_null_control +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS uk_notnull_control; +CREATE TABLE uk_notnull_control +( + `id` int NOT NULL, + `code` int NOT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS uk_added_null; +CREATE TABLE uk_added_null +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key_null_value.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key_null_value.conf new file mode 100644 index 0000000000..5b6619792f --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key_null_value.conf @@ -0,0 +1,46 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + MySQL-CDC { + server-id = 5730-5735 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.uk_null_single", "mysql_cdc.uk_null_composite", "mysql_cdc.pk_null_control", "mysql_cdc.uk_notnull_control", "mysql_cdc.uk_added_null"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + exactly_once = false + } +} + +sink { + jdbc { + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc_null_sink" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = "mysql_cdc_null_sink" + table = "${table_name}" + primary_keys = ["id"] + } +}
