This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 42d2aab3f8 [Fix][Connector-V2][CDC] Do not use a nullable column as
the snapshot split column (#12597)
42d2aab3f8 is described below
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"]
+ }
+}
