yuxiqian commented on code in PR #3749:
URL: https://github.com/apache/flink-cdc/pull/3749#discussion_r1909897648
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java:
##########
@@ -370,6 +371,149 @@ void
testMultiTransformMissingProjection(ValuesDataSink.SinkApi sinkApi) throws
"DataChangeEvent{tableId=default_namespace.default_schema.mytable2, before=[4,
DERRIDA, 25, student], after=[], op=DELETE, meta=()}"));
}
+ @ParameterizedTest
+ @EnumSource
+ void testMultiTransformSchemaColumnsCompatibilityWithNullProjection(
+ ValuesDataSink.SinkApi sinkApi) {
+ TransformDef nullProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ null,
+ "age < 18",
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ assertThatThrownBy(
+ () ->
+ runGenericTransformTest(
+ sinkApi,
+ Arrays.asList(
+ nullProjection,
+ new TransformDef(
+
"default_namespace.default_schema.mytable2",
+ // reference part
column
+ "id,UPPER(name) AS
name",
+ "age >= 18",
+ null,
+ null,
+ null,
+ null,
+ null)),
+ Collections.emptyList()))
+ .rootCause()
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Unable to merge schema columns={`id` BIGINT,`name`
VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, options=() "
+ + "and columns={`id` BIGINT,`name` STRING},
primaryKeys=id, options=() with different column counts.");
+ }
+
+ @ParameterizedTest
+ @EnumSource
+ void testMultiTransformSchemaColumnsCompatibilityWithEmptyProjection(
+ ValuesDataSink.SinkApi sinkApi) {
+ TransformDef emptyProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ "",
+ "age < 18",
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ assertThatThrownBy(
+ () ->
+ runGenericTransformTest(
+ sinkApi,
+ Arrays.asList(
+ emptyProjection,
+ new TransformDef(
+
"default_namespace.default_schema.mytable2",
+ // reference part
column
+ "id,UPPER(name) AS
name",
+ "age >= 18",
+ null,
+ null,
+ null,
+ null,
+ null)),
+ Collections.emptyList()))
+ .rootCause()
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Unable to merge schema columns={`id` BIGINT,`name`
VARCHAR(255),`age` TINYINT,`description` STRING}, primaryKeys=id, options=() "
+ + "and columns={`id` BIGINT,`name` STRING},
primaryKeys=id, options=() with different column counts.");
+ }
+
+ @ParameterizedTest
+ @EnumSource
+ void
testMultiTransformWithNullEmptyAsteriskProjections(ValuesDataSink.SinkApi
sinkApi)
+ throws Exception {
+ TransformDef nullProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ null,
+ "age < 18",
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ TransformDef emptyProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ "",
+ "age < 18",
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ TransformDef asteriskProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ "*",
+ "age < 18",
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ runGenericTransformTest(
+ sinkApi,
+ Arrays.asList(
+ // nullProjection、emptyProjection、asteriskProjection
has the same meaning
Review Comment:
"Setting projection as `null`, `''`, or `*` should be equivalent"
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
