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-12597-33e9153a27dd9a2cda9b3611b911f207af7b8f15 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 42d2aab3f841821753b415d568cc5884d6f13bbd Author: Pravin Jaywantrao Maske <[email protected]> AuthorDate: Wed Oct 14 12:52:55 2026 +0000 [Fix][Connector-V2][CDC] Do not use a nullable column as the snapshot split column (#12597) --- docs/en/connectors/source/MySQL-CDC.md | 3 +- docs/en/connectors/source/Oracle-CDC.md | 3 +- docs/en/connectors/source/PostgreSQL-CDC.md | 3 +- .../introduction/concepts/incompatible-changes.md | 7 + docs/zh/connectors/source/MySQL-CDC.md | 2 +- docs/zh/connectors/source/Oracle-CDC.md | 2 +- docs/zh/connectors/source/PostgreSQL-CDC.md | 2 +- .../introduction/concepts/incompatible-changes.md | 7 + .../cdc/base/dialect/JdbcDataSourceDialect.java | 52 +++++ .../splitter/AbstractJdbcSourceChunkSplitter.java | 36 ++- .../jdbc/source/JdbcSourceChunkSplitterTest.java | 258 +++++++++++++++++++++ .../cdc/mysql/AbstractMysqlCDCITBase.java | 28 +++ .../src/test/resources/ddl/nullable_unique_key.sql | 75 ++++++ ...mysqlcdc_to_mysql_with_nullable_unique_key.conf | 48 ++++ 14 files changed, 517 insertions(+), 9 deletions(-) diff --git a/docs/en/connectors/source/MySQL-CDC.md b/docs/en/connectors/source/MySQL-CDC.md index 0ecc20852d..625ef48d23 100644 --- a/docs/en/connectors/source/MySQL-CDC.md +++ b/docs/en/connectors/source/MySQL-CDC.md @@ -484,7 +484,8 @@ Pick the path that matches what the source table guarantees: row identity. Without a usable key, the connector cannot apply UPDATE/DELETE events safely. - **Unique non-primary column is available**: declare it via `table-names-config.primaryKeys` and set `exactly_once = true` so the snapshot and binlog phases both use the configured key for - consistent row identity. + consistent row identity. The column must be NOT NULL in the database: a nullable column is not + used to split the snapshot, and with `exactly_once = true` the job fails at startup. ```hocon env { diff --git a/docs/en/connectors/source/Oracle-CDC.md b/docs/en/connectors/source/Oracle-CDC.md index 7cbe579abc..11786b13af 100644 --- a/docs/en/connectors/source/Oracle-CDC.md +++ b/docs/en/connectors/source/Oracle-CDC.md @@ -429,7 +429,8 @@ Pick the path that matches what the source table guarantees: row identity. Without a usable key, the connector cannot apply UPDATE/DELETE events safely. - **Unique non-primary column is available**: declare it via `table-names-config.primaryKeys` and set `exactly_once = true` so the snapshot and redo-log phases both use the configured key for - consistent row identity. + consistent row identity. The column must be NOT NULL in the database: a nullable column is not + used to split the snapshot, and with `exactly_once = true` the job fails at startup. ```hocon source { diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md b/docs/en/connectors/source/PostgreSQL-CDC.md index 1cd408eaec..caf02983f1 100644 --- a/docs/en/connectors/source/PostgreSQL-CDC.md +++ b/docs/en/connectors/source/PostgreSQL-CDC.md @@ -299,7 +299,8 @@ Pick the path that matches what the source table guarantees: row identity. Without a usable key, the connector cannot apply UPDATE/DELETE events safely. - **Unique non-primary column is available**: declare it via `table-names-config.primaryKeys` and set `exactly_once = true` so the snapshot and WAL phases both use the configured key for - consistent row identity. + consistent row identity. The column must be NOT NULL in the database: a nullable column is not + used to split the snapshot, and with `exactly_once = true` the job fails at startup. ```hocon source { diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 42c3cc4a2f..09d4fa1e45 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -5,6 +5,13 @@ You need to check this document before you upgrade to related version. ## dev +### CDC: nullable columns are no longer used as the snapshot split column + +- **Behavior change: JDBC-based CDC sources (MySQL, PostgreSQL, Oracle, SQL Server, DB2) no longer split the snapshot on a nullable column** + - **Description**: For a table without a primary key, the snapshot split column was taken from a unique key even when that column was nullable. Rows with NULL in it matched no snapshot chunk and were silently skipped. A nullable column is now skipped, whether it comes from a unique key, from `snapshotSplitColumn` or from a primary key configured in `table-names-config.primaryKeys`; if no non-nullable key column remains, the table is read as a single split. Nullability is read from the [...] + - **Impact**: Such tables are now read completely, but with one snapshot split instead of several. With `exactly_once = true`, a table whose only key is a nullable unique key, or a nullable column configured in `table-names-config.primaryKeys`, now fails at startup with `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...` instead of running and losing rows. This also applies to an existing job whose configured `primaryKeys` column is nullable in [...] + - **Migration Guide**: Declare a NOT NULL unique column, or configure `table-names-config.primaryKeys` / `snapshotSplitColumn` with a non-nullable column, to keep parallel snapshot reading. For `exactly_once = true`, add a primary key or a non-nullable unique key to the table, or make the configured `primaryKeys` column NOT NULL in the database. + ### Runtime Requirements - **Breaking Change: Minimum Java runtime raised from Java 8 to Java 11** diff --git a/docs/zh/connectors/source/MySQL-CDC.md b/docs/zh/connectors/source/MySQL-CDC.md index 9621c75282..f33ebebde1 100644 --- a/docs/zh/connectors/source/MySQL-CDC.md +++ b/docs/zh/connectors/source/MySQL-CDC.md @@ -477,7 +477,7 @@ sink { 根据源表能够提供的保证来选择合适的路径： - **仅追加（append-only）场景**：源表不会产生 UPDATE/DELETE 事件，保持 `exactly_once = false` 且不声明主键，源端会退回到尽力而为的行标识。在没有可用主键的情况下，connector 无法安全地应用 UPDATE/DELETE 事件。 -- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 binlog 阶段都使用同一配置主键作为稳定的行标识。 +- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 binlog 阶段都使用同一配置主键作为稳定的行标识。该列在数据库中必须为 NOT NULL：可空列不会用于切分快照，开启 `exactly_once = true` 时作业会在启动时失败。 ```hocon env { diff --git a/docs/zh/connectors/source/Oracle-CDC.md b/docs/zh/connectors/source/Oracle-CDC.md index 37396fa0f5..bfe2acf683 100644 --- a/docs/zh/connectors/source/Oracle-CDC.md +++ b/docs/zh/connectors/source/Oracle-CDC.md @@ -424,7 +424,7 @@ source { 根据源表能够提供的保证来选择合适的路径： - **仅追加（append-only）场景**：源表不会产生 UPDATE/DELETE 事件，保持 `exactly_once = false` 且不声明主键，源端会退回到尽力而为的行标识。在没有可用主键的情况下，connector 无法安全地应用 UPDATE/DELETE 事件。 -- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 redo log 阶段都使用同一配置主键作为稳定的行标识。 +- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 redo log 阶段都使用同一配置主键作为稳定的行标识。该列在数据库中必须为 NOT NULL：可空列不会用于切分快照，开启 `exactly_once = true` 时作业会在启动时失败。 ```hocon source { diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md b/docs/zh/connectors/source/PostgreSQL-CDC.md index 1b1f6906d4..438b9ac78e 100644 --- a/docs/zh/connectors/source/PostgreSQL-CDC.md +++ b/docs/zh/connectors/source/PostgreSQL-CDC.md @@ -290,7 +290,7 @@ sink { 根据源表能够提供的保证来选择合适的路径： - **仅追加（append-only）场景**：源表不会产生 UPDATE/DELETE 事件，保持 `exactly_once = false` 且不声明主键，源端会退回到尽力而为的行标识。在没有可用主键的情况下，connector 无法安全地应用 UPDATE/DELETE 事件。 -- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 WAL 阶段都使用同一配置主键作为稳定的行标识。 +- **存在唯一非主键列**：通过 `table-names-config.primaryKeys` 显式声明该列，并设置 `exactly_once = true`，让快照阶段与 WAL 阶段都使用同一配置主键作为稳定的行标识。该列在数据库中必须为 NOT NULL：可空列不会用于切分快照，开启 `exactly_once = true` 时作业会在启动时失败。 ```hocon source { diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 454b9848bd..b0c4680c9e 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -4,6 +4,13 @@ ## dev +### CDC：可空列不再用作快照分片列 + +- **行为变更：基于 JDBC 的 CDC 源（MySQL、PostgreSQL、Oracle、SQL Server、DB2）不再使用可空列切分快照** + - **描述**：对于没有主键的表，此前即使唯一键列可空，也会将其作为快照分片列。该列为 NULL 的行不会匹配任何快照分片，会被静默跳过。现在会跳过可空列，无论它来自唯一键、`snapshotSplitColumn`，还是 `table-names-config.primaryKeys` 中配置的主键；如果没有可用的非空键列，则整张表作为单个分片读取。可空性从数据库元数据中读取。 + - **影响**：这类表现在会被完整读取，但快照只有一个分片而不是多个。开启 `exactly_once = true` 时，若表唯一的键是可空唯一键，或 `table-names-config.primaryKeys` 中配置的是可空列，作业会在启动时报错 `Exactly once is enabled, but not found non-nullable primary key or unique key for table ...`，而不是继续运行并丢失数据。对于已有作业，即使配置的 `primaryKeys` 列中实际没有 NULL，只要该列在数据库中可空，也会如此。 + - **迁移指南**：如需保持快照并行读取，请声明一个 NOT NULL 的唯一列，或通过 `table-names-config.primaryKeys` / `snapshotSplitColumn` 配置非空列。对于 `exactly_once = true`，请为表添加主键或非空唯一键，或将配置的 `primaryKeys` 列在数据库中改为 NOT NULL。 + ### 运行环境要求 - **破坏性变更：最低 Java 运行时从 Java 8 提升到 Java 11** diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/JdbcDataSourceDialect.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/JdbcDataSourceDialect.java index 4cc5883662..b384187402 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/JdbcDataSourceDialect.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/dialect/JdbcDataSourceDialect.java @@ -29,7 +29,10 @@ import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase; import org.apache.commons.collections4.CollectionUtils; +import org.slf4j.LoggerFactory; + import io.debezium.jdbc.JdbcConnection; +import io.debezium.relational.Column; import io.debezium.relational.TableId; import io.debezium.relational.history.TableChanges; @@ -123,6 +126,55 @@ public interface JdbcDataSourceDialect extends DataSourceDialect<JdbcSourceConfi .collect(Collectors.toList()); } + /** + * Whether the column can hold NULL, according to the database metadata. + * + * <p>The parsed {@link Column} is not reliable for this: for a MySQL table without a primary + * key, the column of a nullable unique key is reported as not optional. + * + * <p>Called while generating the snapshot splits of a table, for each candidate split column + * (the configured split column, the first primary key column and the first column of each + * unique key). The default implementation runs one {@link DatabaseMetaData#getColumns} query + * per call; a dialect may override it with a cheaper source of the same information. + * + * @param jdbcConnection the JDBC connection + * @param tableId the table the column belongs to + * @param column the column to check + * @return true if the column is nullable, or the metadata does not describe it and the parsed + * column is optional + */ + default boolean isColumnNullable(JdbcConnection jdbcConnection, TableId tableId, Column column) + throws SQLException { + DatabaseMetaData metaData = jdbcConnection.connection().getMetaData(); + String ignoreCaseMatch = null; + try (ResultSet rs = + metaData.getColumns( + tableId.catalog(), tableId.schema(), tableId.table(), column.name())) { + while (rs.next()) { + // table and column names are LIKE patterns, so match them by name + String tableName = rs.getString("TABLE_NAME"); + String columnName = rs.getString("COLUMN_NAME"); + if (tableId.table().equals(tableName) && column.name().equals(columnName)) { + return !"NO".equalsIgnoreCase(rs.getString("IS_NULLABLE")); + } + if (ignoreCaseMatch == null + && tableId.table().equalsIgnoreCase(tableName) + && column.name().equalsIgnoreCase(columnName)) { + ignoreCaseMatch = rs.getString("IS_NULLABLE"); + } + } + } + if (ignoreCaseMatch != null) { + return !"NO".equalsIgnoreCase(ignoreCaseMatch); + } + LoggerFactory.getLogger(JdbcDataSourceDialect.class) + .warn( + "No column metadata found for column {} of table {}, use the parsed column nullability", + column.name(), + tableId); + return column.isOptional(); + } + default List<ConstraintKey> getConstraintKeys(JdbcConnection jdbcConnection, TableId tableId) throws SQLException { DatabaseMetaData metaData = jdbcConnection.connection().getMetaData(); diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java index 6d1667674f..a90f070720 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/splitter/AbstractJdbcSourceChunkSplitter.java @@ -51,6 +51,10 @@ import static org.apache.seatunnel.connectors.cdc.base.utils.ObjectUtils.doubleC @Slf4j public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunkSplitter { + private static final String NULLABLE_SPLIT_COLUMN_HINT = + "To read the snapshot in parallel, declare a NOT NULL unique column or configure a " + + "non-nullable primaryKeys or snapshotSplitColumn in table-names-config."; + private final JdbcSourceConfig sourceConfig; private final JdbcDataSourceDialect dialect; @@ -86,7 +90,7 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk if (sourceConfig.isExactlyOnce()) { throw new UnsupportedOperationException( String.format( - "Exactly once is enabled, but not found primary key or unique key for table %s", + "Exactly once is enabled, but not found non-nullable primary key or unique key for table %s", tableId)); } SnapshotSplit singleSplit = createSnapshotSplit(jdbc, tableId, 0, null, null, null); @@ -489,7 +493,14 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk if (isPrimaryOrUniqueKey.get()) { Column column = table.columnWithName(tableSc); - if (isEvenlySplitColumn(column)) { + if (dialect.isColumnNullable(jdbc, tableId, column)) { + log.warn( + "Config snapshotSplitColumn {} of table {} is nullable, rows with NULL in it " + + "cannot be assigned to any snapshot chunk, ignore it. " + + NULLABLE_SPLIT_COLUMN_HINT, + tableSc, + tableId); + } else if (isEvenlySplitColumn(column)) { return column; } else { log.warn( @@ -507,7 +518,15 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk if (primaryKey.isPresent()) { Column firstColumn = table.columnWithName(primaryKey.get().getColumnNames().get(0)); - if (isEvenlySplitColumn(firstColumn)) { + // A primary key configured in table-names-config may name a nullable column + if (dialect.isColumnNullable(jdbc, tableId, firstColumn)) { + log.warn( + "Skip primary key {} of table {} as snapshot split column: column {} is nullable. " + + NULLABLE_SPLIT_COLUMN_HINT, + primaryKey.get().getColumnNames(), + tableId, + firstColumn.name()); + } else if (isEvenlySplitColumn(firstColumn)) { splitColumn = columnComparable(splitColumn, firstColumn); if (sqlTypePriority(splitColumn) == 1) { return splitColumn; @@ -522,6 +541,17 @@ public abstract class AbstractJdbcSourceChunkSplitter implements JdbcSourceChunk for (ConstraintKey uniqueKey : uniqueKeys) { Column firstColumn = table.columnWithName(uniqueKey.getColumnNames().get(0).getColumnName()); + // A unique key may hold any number of NULLs, and NULL never matches a chunk + // range predicate, so such rows would be silently skipped by the snapshot. + if (dialect.isColumnNullable(jdbc, tableId, firstColumn)) { + log.warn( + "Skip unique key {} of table {} as snapshot split column: column {} is nullable. " + + NULLABLE_SPLIT_COLUMN_HINT, + uniqueKey.getConstraintName(), + tableId, + firstColumn.name()); + continue; + } if (isEvenlySplitColumn(firstColumn)) { splitColumn = columnComparable(splitColumn, firstColumn); if (sqlTypePriority(splitColumn) == 1) { diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/jdbc/source/JdbcSourceChunkSplitterTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/jdbc/source/JdbcSourceChunkSplitterTest.java index 734e98d25f..6e88182987 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/jdbc/source/JdbcSourceChunkSplitterTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/jdbc/source/JdbcSourceChunkSplitterTest.java @@ -41,13 +41,18 @@ import io.debezium.relational.Table; import io.debezium.relational.TableId; import io.debezium.relational.history.TableChanges; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Types; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Iterator; import java.util.List; import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -98,6 +103,169 @@ class JdbcSourceChunkSplitterTest { Assertions.assertEquals("bigint", splitColumn.typeName()); } + @Test + void splitColumnSkipsNullableUniqueKey() throws SQLException { + TestSourceDialectWithNullableUniqueKey dialect = + new TestSourceDialectWithNullableUniqueKey(); + TestJdbcSourceChunkSplitter testJdbcSourceChunkSplitter = + new TestJdbcSourceChunkSplitter(null, dialect); + + // NULLs in a unique key never match a chunk range, so no split column is safe here and + // the table falls back to a single full-scan split. + Assertions.assertNull( + testJdbcSourceChunkSplitter.getSplitColumn(null, dialect, new TableId("", "", ""))); + } + + @Test + void splitColumnPrefersNotNullUniqueKeyOverNullableOne() throws SQLException { + TestSourceDialectWithNullableAndNotNullUniqueKeys dialect = + new TestSourceDialectWithNullableAndNotNullUniqueKeys(); + TestJdbcSourceChunkSplitter testJdbcSourceChunkSplitter = + new TestJdbcSourceChunkSplitter(null, dialect); + + Column splitColumn = + testJdbcSourceChunkSplitter.getSplitColumn(null, dialect, new TableId("", "", "")); + + Assertions.assertEquals("int", splitColumn.name()); + } + + @Test + void splitColumnIgnoresConfiguredNullableColumn() throws SQLException { + JdbcSourceConfig sourceConfig = mock(JdbcSourceConfig.class); + when(sourceConfig.getSplitColumn()) + .thenReturn(Collections.singletonMap(".", "nullable_int")); + TestSourceDialectWithNullableAndNotNullUniqueKeys dialect = + new TestSourceDialectWithNullableAndNotNullUniqueKeys(); + TestJdbcSourceChunkSplitter testJdbcSourceChunkSplitter = + new TestJdbcSourceChunkSplitter(sourceConfig, dialect); + + Column splitColumn = + testJdbcSourceChunkSplitter.getSplitColumn(null, dialect, new TableId("", "", "")); + + Assertions.assertEquals("int", splitColumn.name()); + } + + @Test + void splitColumnSkipsNullableConfiguredPrimaryKey() throws SQLException { + // table-names-config.primaryKeys may name a nullable column + TestSourceDialectWithNullablePrimaryKey dialect = + new TestSourceDialectWithNullablePrimaryKey(); + TestJdbcSourceChunkSplitter testJdbcSourceChunkSplitter = + new TestJdbcSourceChunkSplitter(null, dialect); + + Assertions.assertNull( + testJdbcSourceChunkSplitter.getSplitColumn(null, dialect, new TableId("", "", ""))); + } + + @Test + void splitColumnPrefersNotNullUniqueKeyOverNullablePrimaryKey() throws SQLException { + TestSourceDialectWithNullablePrimaryKeyAndNotNullUniqueKey dialect = + new TestSourceDialectWithNullablePrimaryKeyAndNotNullUniqueKey(); + TestJdbcSourceChunkSplitter testJdbcSourceChunkSplitter = + new TestJdbcSourceChunkSplitter(null, dialect); + + Column splitColumn = + testJdbcSourceChunkSplitter.getSplitColumn(null, dialect, new TableId("", "", "")); + + Assertions.assertEquals("int", splitColumn.name()); + } + + @Test + void isColumnNullableReadsDatabaseMetadata() throws SQLException { + TestSourceDialect dialect = new TestSourceDialect(); + TableId tableId = new TableId("db", null, "no_pk"); + // the parsed column says NOT NULL, the database metadata decides + Column code = intColumn("code", false); + + Assertions.assertTrue( + dialect.isColumnNullableFromMetadata( + jdbcWithColumns(tableId, "code", row("no_pk", "code", "YES")), + tableId, + code)); + Assertions.assertFalse( + dialect.isColumnNullableFromMetadata( + jdbcWithColumns(tableId, "code", row("no_pk", "code", "NO")), + tableId, + intColumn("code", true))); + } + + @Test + void isColumnNullableFallsBackToParsedColumnWithoutMetadataRow() throws SQLException { + TestSourceDialect dialect = new TestSourceDialect(); + TableId tableId = new TableId("db", null, "no_pk"); + + Assertions.assertTrue( + dialect.isColumnNullableFromMetadata( + jdbcWithColumns(tableId, "code"), tableId, intColumn("code", true))); + Assertions.assertFalse( + dialect.isColumnNullableFromMetadata( + jdbcWithColumns(tableId, "code"), tableId, intColumn("code", false))); + } + + @Test + void isColumnNullableIgnoresRowsMatchedOnlyByNamePattern() throws SQLException { + TestSourceDialect dialect = new TestSourceDialect(); + TableId tableId = new TableId("db", null, "no_pk"); + // getColumns takes LIKE patterns: `_` also matches the column of table `noXpk` + JdbcConnection jdbc = jdbcWithColumns(tableId, "code", row("noXpk", "code", "YES")); + + Assertions.assertFalse( + dialect.isColumnNullableFromMetadata(jdbc, tableId, intColumn("code", false))); + } + + @Test + void isColumnNullableMatchesNameIgnoringCaseWithoutExactRow() throws SQLException { + TestSourceDialect dialect = new TestSourceDialect(); + TableId tableId = new TableId("db", null, "No_Pk"); + // for example MySQL with lower_case_table_names=1 returns the name in lower case + JdbcConnection jdbc = jdbcWithColumns(tableId, "Code", row("no_pk", "code", "YES")); + + Assertions.assertTrue( + dialect.isColumnNullableFromMetadata(jdbc, tableId, intColumn("Code", false))); + } + + private static Column intColumn(String name, boolean optional) { + return Column.editor() + .name(name) + .jdbcType(Types.INTEGER) + .type("int") + .optional(optional) + .create(); + } + + private static String[] row(String tableName, String columnName, String isNullable) { + return new String[] {tableName, columnName, isNullable}; + } + + /** A connection whose metadata returns the given TABLE_NAME, COLUMN_NAME, IS_NULLABLE rows. */ + private static JdbcConnection jdbcWithColumns( + TableId tableId, String columnName, String[]... rows) throws SQLException { + Iterator<String[]> iterator = Arrays.asList(rows).iterator(); + AtomicReference<String[]> current = new AtomicReference<>(); + ResultSet resultSet = mock(ResultSet.class); + when(resultSet.next()) + .thenAnswer( + invocation -> { + if (!iterator.hasNext()) { + return false; + } + current.set(iterator.next()); + return true; + }); + when(resultSet.getString("TABLE_NAME")).thenAnswer(invocation -> current.get()[0]); + when(resultSet.getString("COLUMN_NAME")).thenAnswer(invocation -> current.get()[1]); + when(resultSet.getString("IS_NULLABLE")).thenAnswer(invocation -> current.get()[2]); + + DatabaseMetaData metaData = mock(DatabaseMetaData.class); + when(metaData.getColumns(tableId.catalog(), tableId.schema(), tableId.table(), columnName)) + .thenReturn(resultSet); + Connection connection = mock(Connection.class); + when(connection.getMetaData()).thenReturn(metaData); + JdbcConnection jdbc = mock(JdbcConnection.class); + when(jdbc.connection()).thenReturn(connection); + return jdbc; + } + private class TestJdbcSourceChunkSplitter extends AbstractJdbcSourceChunkSplitter { public TestJdbcSourceChunkSplitter( @@ -223,31 +391,43 @@ class JdbcSourceChunkSplitterTest { .name("string_col") .jdbcType(Types.VARCHAR) .type("varchar") + .optional(false) .create(), Column.editor() .name("smallint") .jdbcType(Types.SMALLINT) .type("smallint") + .optional(false) .create(), Column.editor() .name("int") .jdbcType(Types.INTEGER) .type("int") + .optional(false) .create(), Column.editor() .name("decimal") .jdbcType(Types.DECIMAL) .type("decimal") + .optional(false) .create(), Column.editor() .name("tinyint_col") .jdbcType(Types.TINYINT) .type("tinyint") + .optional(false) .create(), Column.editor() .name("bigint_col") .jdbcType(Types.BIGINT) .type("bigint") + .optional(false) + .create(), + Column.editor() + .name("nullable_int") + .jdbcType(Types.INTEGER) + .type("int") + .optional(true) .create()) .create(); return new TableChanges.TableChange(TableChanges.TableChangeType.CREATE, table); @@ -284,6 +464,19 @@ class JdbcSourceChunkSplitterTest { throws SQLException { return new ArrayList<ConstraintKey>(); } + + @Override + public boolean isColumnNullable( + JdbcConnection jdbcConnection, TableId tableId, Column column) { + // no database here, the fixture columns carry the nullability + return column.isOptional(); + } + + /** Runs the default, metadata based implementation of the dialect. */ + boolean isColumnNullableFromMetadata( + JdbcConnection jdbcConnection, TableId tableId, Column column) throws SQLException { + return JdbcDataSourceDialect.super.isColumnNullable(jdbcConnection, tableId, column); + } } private class TestSourceDialectWithUniqueKey extends TestSourceDialect { @@ -347,4 +540,69 @@ class JdbcSourceChunkSplitterTest { return keys; } } + + private class TestSourceDialectWithNullableUniqueKey extends TestSourceDialect { + + @Override + public Optional<PrimaryKey> getPrimaryKey(JdbcConnection jdbcConnection, TableId tableId) + throws SQLException { + return Optional.empty(); + } + + @Override + public List<ConstraintKey> getUniqueKeys(JdbcConnection jdbcConnection, TableId tableId) + throws SQLException { + return Collections.singletonList( + ConstraintKey.of( + ConstraintKey.ConstraintType.UNIQUE_KEY, + "uk_nullable", + Collections.singletonList( + ConstraintKey.ConstraintKeyColumn.of( + "nullable_int", ConstraintKey.ColumnSortType.ASC)))); + } + } + + private class TestSourceDialectWithNullableAndNotNullUniqueKeys + extends TestSourceDialectWithNullableUniqueKey { + + @Override + public List<ConstraintKey> getUniqueKeys(JdbcConnection jdbcConnection, TableId tableId) + throws SQLException { + List<ConstraintKey> keys = + new ArrayList<>(super.getUniqueKeys(jdbcConnection, tableId)); + keys.add( + ConstraintKey.of( + ConstraintKey.ConstraintType.UNIQUE_KEY, + "uk_not_null", + Collections.singletonList( + ConstraintKey.ConstraintKeyColumn.of( + "int", ConstraintKey.ColumnSortType.ASC)))); + return keys; + } + } + + private class TestSourceDialectWithNullablePrimaryKey extends TestSourceDialect { + + @Override + public Optional<PrimaryKey> getPrimaryKey(JdbcConnection jdbcConnection, TableId tableId) + throws SQLException { + return Optional.of(PrimaryKey.of("pkName", Collections.singletonList("nullable_int"))); + } + } + + private class TestSourceDialectWithNullablePrimaryKeyAndNotNullUniqueKey + extends TestSourceDialectWithNullablePrimaryKey { + + @Override + public List<ConstraintKey> getUniqueKeys(JdbcConnection jdbcConnection, TableId tableId) + throws SQLException { + return Collections.singletonList( + ConstraintKey.of( + ConstraintKey.ConstraintType.UNIQUE_KEY, + "uk_not_null", + Collections.singletonList( + ConstraintKey.ConstraintKeyColumn.of( + "int", ConstraintKey.ColumnSortType.ASC)))); + } + } } 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..55fa1d9662 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,34 @@ public abstract class AbstractMysqlCDCITBase extends TestSuiteBase implements Te }); } + @TestTemplate + public void testMysqlCdcSnapshotWithNullableUniqueKey(TestContainer container) { + // A table without a primary key whose only unique key is nullable: rows with NULL in it + // must not be skipped by the snapshot chunks. + inventoryDatabase.setTemplateName("nullable_unique_key").createAndInitialize(); + + CompletableFuture.supplyAsync( + () -> { + try { + container.executeJob("/mysqlcdc_to_mysql_with_nullable_unique_key.conf"); + } catch (Exception e) { + log.error("Commit task exception :" + e.getMessage()); + throw new RuntimeException(e); + } + return null; + }); + + // Compare every column except code: NULL codes are emitted as 0 by the MySQL schema + // parsing, a separate issue (#12599) from the skipped rows. + String rowQuery = "select id, name from " + MYSQL_DATABASE + ".%s order by id"; + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertIterableEquals( + query(String.format(rowQuery, "nullable_uk_src")), + query(String.format(rowQuery, "nullable_uk_sink")))); + } + @TestTemplate @DisabledOnContainer( value = {}, diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key.sql b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key.sql new file mode 100644 index 0000000000..189817f179 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/nullable_unique_key.sql @@ -0,0 +1,75 @@ +-- +-- 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 snapshot split E2E test of a table without a primary key whose only unique key +-- is nullable. Rows with NULL in the unique key must still be read by the snapshot. + +CREATE DATABASE IF NOT EXISTS `mysql_cdc`; + +use mysql_cdc; + +DROP TABLE IF EXISTS nullable_uk_src; +CREATE TABLE nullable_uk_src +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + UNIQUE KEY `uk_code` (`code`) +) ENGINE = InnoDB; + +DROP TABLE IF EXISTS nullable_uk_sink; +CREATE TABLE nullable_uk_sink +( + `id` int NOT NULL, + `code` int DEFAULT NULL, + `name` varchar(32) DEFAULT NULL, + PRIMARY KEY (`id`) +) ENGINE = InnoDB; + +-- 20 rows with a unique code and 10 rows with a NULL code +INSERT INTO nullable_uk_src (id, code, name) +VALUES + (1, 1, 'row-1'), + (2, 2, 'row-2'), + (3, 3, 'row-3'), + (4, 4, 'row-4'), + (5, 5, 'row-5'), + (6, 6, 'row-6'), + (7, 7, 'row-7'), + (8, 8, 'row-8'), + (9, 9, 'row-9'), + (10, 10, 'row-10'), + (11, 11, 'row-11'), + (12, 12, 'row-12'), + (13, 13, 'row-13'), + (14, 14, 'row-14'), + (15, 15, 'row-15'), + (16, 16, 'row-16'), + (17, 17, 'row-17'), + (18, 18, 'row-18'), + (19, 19, 'row-19'), + (20, 20, 'row-20'), + (21, NULL, 'row-21'), + (22, NULL, 'row-22'), + (23, NULL, 'row-23'), + (24, NULL, 'row-24'), + (25, NULL, 'row-25'), + (26, NULL, 'row-26'), + (27, NULL, 'row-27'), + (28, NULL, 'row-28'), + (29, NULL, 'row-29'), + (30, NULL, 'row-30'); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key.conf new file mode 100644 index 0000000000..7bf04f2ee5 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_nullable_unique_key.conf @@ -0,0 +1,48 @@ +# +# 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 = 5720-5725 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.nullable_uk_src"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + exactly_once = false + # small chunks so the snapshot of the 30 row table is split into several chunks + snapshot.split.size = 5 + } +} + +sink { + jdbc { + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = mysql_cdc + table = nullable_uk_sink + primary_keys = ["id"] + } +}
