You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "aiwenmo (via GitHub)" <gi...@apache.org> on 2024/07/06 03:35:54 UTC
[PR] [FLINK-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
aiwenmo opened a new pull request, #3455:
URL: https://github.com/apache/flink-cdc/pull/3455
This closes [FLINK-35774](https://issues.apache.org/jira/browse/FLINK-35774).
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#issuecomment-2276030958
The issue has been fixed by https://github.com/apache/flink-cdc/pull/3285.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708380132
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -332,8 +332,8 @@ void testTransform(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20], after=[2, x, 20], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 10, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 20, ], after=[2, 20, x], op=UPDATE, meta=()}");
Review Comment:
Could you please also add tests to verify if downstream schema (could be accessed by `ValuesDatabase#getTableSchema`) is as expected?
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708376615
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -399,8 +399,8 @@ void testTransformTwice(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 11], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 22], after=[2, x, 22], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 11, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 22, ], after=[2, 22, x], op=UPDATE, meta=()}");
Review Comment:
Is transform semantic changed here? Seems after the schema evolution, upstream schema will be `[col1, newCol3]`, and downstream schema should be inferred as `[col1, newCol3, col12]` instead of `[col1, col12, newCol3]`.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#issuecomment-2211616141
@yuxiqian @lvyanquan 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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#issuecomment-2260910689
I have rebased this branch onto master.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708376615
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -399,8 +399,8 @@ void testTransformTwice(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 11], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 22], after=[2, x, 22], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 11, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 22, ], after=[2, 22, x], op=UPDATE, meta=()}");
Review Comment:
Is transform semantic changed here? Seems after the schema evolution, upstream schema will be `[col1, newCol3]`, and downstream should be inferred as `[col1, newCol3, col12]` instead of `[col1, col12, newCol3]`.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1668456054
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformDataOperator.java:
##########
Review Comment:
~~I'm a bit unsure if `transformProjectionProcessorMap` and `transformFilterProcessorMap` will be flushed when upstream schema change event arrives at `TransformDataOperator`?~~
Got it, it will be refreshed by updating `tableInfo`.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708380132
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -332,8 +332,8 @@ void testTransform(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20], after=[2, x, 20], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 10, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 20, ], after=[2, 20, x], op=UPDATE, meta=()}");
Review Comment:
Could you please also add tests to verify if `ValuesDatabase#getTableSchema` is as expected?
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo closed pull request #3455: [FLINK-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event
URL: https://github.com/apache/flink-cdc/pull/3455
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708810777
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -399,8 +399,8 @@ void testTransformTwice(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 11], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 22], after=[2, x, 22], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 11, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 22, ], after=[2, 22, x], op=UPDATE, meta=()}");
Review Comment:
Yes. Transform semantic has changed. The previous discussion resulted in the current code architecture being unable to achieve the expected semantics.
Based on the following code, it may be possible to achieve.
https://github.com/apache/flink-cdc/pull/3285
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1708816338
##########
flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java:
##########
@@ -332,8 +332,8 @@ void testTransform(ValuesDataSink.SinkApi sinkApi) throws Exception {
"AddColumnEvent{tableId=default_namespace.default_schema.table1, addedColumns=[ColumnWithPosition{column=`col3` STRING, position=LAST, existedColumnName=null}]}",
"RenameColumnEvent{tableId=default_namespace.default_schema.table1, nameMapping={col2=newCol2, col3=newCol3}}",
"DropColumnEvent{tableId=default_namespace.default_schema.table1, droppedColumnNames=[newCol2]}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 1, 10], after=[], op=DELETE, meta=()}",
- "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, , 20], after=[2, x, 20], op=UPDATE, meta=()}");
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[1, 10, 1], after=[], op=DELETE, meta=()}",
+ "DataChangeEvent{tableId=default_namespace.default_schema.table1, before=[2, 20, ], after=[2, 20, x], op=UPDATE, meta=()}");
Review Comment:
I will have a try tonight.
--
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-35774][cdc-runtime] Fix the cache of transform is not updated after process schema change event [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3455:
URL: https://github.com/apache/flink-cdc/pull/3455#discussion_r1668456054
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformDataOperator.java:
##########
Review Comment:
I'm a bit unsure if `transformProjectionProcessorMap` and `transformFilterProcessorMap` be flushed when upstream schema change event arrives at `TransformDataOperator`?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformDataOperator.java:
##########
Review Comment:
I'm a bit unsure if `transformProjectionProcessorMap` and `transformFilterProcessorMap` will be flushed when upstream schema change event arrives at `TransformDataOperator`?
--
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