Skip to content

Commit 36d2121

Browse files
haruki-830haruki-830
authored
[FLINK-40130][connect/paimon] Fix missing primary key type compatibility check of Paimon sink (#4471)
1 parent 4371a4e commit 36d2121

3 files changed

Lines changed: 98 additions & 5 deletions

File tree

‎flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -327,8 +327,12 @@ public static Object[] coerceRow(
327327
return coercedRow;
328328
}
329329

330-
@VisibleForTesting
331-
static boolean isDataTypeCompatible(@Nullable DataType currentType, DataType upcomingType) {
330+
/**
331+
* Checks whether the upcoming data type can fit into the current data type. A missing current
332+
* type is treated as incompatible.
333+
*/
334+
public static boolean isDataTypeCompatible(
335+
@Nullable DataType currentType, DataType upcomingType) {
332336
// If two types are identical, they're compatible of course.
333337
if (Objects.equals(currentType, upcomingType)) {
334338
return true;

‎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‎

Lines changed: 37 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,10 @@
2828
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
2929
import org.apache.flink.cdc.common.event.TableId;
3030
import org.apache.flink.cdc.common.event.TruncateTableEvent;
31+
import org.apache.flink.cdc.common.schema.Column;
3132
import org.apache.flink.cdc.common.schema.Schema;
3233
import org.apache.flink.cdc.common.utils.Preconditions;
34+
import org.apache.flink.cdc.common.utils.SchemaMergingUtils;
3335
import org.apache.flink.cdc.common.utils.SchemaUtils;
3436
import org.apache.flink.cdc.connectors.paimon.sink.v2.OperatorIDGenerator;
3537
import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonWriterHelper;
@@ -248,6 +250,7 @@ public SchemaChangeEvent convertSchemaChangeEvent(SchemaChangeEvent schemaChange
248250
catalog.getTable(PaimonWriterHelper.identifierFromTableId(tableId)));
249251
MixedSchemaInfo mixedSchemaInfo =
250252
new MixedSchemaInfo(
253+
tableId,
251254
new TableSchemaInfo(upstreamSchema, zoneId),
252255
new TableSchemaInfo(physicalSchema, zoneId));
253256
if (!mixedSchemaInfo.isSameColumnsIgnoringCommentAndDefaultValue()) {
@@ -276,6 +279,7 @@ public DataChangeEvent convertDataChangeEvent(DataChangeEvent dataChangeEvent)
276279
if (schema.isPresent()) {
277280
MixedSchemaInfo mixedSchemaInfo =
278281
new MixedSchemaInfo(
282+
tableId,
279283
new TableSchemaInfo(schema.get(), zoneId),
280284
new TableSchemaInfo(
281285
PaimonWriterHelper.deduceSchemaForPaimonTable(
@@ -366,19 +370,25 @@ public DataChangeEvent convertDataChangeEvent(DataChangeEvent dataChangeEvent)
366370

367371
/** MixedSchemaInfo is used to store the mixed schema info of upstream and paimon table. */
368372
private static class MixedSchemaInfo {
373+
private final TableId tableId;
374+
369375
private final TableSchemaInfo upstreamSchemaInfo;
370376

371377
private final TableSchemaInfo paimonSchemaInfo;
372378

373379
private final boolean sameColumnsIgnoringCommentAndDefaultValue;
374380

375381
public MixedSchemaInfo(
376-
TableSchemaInfo upstreamSchemaInfo, TableSchemaInfo paimonSchemaInfo) {
382+
TableId tableId,
383+
TableSchemaInfo upstreamSchemaInfo,
384+
TableSchemaInfo paimonSchemaInfo) {
385+
this.tableId = tableId;
377386
this.upstreamSchemaInfo = upstreamSchemaInfo;
378387
this.paimonSchemaInfo = paimonSchemaInfo;
379388
this.sameColumnsIgnoringCommentAndDefaultValue =
380389
PaimonWriterHelper.sameColumnsIgnoreCommentAndDefaultValue(
381390
upstreamSchemaInfo.getSchema(), paimonSchemaInfo.getSchema());
391+
validatePrimaryKeyTypes();
382392
}
383393

384394
public TableSchemaInfo getUpstreamSchemaInfo() {
@@ -392,5 +402,31 @@ public TableSchemaInfo getPaimonSchemaInfo() {
392402
public boolean isSameColumnsIgnoringCommentAndDefaultValue() {
393403
return sameColumnsIgnoringCommentAndDefaultValue;
394404
}
405+
406+
private void validatePrimaryKeyTypes() {
407+
for (String columnName : paimonSchemaInfo.getSchema().primaryKeys()) {
408+
Column upstreamPrimaryKeyColumn =
409+
upstreamSchemaInfo.getSchema().getColumn(columnName).orElse(null);
410+
Column paimonPrimaryKeyColumn =
411+
paimonSchemaInfo.getSchema().getColumn(columnName).get();
412+
if (upstreamPrimaryKeyColumn == null) {
413+
throw new IllegalStateException(
414+
String.format(
415+
"The primary key column %s of %s is not found in upstream schema.",
416+
columnName, tableId));
417+
}
418+
if (!SchemaMergingUtils.isDataTypeCompatible(
419+
paimonPrimaryKeyColumn.getType().nullable(),
420+
upstreamPrimaryKeyColumn.getType().nullable())) {
421+
throw new IllegalStateException(
422+
String.format(
423+
"The primary key column %s of %s is %s, which is not compatible with upstream column type %s.",
424+
columnName,
425+
tableId,
426+
paimonPrimaryKeyColumn.getType(),
427+
upstreamPrimaryKeyColumn.getType()));
428+
}
429+
}
430+
}
395431
}
396432
}

‎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‎

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,18 @@ private List<Event> createTestEvents(
216216
.physicalColumn("col2", STRING())
217217
.option("deletion-vectors.enabled", String.valueOf(enableDeleteVectors))
218218
.build();
219-
CreateTableEvent createTableEvent = new CreateTableEvent(table1, schema);
219+
220+
Schema upstreamSchema = schema;
221+
if (schemaChange == SchemaChange.COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN) {
222+
upstreamSchema =
223+
schema.copy(
224+
Schema.newBuilder()
225+
.physicalColumn("col1", VARCHAR(64).notNull())
226+
.physicalColumn("col2", STRING())
227+
.build()
228+
.getColumns());
229+
}
230+
CreateTableEvent createTableEvent = new CreateTableEvent(table1, upstreamSchema);
220231
testEvents.add(createTableEvent);
221232
PaimonMetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions);
222233
if (schemaChange != null) {
@@ -263,6 +274,18 @@ private Schema generateRandomSchema(Schema schema, SchemaChange schemaChange) {
263274
.physicalColumn("col2", VARCHAR(10));
264275
break;
265276
}
277+
case INCOMPATIBLE_PRIMARY_KEY_COLUMN:
278+
{
279+
builder.physicalColumn("col1", INT().notNull())
280+
.physicalColumn("col2", STRING());
281+
break;
282+
}
283+
case COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN:
284+
{
285+
builder.physicalColumn("col1", STRING().notNull())
286+
.physicalColumn("col2", STRING());
287+
break;
288+
}
266289
}
267290
return schema.copy(builder.build().getColumns());
268291
}
@@ -503,7 +526,9 @@ enum SchemaChange {
503526
ADD_COLUMN,
504527
REMOVE_COLUMN,
505528
REORDER_COLUMN,
506-
MODIFY_COLUMN;
529+
MODIFY_COLUMN,
530+
INCOMPATIBLE_PRIMARY_KEY_COLUMN,
531+
COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN;
507532
}
508533

509534
@ParameterizedTest
@@ -524,6 +549,34 @@ void testSinkWithSchemaChangeForExistedTable(SchemaChange schemaChange) throws E
524549
bucketAssignOperator.setSchemaEvolutionClient(schemaEvolutionClient);
525550
bucketAssignOperator.open(new TaskInfoImpl("test_TaskInfo", 1, 0, 1, 0));
526551

552+
if (schemaChange == SchemaChange.INCOMPATIBLE_PRIMARY_KEY_COLUMN) {
553+
Assertions.assertThatThrownBy(
554+
() ->
555+
writeAndCommit(
556+
bucketAssignOperator,
557+
writer,
558+
committer,
559+
createTestEvents(false, false, true, schemaChange)
560+
.toArray(new Event[0])))
561+
.isExactlyInstanceOf(IllegalStateException.class)
562+
.hasMessage(
563+
"The primary key column col1 of test.table1 is INT NOT NULL, which is not compatible with upstream column type STRING NOT NULL.");
564+
return;
565+
}
566+
567+
if (schemaChange == SchemaChange.COMPATIBLE_NOT_NULL_PRIMARY_KEY_COLUMN) {
568+
writeAndCommit(
569+
bucketAssignOperator,
570+
writer,
571+
committer,
572+
createTestEvents(false, false, true, schemaChange).toArray(new Event[0]));
573+
Assertions.assertThat(fetchResults(table1))
574+
.containsExactlyInAnyOrder(
575+
Row.ofKind(RowKind.INSERT, "1", "1"),
576+
Row.ofKind(RowKind.INSERT, "2", "2"));
577+
return;
578+
}
579+
527580
// 1. receive only DataChangeEvents during one checkpoint
528581
writeAndCommit(
529582
bucketAssignOperator,

0 commit comments

Comments
 (0)