You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "qg-lin (via GitHub)" <gi...@apache.org> on 2024/09/26 05:16:40 UTC
[PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent [flink-cdc]
qg-lin opened a new pull request, #3622:
URL: https://github.com/apache/flink-cdc/pull/3622
SchemaOperator receives an AddColumnEvent, the default value will be lost when sent downstream.
JIRA: [https://issues.apache.org/jira/browse/FLINK-36375](url)
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "qg-lin (via GitHub)" <gi...@apache.org>.
qg-lin commented on code in PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#discussion_r1815121775
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java:
##########
@@ -50,6 +50,7 @@ public class ValuesDataSourceHelper {
*/
public enum EventSetId {
SINGLE_SPLIT_SINGLE_TABLE,
+ SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE,
Review Comment:
done
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2458776559
Agree to separate the change of `System.lineSeparator` to another pr.
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2472213007
@qg-lin Would you like to backport this PR for branch `release-3.2`?
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#discussion_r1814417426
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values/src/main/java/org/apache/flink/cdc/connectors/values/source/ValuesDataSourceHelper.java:
##########
@@ -50,6 +50,7 @@ public class ValuesDataSourceHelper {
*/
public enum EventSetId {
SINGLE_SPLIT_SINGLE_TABLE,
+ SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE,
Review Comment:
Maybe `ConfigOption<ValuesDataSourceHelper.EventSetId>`'s description needs to be updated correspondingly
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java:
##########
@@ -169,6 +171,66 @@ void testSingleSplitSingleTable(ValuesDataSink.SinkApi sinkApi) throws Exception
"DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, null, null, ], after=[2, null, null, null, x], op=UPDATE, meta=()}");
}
+ @ParameterizedTest
+ @EnumSource
+ void testSingleSplitSingleTableWithDefaultValue(ValuesDataSink.SinkApi sinkApi)
+ throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Setup value source
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ // Setup value sink
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ sinkConfig.set(ValuesDataSinkOptions.SINK_API, sinkApi);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ // Setup pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ // Check result in ValuesDatabase
+ List<String> results = ValuesDatabase.getResults(TABLE_1);
+ assertThat(results)
+ .contains(
Review Comment:
Will `.containsExactlyInAnyOrder` be more suitable?
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2484846111
I've backported this commit to 3.2 branch
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent [flink-cdc]
Posted by "qg-lin (via GitHub)" <gi...@apache.org>.
qg-lin commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2375891374
@lvyanquan @yuxiqian PTAL
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent [flink-cdc]
Posted by "qg-lin (via GitHub)" <gi...@apache.org>.
qg-lin commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2376090149
> Thanks for @qg-lin's contribution! Seems lenientized `RenameColumnEvent` is also prone to this issue. Would you like to fix this in this PR?
my pleasure
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#discussion_r1815882769
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java:
##########
@@ -169,6 +171,66 @@ void testSingleSplitSingleTable(ValuesDataSink.SinkApi sinkApi) throws Exception
"DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, null, null, ], after=[2, null, null, null, x], op=UPDATE, meta=()}");
}
+ @ParameterizedTest
+ @EnumSource
+ void testSingleSplitSingleTableWithDefaultValue(ValuesDataSink.SinkApi sinkApi)
+ throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Setup value source
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ // Setup value sink
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ sinkConfig.set(ValuesDataSinkOptions.SINK_API, sinkApi);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ // Setup pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ // Check result in ValuesDatabase
+ List<String> results = ValuesDatabase.getResults(TABLE_1);
+ assertThat(results)
+ .containsExactly(
Review Comment:
`ValuesDatabase` uses `HashMap` to store records internally, and does not guarantee orderness IIUC. `containsExactlyInAnyOrder` might be more appropriate.
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerLenientITCase.java:
##########
@@ -169,6 +171,66 @@ void testSingleSplitSingleTable(ValuesDataSink.SinkApi sinkApi) throws Exception
"DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, null, null, null, ], after=[2, null, null, null, x], op=UPDATE, meta=()}");
}
+ @ParameterizedTest
+ @EnumSource
+ void testSingleSplitSingleTableWithDefaultValue(ValuesDataSink.SinkApi sinkApi)
+ throws Exception {
+ FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+
+ // Setup value source
+ Configuration sourceConfig = new Configuration();
+ sourceConfig.set(
+ ValuesDataSourceOptions.EVENT_SET_ID,
+ ValuesDataSourceHelper.EventSetId.SINGLE_SPLIT_SINGLE_TABLE_WITH_DEFAULT_VALUE);
+ SourceDef sourceDef =
+ new SourceDef(ValuesDataFactory.IDENTIFIER, "Value Source", sourceConfig);
+
+ // Setup value sink
+ Configuration sinkConfig = new Configuration();
+ sinkConfig.set(ValuesDataSinkOptions.MATERIALIZED_IN_MEMORY, true);
+ sinkConfig.set(ValuesDataSinkOptions.SINK_API, sinkApi);
+ SinkDef sinkDef = new SinkDef(ValuesDataFactory.IDENTIFIER, "Value Sink", sinkConfig);
+
+ // Setup pipeline
+ Configuration pipelineConfig = new Configuration();
+ pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+
+ PipelineDef pipelineDef =
+ new PipelineDef(
+ sourceDef,
+ sinkDef,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ pipelineConfig);
+
+ // Execute the pipeline
+ PipelineExecution execution = composer.compose(pipelineDef);
+ execution.execute();
+
+ // Check result in ValuesDatabase
+ List<String> results = ValuesDatabase.getResults(TABLE_1);
+ assertThat(results)
+ .containsExactly(
Review Comment:
`ValuesDatabase` uses `HashMap` to store records internally, and does not guarantee orderliness IIUC. `containsExactlyInAnyOrder` might be more appropriate.
--
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
Re: [PR] [FLINK-36375][cdc-runtime] fix missing default value in AddColumnEvent/RenameColunEvent [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3622:
URL: https://github.com/apache/flink-cdc/pull/3622#issuecomment-2434508943
Also could you please please rebase this PR with latest `master` branch?
--
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