You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "VinaySagarGonabavi (via GitHub)" <gi...@apache.org> on 2026/03/04 01:14:50 UTC
Re: [PR] [FLINK-38830][runtime] Handle duplicate AddColumnEvent gracefully to prevent pipeline crashes [flink-cdc]
VinaySagarGonabavi commented on code in PR #4279:
URL: https://github.com/apache/flink-cdc/pull/4279#discussion_r2881291112
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/TransformOperatorWithSchemaEvolveTest.java:
##########
@@ -840,6 +840,245 @@ void testSchemaChangeWithPostWildcard() throws Exception {
.runTests("inserting columns at last");
}
+ /**
+ * This case tests that duplicate AddColumnEvents are handled gracefully by both
+ * PreTransformOperator and PostTransformOperator. When the same column is added twice (e.g.,
+ * from gh-ost online schema migrations), the second event should be filtered out.
+ */
+ @Test
+ void testDuplicateAddColumnEventPreTransform() throws Exception {
+ TableId tableId = TableId.tableId("my_company", "my_branch", "data_changes");
+ TransformWithSchemaEvolveTestCase.of(
+ tableId,
+ "*, id + age as computed",
+ "name <> 'Alice'",
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .primaryKey("id")
+ .build(),
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.STRING())
+ .physicalColumn("age", DataTypes.INT())
+ .physicalColumn("computed", DataTypes.INT())
+ .primaryKey("id")
+ .build())
+ .initializeHarness()
+ .runTests("initializing table")
+ // First AddColumnEvent: add "extras" column
+ .evolveFromSource(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extras", DataTypes.FLOAT()),
+ AddColumnEvent.ColumnPosition.LAST,
+ null))))
+ .expectInPreTransformed(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extras", DataTypes.FLOAT()),
+ AddColumnEvent.ColumnPosition.AFTER,
+ "age"))))
+ .expectInPostTransformed(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extras", DataTypes.FLOAT()),
+ AddColumnEvent.ColumnPosition.AFTER,
+ "age"))))
+ .runTests("adding extras column first time")
+ // Duplicate AddColumnEvent: add "extras" column again (should be filtered)
+ .evolveFromSource(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ Column.physicalColumn("extras", DataTypes.FLOAT()),
+ AddColumnEvent.ColumnPosition.LAST,
+ null))))
+ .expectNothingInPreTransformed()
+ .expectNothingInPostTransformed()
+ .runTests("duplicate add extras column should be filtered")
Review Comment:
In the new commit, both testDuplicateAddColumnEventPreTransform() and testDuplicateAddColumnEventPostTransform() send a DataChangeEvent after the duplicate AddColumnEvent and assert it's processed correctly through the transform pipeline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org