This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 36d212124 [FLINK-40130][connect/paimon] Fix missing primary key type
compatibility check of Paimon sink (#4471)
36d212124 is described below
commit 36d2121246ce1250883e6b79df426a86ae8fa20a
Author: haruki <[email protected]>
AuthorDate: Mon Jul 13 20:03:24 2026 +0800
[FLINK-40130][connect/paimon] Fix missing primary key type compatibility
check of Paimon sink (#4471)
---
.../flink/cdc/common/utils/SchemaMergingUtils.java | 8 ++-
.../sink/v2/bucket/BucketAssignOperator.java | 38 ++++++++++++++-
.../paimon/sink/v2/PaimonSinkITCase.java | 57 +++++++++++++++++++++-
3 files changed, 98 insertions(+), 5 deletions(-)
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
index ad781d19a..456a7ebc2 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
@@ -327,8 +327,12 @@ public class SchemaMergingUtils {
return coercedRow;
}
- @VisibleForTesting
- static boolean isDataTypeCompatible(@Nullable DataType currentType,
DataType upcomingType) {
+ /**
+ * Checks whether the upcoming data type can fit into the current data
type. A missing current
+ * type is treated as incompatible.
+ */
+ public static boolean isDataTypeCompatible(
+ @Nullable DataType currentType, DataType upcomingType) {
// If two types are identical, they're compatible of course.
if (Objects.equals(currentType, upcomingType)) {
return true;
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
index d523feb72..e3c6f095b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
@@ -28,8 +28,10 @@ import org.apache.flink.cdc.common.event.FlushEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.event.TruncateTableEvent;
+import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.utils.Preconditions;
+import org.apache.flink.cdc.common.utils.SchemaMergingUtils;
import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.connectors.paimon.sink.v2.OperatorIDGenerator;
import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonWriterHelper;
@@ -248,6 +250,7 @@ public class BucketAssignOperator extends
AbstractStreamOperatorAdapter<Event>
catalog.getTable(PaimonWriterHelper.identifierFromTableId(tableId)));
MixedSchemaInfo mixedSchemaInfo =
new MixedSchemaInfo(
+ tableId,
new TableSchemaInfo(upstreamSchema, zoneId),
new TableSchemaInfo(physicalSchema, zoneId));
if (!mixedSchemaInfo.isSameColumnsIgnoringCommentAndDefaultValue()) {
@@ -276,6 +279,7 @@ public class BucketAssignOperator extends
AbstractStreamOperatorAdapter<Event>
if (schema.isPresent()) {
MixedSchemaInfo mixedSchemaInfo =
new MixedSchemaInfo(
+ tableId,
new TableSchemaInfo(schema.get(), zoneId),
new TableSchemaInfo(
PaimonWriterHelper.deduceSchemaForPaimonTable(
@@ -366,6 +370,8 @@ public class BucketAssignOperator extends
AbstractStreamOperatorAdapter<Event>
/** MixedSchemaInfo is used to store the mixed schema info of upstream and
paimon table. */
private static class MixedSchemaInfo {
+ private final TableId tableId;
+
private final TableSchemaInfo upstreamSchemaInfo;
private final TableSchemaInfo paimonSchemaInfo;
@@ -373,12 +379,16 @@ public class BucketAssignOperator extends
AbstractStreamOperatorAdapter<Event>
private final boolean sameColumnsIgnoringCommentAndDefaultValue;
public MixedSchemaInfo(
- TableSchemaInfo upstreamSchemaInfo, TableSchemaInfo
paimonSchemaInfo) {
+ TableId tableId,
+ TableSchemaInfo upstreamSchemaInfo,
+ TableSchemaInfo paimonSchemaInfo) {
+ this.tableId = tableId;
this.upstreamSchemaInfo = upstreamSchemaInfo;
this.paimonSchemaInfo = paimonSchemaInfo;
this.sameColumnsIgnoringCommentAndDefaultValue =
PaimonWriterHelper.sameColumnsIgnoreCommentAndDefaultValue(
upstreamSchemaInfo.getSchema(),
paimonSchemaInfo.getSchema());
+ validatePrimaryKeyTypes();
}
public TableSchemaInfo getUpstreamSchemaInfo() {
@@ -392,5 +402,31 @@ public class BucketAssignOperator extends
AbstractStreamOperatorAdapter<Event>
public boolean isSameColumnsIgnoringCommentAndDefaultValue() {
return sameColumnsIgnoringCommentAndDefaultValue;
}
+
+ private void validatePrimaryKeyTypes() {
+ for (String columnName :
paimonSchemaInfo.getSchema().primaryKeys()) {
+ Column upstreamPrimaryKeyColumn =
+
upstreamSchemaInfo.getSchema().getColumn(columnName).orElse(null);
+ Column paimonPrimaryKeyColumn =
+
paimonSchemaInfo.getSchema().getColumn(columnName).get();
+ if (upstreamPrimaryKeyColumn == null) {
+ throw new IllegalStateException(
+ String.format(
+ "The primary key column %s of %s is not
found in upstream schema.",
+ columnName, tableId));
+ }
+ if (!SchemaMergingUtils.isDataTypeCompatible(
+ paimonPrimaryKeyColumn.getType().nullable(),
+ upstreamPrimaryKeyColumn.getType().nullable())) {
+ throw new IllegalStateException(
+ String.format(
+ "The primary key column %s of %s is %s,
which is not compatible with upstream column type %s.",
+ columnName,
+ tableId,
+ paimonPrimaryKeyColumn.getType(),
+ upstreamPrimaryKeyColumn.getType()));
+ }
+ }
+ }
}
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
index 381c87b24..9a9ab6a42 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
@@ -216,7 +216,18 @@ public class PaimonSinkITCase {
.physicalColumn("col2", STRING())
.option("deletion-vectors.enabled",
String.valueOf(enableDeleteVectors))
.build();
- CreateTableEvent createTableEvent = new CreateTableEvent(table1,
schema);
+
+ Schema upstreamSchema = schema;
+ if (schemaChange ==
SchemaChange.COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN) {
+ upstreamSchema =
+ schema.copy(
+ Schema.newBuilder()
+ .physicalColumn("col1",
VARCHAR(64).notNull())
+ .physicalColumn("col2", STRING())
+ .build()
+ .getColumns());
+ }
+ CreateTableEvent createTableEvent = new CreateTableEvent(table1,
upstreamSchema);
testEvents.add(createTableEvent);
PaimonMetadataApplier metadataApplier = new
PaimonMetadataApplier(catalogOptions);
if (schemaChange != null) {
@@ -263,6 +274,18 @@ public class PaimonSinkITCase {
.physicalColumn("col2", VARCHAR(10));
break;
}
+ case INCOMPATIBLE_PRIMARY_KEY_COLUMN:
+ {
+ builder.physicalColumn("col1", INT().notNull())
+ .physicalColumn("col2", STRING());
+ break;
+ }
+ case COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN:
+ {
+ builder.physicalColumn("col1", STRING().notNull())
+ .physicalColumn("col2", STRING());
+ break;
+ }
}
return schema.copy(builder.build().getColumns());
}
@@ -503,7 +526,9 @@ public class PaimonSinkITCase {
ADD_COLUMN,
REMOVE_COLUMN,
REORDER_COLUMN,
- MODIFY_COLUMN;
+ MODIFY_COLUMN,
+ INCOMPATIBLE_PRIMARY_KEY_COLUMN,
+ COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN;
}
@ParameterizedTest
@@ -524,6 +549,34 @@ public class PaimonSinkITCase {
bucketAssignOperator.setSchemaEvolutionClient(schemaEvolutionClient);
bucketAssignOperator.open(new TaskInfoImpl("test_TaskInfo", 1, 0, 1,
0));
+ if (schemaChange == SchemaChange.INCOMPATIBLE_PRIMARY_KEY_COLUMN) {
+ Assertions.assertThatThrownBy(
+ () ->
+ writeAndCommit(
+ bucketAssignOperator,
+ writer,
+ committer,
+ createTestEvents(false, false,
true, schemaChange)
+ .toArray(new Event[0])))
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "The primary key column col1 of test.table1 is INT
NOT NULL, which is not compatible with upstream column type STRING NOT NULL.");
+ return;
+ }
+
+ if (schemaChange ==
SchemaChange.COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN) {
+ writeAndCommit(
+ bucketAssignOperator,
+ writer,
+ committer,
+ createTestEvents(false, false, true,
schemaChange).toArray(new Event[0]));
+ Assertions.assertThat(fetchResults(table1))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.INSERT, "1", "1"),
+ Row.ofKind(RowKind.INSERT, "2", "2"));
+ return;
+ }
+
// 1. receive only DataChangeEvents during one checkpoint
writeAndCommit(
bucketAssignOperator,
