This is an automated email from the ASF dual-hosted git repository.
corgy-w 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 50e37b5e09 [Fix][Connectors-v2] Require explicit primary keys for
Iceberg upsert mode (#10762)
50e37b5e09 is described below
commit 50e37b5e092d5646297062c3d24ba3c70532214d
Author: Ricky Makhija <[email protected]>
AuthorDate: Wed May 13 07:25:03 2026 +0530
[Fix][Connectors-v2] Require explicit primary keys for Iceberg upsert mode
(#10762)
---
.../introduction/concepts/incompatible-changes.md | 21 +++
.../introduction/concepts/incompatible-changes.md | 17 +++
.../iceberg/config/IcebergSinkConfig.java | 15 ++
.../seatunnel/iceberg/sink/IcebergSink.java | 6 -
.../seatunnel/iceberg/utils/SchemaUtils.java | 49 +++++--
.../seatunnel/iceberg/utils/SchemaUtilsTest.java | 161 +++++++++++++++++++++
6 files changed, 253 insertions(+), 16 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 7cdb776dde..93fb3b75eb 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -35,6 +35,27 @@ You need to check this document before you upgrade to
related version.
### Connector Changes
+- **Breaking Change: Iceberg Connector — source table primary key is no longer
silently inherited**
+ - **Affected component**: `seatunnel-connectors-v2/connector-iceberg`
+ - **Description**: `SchemaUtils.toIcebergSchema()` previously fell back to
the CDC source
+ table's primary key when `iceberg.table.primary-keys` was not explicitly
configured. This
+ silently set `identifier-field-ids` on auto-created Iceberg tables,
activating equality-delete
+ semantics and causing silent INSERT data loss in append-only CDC pipelines
+ (see [#10747](https://github.com/apache/seatunnel/issues/10747)). The
fallback has been
+ removed.
+ - **Impact**: Jobs that set `iceberg.table.upsert-mode-enabled=true` without
an explicit
+ `iceberg.table.primary-keys` will now fail at startup with a clear
`IllegalArgumentException`.
+ Jobs that relied on implicit PK inheritance to drive upsert semantics must
now set
+ `iceberg.table.primary-keys` explicitly.
+ - **Migration Guide**:
+ - **Upsert mode jobs**: Add `iceberg.table.primary-keys = "<your key
columns>"` to the Iceberg
+ sink config.
+ - **Append-only CDC jobs**: No action needed — omitting
`iceberg.table.primary-keys` now
+ correctly routes writes through the pure append writer with no equality
deletes.
+ - **Existing Iceberg tables** that already have `identifier-field-ids`
stored in their
+ Glue/Hive metastore schema are not affected at runtime; only newly
auto-created tables change
+ behavior.
+
### Transform Changes
- **[BREAKING]** SQL Transform `PARSEDATETIME`, `TO_DATE`, and `IS_DATE`
functions now only accept whitelisted datetime format patterns. Custom format
patterns that were previously accepted will now fail at runtime. The supported
patterns are:
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 784729cec6..900fdfd126 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -34,6 +34,23 @@
### 连接器变更
+- **破坏性变更：Iceberg 连接器 — 不再自动继承源表主键**
+ - **影响范围**：`seatunnel-connectors-v2/connector-iceberg`
+ - **变更说明**：当未显式配置 `iceberg.table.primary-keys`
时，`SchemaUtils.toIcebergSchema()`
+ 以前会回退使用 CDC 源表的主键。这会静默地将 `identifier-field-ids` 设置到自动创建的 Iceberg
+ 表上，激活等值删除语义，导致 append-only CDC 管道中的 INSERT 数据静默丢失
+ （详见 [#10747](https://github.com/apache/seatunnel/issues/10747)）。该回退行为已被移除。
+ - **影响**：使用 `iceberg.table.upsert-mode-enabled=true` 但未显式配置
+ `iceberg.table.primary-keys` 的任务，启动时将抛出 `IllegalArgumentException` 并快速失败。
+ 依赖隐式 PK 继承来实现 upsert 语义的任务，需要显式设置 `iceberg.table.primary-keys`。
+ - **迁移指南**：
+ - **Upsert 模式任务**：在 Iceberg sink 配置中添加
+ `iceberg.table.primary-keys = "<主键列名>"`。
+ - **Append-only CDC 任务**：无需任何操作 — 不配置 `iceberg.table.primary-keys`
+ 现在会正确使用纯 append writer，不会产生等值删除文件。
+ - **已存在的 Iceberg 表**（Glue/Hive 元数据中已有 `identifier-field-ids`）在运行时不受影响；
+ 只有 sink 新建的表会改变行为。
+
### 转换变更
- **[BREAKING]** SQL Transform 的 `PARSEDATETIME`、`TO_DATE` 和 `IS_DATE`
函数现在只接受白名单中的日期时间格式模式。以前接受的自定义格式模式现在将在运行时失败。支持的模式有：
diff --git
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/config/IcebergSinkConfig.java
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/config/IcebergSinkConfig.java
index 6567208e4e..4fa9806eeb 100644
---
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/config/IcebergSinkConfig.java
+++
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/config/IcebergSinkConfig.java
@@ -81,6 +81,21 @@ public class IcebergSinkConfig extends IcebergCommonConfig {
this.dataSaveMode =
readonlyConfig.get(IcebergSinkOptions.DATA_SAVE_MODE);
this.dataSaveModeSQL =
readonlyConfig.get(IcebergSinkOptions.DATA_SAVE_MODE_CUSTOM_SQL);
this.commitBranch =
readonlyConfig.get(IcebergSinkOptions.TABLES_DEFAULT_COMMIT_BRANCH);
+ if (this.upsertModeEnabled
+ && readonlyConfig
+ .getOptional(IcebergSinkOptions.TABLE_PRIMARY_KEYS)
+ .map(String::trim)
+ .map(String::isEmpty)
+ .orElse(true)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' is enabled but '%s' is not explicitly
configured. "
+ + "Upsert mode requires an explicit
primary-key list. "
+ + "The source table primary key is no
longer inherited "
+ + "automatically (see:
apache/seatunnel#10747).",
+
IcebergSinkOptions.TABLE_UPSERT_MODE_ENABLED_PROP.key(),
+ IcebergSinkOptions.TABLE_PRIMARY_KEYS.key()));
+ }
}
@VisibleForTesting
diff --git
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/IcebergSink.java
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/IcebergSink.java
index 936471bfd4..d956ac2582 100644
---
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/IcebergSink.java
+++
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/IcebergSink.java
@@ -68,12 +68,6 @@ public class IcebergSink
this.readonlyConfig = pluginConfig;
this.config = new IcebergSinkConfig(pluginConfig);
this.catalogTable = catalogTable;
- // Reset primary keys if need
- if (config.getPrimaryKeys().isEmpty()
- &&
Objects.nonNull(this.catalogTable.getTableSchema().getPrimaryKey())) {
- this.config.setPrimaryKeys(
-
this.catalogTable.getTableSchema().getPrimaryKey().getColumnNames());
- }
// reset partition keys if need
if (config.getPartitionKeys().isEmpty()
&& Objects.nonNull(this.catalogTable.getPartitionKeys())) {
diff --git
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtils.java
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtils.java
index 24161c2585..3afc5efeaa 100644
---
a/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtils.java
+++
b/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtils.java
@@ -24,6 +24,7 @@ import
org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTestin
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
import org.apache.seatunnel.api.table.catalog.TablePath;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
@@ -102,8 +103,20 @@ public class SchemaUtils {
Catalog catalog, TablePath tablePath, CatalogTable table,
ReadonlyConfig readonlyConfig)
throws TableAlreadyExistException, DatabaseNotExistException,
CatalogException {
TableSchema tableSchema = table.getTableSchema();
- // Convert to iceberg schema
- Schema schema = toIcebergSchema(tableSchema, readonlyConfig);
+ // Resolve primary keys: explicit config takes precedence; fall back
to the CatalogTable's
+ // own PK for this catalog-creation path. This fallback is
intentionally absent from the
+ // CDC sink path (toIcebergSchema(tableSchema, readonlyConfig)) where
inheriting the source
+ // table PK would silently activate equality-delete semantics
(apache/seatunnel#10747).
+ List<String> pkColumns =
+ readonlyConfig
+ .getOptional(IcebergSinkOptions.TABLE_PRIMARY_KEYS)
+ .map(e -> IcebergSinkConfig.stringToList(e, ","))
+ .orElseGet(
+ () ->
+
Optional.ofNullable(tableSchema.getPrimaryKey())
+
.map(PrimaryKey::getColumnNames)
+
.orElse(Collections.emptyList()));
+ Schema schema = toIcebergSchema(tableSchema, pkColumns);
// Convert sink config
IcebergSinkConfig config = new IcebergSinkConfig(readonlyConfig);
// build auto create table
@@ -161,19 +174,35 @@ public class SchemaUtils {
return result.get();
}
+ /**
+ * Converts a {@link TableSchema} to an Iceberg {@link Schema} for the CDC
sink path.
+ *
+ * <p>Identifier fields are set ONLY when {@code
iceberg.table.primary-keys} is explicitly
+ * configured. The source table's primary key is intentionally NOT
inherited: silently using the
+ * MySQL PK activates {@code BaseEqualityDeltaWriter}, which emits
positional deletes for
+ * repeated keys within the same checkpoint window, causing silent data
loss in append-only CDC
+ * pipelines (apache/seatunnel#10747).
+ *
+ * <p>For explicit catalog table creation where the CatalogTable's own PK
should be respected,
+ * use {@link #autoCreateTable(Catalog, TablePath, CatalogTable,
ReadonlyConfig)} which applies
+ * the CatalogTable PK as a fallback.
+ */
@VisibleForTesting
@NotNull protected static Schema toIcebergSchema(
TableSchema tableSchema, ReadonlyConfig readonlyConfig) {
+ List<String> pkColumns =
+ readonlyConfig
+ .getOptional(IcebergSinkOptions.TABLE_PRIMARY_KEYS)
+ .map(e -> IcebergSinkConfig.stringToList(e, ","))
+ .orElse(Collections.emptyList());
+ return toIcebergSchema(tableSchema, pkColumns);
+ }
+
+ @VisibleForTesting
+ static Schema toIcebergSchema(TableSchema tableSchema, List<String>
primaryKeyColumns) {
Types.StructType structType = SchemaUtils.toIcebergType(tableSchema);
Set<Integer> identifierFieldIds =
-
readonlyConfig.getOptional(IcebergSinkOptions.TABLE_PRIMARY_KEYS)
- .map(e -> IcebergSinkConfig.stringToList(e, ","))
- .orElseGet(
- () ->
-
Optional.ofNullable(tableSchema.getPrimaryKey())
- .map(e -> e.getColumnNames())
-
.orElse(Collections.emptyList()))
- .stream()
+ primaryKeyColumns.stream()
.map(f -> structType.field(f).fieldId())
.collect(Collectors.toSet());
List<Types.NestedField> fields = new ArrayList<>();
diff --git
a/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtilsTest.java
b/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtilsTest.java
index 6962fd942a..030e2fc66d 100644
---
a/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtilsTest.java
+++
b/seatunnel-connectors-v2/connector-iceberg/src/test/java/org/apache/seatunnel/connectors/seatunnel/iceberg/utils/SchemaUtilsTest.java
@@ -19,9 +19,14 @@ package
org.apache.seatunnel.connectors.seatunnel.iceberg.utils;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergSinkConfig;
import
org.apache.seatunnel.connectors.seatunnel.iceberg.config.IcebergSinkOptions;
import org.apache.iceberg.Schema;
@@ -31,8 +36,10 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Optional;
class SchemaUtilsTest {
@@ -106,5 +113,159 @@ class SchemaUtilsTest {
Assertions.assertEquals(fieldNames[column.fieldId() - 1],
column.name());
Assertions.assertEquals(Boolean.FALSE, column.isRequired());
}
+ Assertions.assertTrue(
+ schema.identifierFieldIds().isEmpty(),
+ "identifier-field-ids must be empty when
iceberg.table.primary-keys is not configured");
+ }
+
+ /**
+ * Regression test for the MySQL PK fallback bug.
+ *
+ * <p>Before the fix, {@code SchemaUtils.toIcebergSchema()} fell back to
{@code
+ * tableSchema.getPrimaryKey()} when {@code iceberg.table.primary-keys}
was not set, silently
+ * copying the CDC source table PK into {@code identifier-field-ids}. This
activated {@code
+ * BaseEqualityDeltaWriter}, which emits positional deletes for repeated
keys within the same
+ * checkpoint window, causing silent INSERT data loss in append-only CDC
pipelines.
+ *
+ * <p>After the fix, {@code identifier-field-ids} must remain empty when
the config key is
+ * absent, regardless of whether the source {@link TableSchema} carries a
primary key.
+ */
+ @Test
+ void testToIcebergSchemaDoesNotInheritSourceTablePkWhenNotConfigured() {
+ String[] fieldNames = new String[] {"id", "name", "value"};
+ SeaTunnelDataType<?>[] dataTypes =
+ new SeaTunnelDataType[] {
+ BasicType.LONG_TYPE, BasicType.STRING_TYPE,
BasicType.STRING_TYPE
+ };
+
+ // Build a TableSchema that has a primary key on "id" — simulating a
CDC source table
+ // (e.g. MySQL) whose PK would previously be silently inherited as
identifier-field-ids.
+ List<Column> columns =
+ Arrays.asList(
+ PhysicalColumn.of(
+ "id", BasicType.LONG_TYPE, (Long) null, false,
null, null),
+ PhysicalColumn.of(
+ "name", BasicType.STRING_TYPE, (Long) null,
true, null, null),
+ PhysicalColumn.of(
+ "value", BasicType.STRING_TYPE, (Long) null,
true, null, null));
+ PrimaryKey sourcePk = PrimaryKey.of("pk_id",
Collections.singletonList("id"));
+ TableSchema tableSchemaWithPk =
+ TableSchema.builder()
+ .columns(columns)
+ .primaryKey(sourcePk)
+ .constraintKey(Collections.emptyList())
+ .build();
+
+ // No iceberg.table.primary-keys in config — append-only CDC pipeline
scenario
+ ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(new
HashMap<>());
+
+ Schema schema = SchemaUtils.toIcebergSchema(tableSchemaWithPk,
readonlyConfig);
+
+ Assertions.assertNotNull(schema);
+ Assertions.assertEquals(fieldNames.length, schema.columns().size());
+ // identifier-field-ids must be empty — source PK must NOT be inherited
+ Assertions.assertTrue(
+ schema.identifierFieldIds().isEmpty(),
+ "identifier-field-ids must be empty when
iceberg.table.primary-keys is not "
+ + "configured, even if the source TableSchema has a
primary key. "
+ + "Inheriting the source PK silently activates
BaseEqualityDeltaWriter "
+ + "and causes positional deletes in append-only CDC
pipelines.");
+ }
+
+ /**
+ * Guard against the regression introduced by the SchemaUtils PK-fallback
fix: when {@code
+ * iceberg.table.upsert-mode-enabled=true} is set without an explicit
{@code
+ * iceberg.table.primary-keys}, the sink must fail fast with a clear error
rather than silently
+ * creating a delta writer with an empty identifier-field set (which would
produce broken upsert
+ * semantics).
+ */
+ @Test
+ void testIcebergSinkConfigThrowsWhenUpsertModeEnabledWithoutPrimaryKeys() {
+ ReadonlyConfig config =
+ ReadonlyConfig.fromMap(
+ new HashMap<String, Object>() {
+ {
+
put(IcebergSinkOptions.TABLE_UPSERT_MODE_ENABLED_PROP.key(), true);
+ // iceberg.table.primary-keys deliberately
absent
+ }
+ });
+ IllegalArgumentException ex =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new IcebergSinkConfig(config),
+ "IcebergSinkConfig must reject
upsert-mode-enabled=true without an "
+ + "explicit iceberg.table.primary-keys
configuration");
+ Assertions.assertTrue(
+
ex.getMessage().contains(IcebergSinkOptions.TABLE_PRIMARY_KEYS.key()),
+ "Error message should name the missing config key");
+ }
+
+ /**
+ * Regression test for the catalog-path CatalogTable PK fallback.
+ *
+ * <p>{@code SchemaUtils.autoCreateTable(Catalog, TablePath, CatalogTable,
ReadonlyConfig)}
+ * overload 1 falls back to the CatalogTable's own primary key when {@code
+ * iceberg.table.primary-keys} is not in config. This is intentionally
different from the CDC
+ * sink path ({@code toIcebergSchema(TableSchema, ReadonlyConfig)}) which
must NOT inherit the
+ * source PK (see apache/seatunnel#10747).
+ *
+ * <p>This test exercises the schema-building half of that overload
directly via the
+ * package-visible {@code toIcebergSchema(TableSchema, List)} overload to
verify that the PK
+ * fallback correctly sets identifier-field-ids.
+ */
+ @Test
+ void testAutoCreateTableCatalogPathPreservesCatalogTablePk() {
+ List<Column> columns =
+ Arrays.asList(
+ PhysicalColumn.of(
+ "id", BasicType.LONG_TYPE, (Long) null, false,
null, null),
+ PhysicalColumn.of(
+ "name", BasicType.STRING_TYPE, (Long) null,
true, null, null),
+ PhysicalColumn.of(
+ "value", BasicType.STRING_TYPE, (Long) null,
true, null, null));
+ PrimaryKey catalogPk = PrimaryKey.of("catalog_pk",
Collections.singletonList("id"));
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .columns(columns)
+ .primaryKey(catalogPk)
+ .constraintKey(Collections.emptyList())
+ .build();
+
+ // No TABLE_PRIMARY_KEYS in config — autoCreateTable overload 1 falls
back to CatalogTable
+ // PK.
+ ReadonlyConfig configWithoutPkKey = ReadonlyConfig.fromMap(new
HashMap<>());
+
+ // Replicate the catalog-path PK resolution logic from autoCreateTable
overload 1.
+ List<String> pkColumns =
+ configWithoutPkKey
+ .getOptional(IcebergSinkOptions.TABLE_PRIMARY_KEYS)
+ .map(e -> IcebergSinkConfig.stringToList(e, ","))
+ .orElseGet(
+ () ->
+
Optional.ofNullable(tableSchema.getPrimaryKey())
+
.map(PrimaryKey::getColumnNames)
+
.orElse(Collections.emptyList()));
+
+ // Call the package-private core method directly (same call
autoCreateTable uses).
+ Schema schema = SchemaUtils.toIcebergSchema(tableSchema, pkColumns);
+
+ Assertions.assertFalse(
+ schema.identifierFieldIds().isEmpty(),
+ "autoCreateTable catalog path must preserve CatalogTable PK
into "
+ + "identifier-field-ids when
iceberg.table.primary-keys is absent");
+ Assertions.assertEquals(
+ 1,
+ schema.identifierFieldIds().size(),
+ "Only the single CatalogTable PK column (id) should be an
identifier field");
+ Types.NestedField idField = schema.findField("id");
+ Assertions.assertNotNull(idField, "Field 'id' must exist in schema");
+ Assertions.assertTrue(
+ schema.identifierFieldIds().contains(idField.fieldId()),
+ "id field must be in identifier-field-ids");
+ Assertions.assertTrue(idField.isRequired(), "PK field 'id' must be
marked required");
+
+ // Non-PK fields must remain optional.
+ Assertions.assertFalse(schema.findField("name").isRequired());
+ Assertions.assertFalse(schema.findField("value").isRequired());
}
}
