yuxiqian commented on code in PR #3749:
URL: https://github.com/apache/flink-cdc/pull/3749#discussion_r1906628133
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/TransformDef.java:
##########
@@ -72,16 +72,16 @@ public String getSourceTable() {
return sourceTable;
}
- public Optional<String> getProjection() {
- return Optional.ofNullable(projection);
+ public String getProjection() {
+ return projection;
}
public boolean isValidProjection() {
return !StringUtils.isNullOrWhitespaceOnly(projection);
}
- public Optional<String> getFilter() {
- return Optional.ofNullable(filter);
+ public String getFilter() {
+ return filter;
Review Comment:
CMIIW, This class has no accessibility annotation... so it's an `@Internal`
class and breaking changes should be fine.
Maybe we can also tweak `getDescription` method to return nullable `String`
for consistency?
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java:
##########
@@ -354,6 +355,93 @@ 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 testMultiTransformSchemaColumnsCompatibility(ValuesDataSink.SinkApi
sinkApi) {
+ TransformDef nullProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ null,
+ "age < 18",
+ null,
+ null,
+ null,
+ null);
+
+ TransformDef emptyProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ "",
+ "age < 18",
+ null,
+ null,
+ null,
+ null);
+
+ TransformDef asteriskProjection =
+ new TransformDef(
+ "default_namespace.default_schema.mytable2",
+ "*",
+ "age < 18",
+ null,
+ null,
+ null,
+ null);
+
+ List<TransformDef> transformDefList =
+ Arrays.asList(nullProjection, emptyProjection,
asteriskProjection);
+
+ for (TransformDef transformDef : transformDefList) {
+ assertThatThrownBy(
+ () ->
+ runGenericTransformTest(
+ sinkApi,
+ Arrays.asList(
+ transformDef,
+ new TransformDef(
+
"default_namespace.default_schema.mytable2",
+ // reference part
column
+ "id,UPPER(name) AS
name",
+ "age >= 18",
+ null,
+ null,
+ null,
+ null)),
+ Collections.emptyList()))
+ .rootCause()
+ .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.");
+ }
+
+ for (TransformDef transformDef : transformDefList) {
+ assertThatThrownBy(
+ () ->
Review Comment:
Also add some chaining assertions to make sure we're catching correct
Exception and messages?
--
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]
