You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "MOBIN-F (via GitHub)" <gi...@apache.org> on 2024/08/24 15:30:19 UTC
[PR] [FLINK-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
MOBIN-F opened a new pull request, #3572:
URL: https://github.com/apache/flink-cdc/pull/3572
Currently, such transform rule could not work:
```yaml
transform:
projection: 'A' as result
filter: tag >= 0
projection: score as result
filter: tag < 0
```
Here, `score` column will be filtered out in PreTransform stage, since it wasn't referenced in the first transform rule. As a result, the following transform rule will fail since `score` does not exist in PostTransform operator.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2475350233
@MOBIN-F Would you like to open a backport PR for `release-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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2469497337
> Thanks for cleaning up `PreTransformOperator`! Looks much neater now, just left some trivial comments.
>
> Also +@aiwenmo who is familiar with transform modules.
Hi. @MOBIN-F. Thanks for the contribution. And I will assist in reviewing the code 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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2472340148
oceanbase tests failure,but it seems to have nothing to do with this 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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1836206210
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java:
##########
@@ -342,27 +342,77 @@ private CreateTableEvent transformCreateTableEvent(CreateTableEvent createTableE
}
}
+ cachePreTransformProcessor(tableId, createTableEvent.getSchema());
+ if (preTransformProcessorMap.containsKey(tableId)) {
+ return preTransformProcessorMap
+ .get(tableId)
+ .preTransformCreateTableEvent(createTableEvent);
+ }
+ return createTableEvent;
+ }
+
+ private void cachePreTransformProcessor(TableId tableId, Schema tableSchema) {
+ LinkedHashSet<Column> referencedColumnsSet = new LinkedHashSet<>();
+ boolean hasMatchTransform = false;
for (PreTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
- if (selectors.isMatch(tableId) && transform.getProjection().isPresent()) {
+ if (!transform.getSelectors().isMatch(tableId)) {
+ continue;
+ }
+ if (!transform.getProjection().isPresent()) {
+ processProjectionTransform(tableId, tableSchema, referencedColumnsSet, null);
+ hasMatchTransform = true;
+ } else {
Review Comment:
Is it correct if we omit this branch and let it fallback to `if(!hasMatchTransform)` condition? Seems the handling logic is the same.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java:
##########
@@ -342,27 +342,77 @@ private CreateTableEvent transformCreateTableEvent(CreateTableEvent createTableE
}
}
+ cachePreTransformProcessor(tableId, createTableEvent.getSchema());
+ if (preTransformProcessorMap.containsKey(tableId)) {
+ return preTransformProcessorMap
+ .get(tableId)
+ .preTransformCreateTableEvent(createTableEvent);
+ }
+ return createTableEvent;
+ }
+
+ private void cachePreTransformProcessor(TableId tableId, Schema tableSchema) {
+ LinkedHashSet<Column> referencedColumnsSet = new LinkedHashSet<>();
+ boolean hasMatchTransform = false;
for (PreTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
- if (selectors.isMatch(tableId) && transform.getProjection().isPresent()) {
+ if (!transform.getSelectors().isMatch(tableId)) {
+ continue;
+ }
+ if (!transform.getProjection().isPresent()) {
+ processProjectionTransform(tableId, tableSchema, referencedColumnsSet, null);
+ hasMatchTransform = true;
+ } else {
TransformProjection transformProjection = transform.getProjection().get();
- TransformFilter transformFilter = transform.getFilter().orElse(null);
if (transformProjection.isValid()) {
- if (!preTransformProcessorMap.containsKey(tableId)) {
- preTransformProcessorMap.put(
- tableId,
- new PreTransformProcessor(
- tableChangeInfo, transformProjection, transformFilter));
- }
- PreTransformProcessor preTransformProcessor =
- preTransformProcessorMap.get(tableId);
- // TODO: Currently this code wrongly filters out rows that weren't referenced in
- // the first matching transform rule but in the following transform rules.
- return preTransformProcessor.preTransformCreateTableEvent(createTableEvent);
+ processProjectionTransform(
+ tableId, tableSchema, referencedColumnsSet, transform);
+ hasMatchTransform = true;
}
}
}
- return createTableEvent;
+ if (!hasMatchTransform) {
+ processProjectionTransform(tableId, tableSchema, referencedColumnsSet, null);
+ }
+ }
+
+ public void processProjectionTransform(
+ TableId tableId,
+ Schema tableSchema,
+ LinkedHashSet<Column> referencedColumnsSet,
+ @Nullable PreTransformer transform) {
+ // If this TableId isn't presented in any transform block, it should behave like a "*"
+ // projection and should be regarded as asterisk-ful.
+ if (transform == null) {
+ referencedColumnsSet.addAll(tableSchema.getColumns());
+ hasAsteriskMap.put(tableId, true);
+ } else {
+ TransformProjection transformProjection = transform.getProjection().get();
+ boolean hasAsterisk = TransformParser.hasAsterisk(transformProjection.getProjection());
+ if (hasAsterisk) {
+ referencedColumnsSet.addAll(tableSchema.getColumns());
+ hasAsteriskMap.put(tableId, true);
+ } else {
+ TransformFilter transformFilter = transform.getFilter().orElse(null);
+ List<Column> referencedColumns =
+ TransformParser.generateReferencedColumns(
+ transformProjection.getProjection(),
+ transformFilter != null ? transformFilter.getExpression() : null,
+ tableSchema.getColumns());
+ // update referenced columns of other projections of the same tableId, if any
+ referencedColumnsSet.addAll(referencedColumns);
+ hasAsteriskMap.put(tableId, false);
Review Comment:
Should we check if `hasAsteriskMap[tableId]` is already `true`, perhaps set by a previous transform rule?
Add: I think it should not be a problem, since if someone puts asteriskful and non-asteriskful rules together, it will fail when column count changes from upstream anyway.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1837413936
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java:
##########
@@ -526,4 +526,39 @@ void testMetadataTransform() throws Exception {
new StreamRecord<>(
new CreateTableEvent(METADATA_TABLEID, EXPECTED_METADATA_SCHEMA)));
}
+
+ @Test
+ void testMultiTransform() throws Exception {
+ PreTransformOperator transform =
+ PreTransformOperator.newBuilder()
Review Comment:
done~, also above
added some tests
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2461825121
@MOBIN-F Could you help rebase this 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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2463867357
> @MOBIN-F Could you kindly rebase again?
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1836241798
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java:
##########
@@ -386,37 +436,16 @@ private Schema transformSchemaMetaData(
return schemaBuilder.build();
}
- private DataChangeEvent processDataChangeEvent(DataChangeEvent dataChangeEvent)
- throws Exception {
- TableId tableId = dataChangeEvent.tableId();
- for (PreTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
-
- if (selectors.isMatch(tableId) && transform.getProjection().isPresent()) {
- TransformProjection transformProjection = transform.getProjection().get();
- TransformFilter transformFilter = transform.getFilter().orElse(null);
- if (transformProjection.isValid()) {
- return processProjection(transformProjection, transformFilter, dataChangeEvent);
- }
- }
+ private DataChangeEvent processDataChangeEvent(DataChangeEvent dataChangeEvent) {
+ if (!transforms.isEmpty()) {
+ return processProjection(dataChangeEvent);
}
return dataChangeEvent;
}
Review Comment:
Seems the logic is too simple and not worth being an independent method.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1836254296
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java:
##########
@@ -526,4 +526,39 @@ void testMetadataTransform() throws Exception {
new StreamRecord<>(
new CreateTableEvent(METADATA_TABLEID, EXPECTED_METADATA_SCHEMA)));
}
+
+ @Test
+ void testMultiTransform() throws Exception {
+ PreTransformOperator transform =
+ PreTransformOperator.newBuilder()
+ .addTransform(
+ CUSTOMERS_TABLEID.identifier(),
+ "col1, upper(col1) col12",
+ "col1 = 'col1'",
+ "col2",
+ "col12",
+ "key1=value1,key2=value2")
+ .addTransform(
+ CUSTOMERS_TABLEID.identifier(),
+ "col2, upper(col2) col12",
+ "col1 != 'col1'",
+ "col2",
+ "col12",
+ "key1=value1,key2=value2")
+ .build();
Review Comment:
Considering this bug is sequential specific, swapping rules to verify it would be nice.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2467581975
I previously ignored the scenario where projection was missing. I fixed it and added integrated / E2e tests , Can you review it again? @yuxiqian
In my CI Aaction,the tests passed
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2463861429
@MOBIN-F Could you kindly rebase again?
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2475352297
> @MOBIN-F Would you like to open a backport PR for `release-3.2` branch?
OK
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1836206210
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java:
##########
@@ -342,27 +342,77 @@ private CreateTableEvent transformCreateTableEvent(CreateTableEvent createTableE
}
}
+ cachePreTransformProcessor(tableId, createTableEvent.getSchema());
+ if (preTransformProcessorMap.containsKey(tableId)) {
+ return preTransformProcessorMap
+ .get(tableId)
+ .preTransformCreateTableEvent(createTableEvent);
+ }
+ return createTableEvent;
+ }
+
+ private void cachePreTransformProcessor(TableId tableId, Schema tableSchema) {
+ LinkedHashSet<Column> referencedColumnsSet = new LinkedHashSet<>();
+ boolean hasMatchTransform = false;
for (PreTransformer transform : transforms) {
- Selectors selectors = transform.getSelectors();
- if (selectors.isMatch(tableId) && transform.getProjection().isPresent()) {
+ if (!transform.getSelectors().isMatch(tableId)) {
+ continue;
+ }
+ if (!transform.getProjection().isPresent()) {
+ processProjectionTransform(tableId, tableSchema, referencedColumnsSet, null);
+ hasMatchTransform = true;
+ } else {
Review Comment:
Is it correct if we omit this branch and let it fallback to `if(!hasMatchTransform)` condition? Seems the handling logic is the same.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#discussion_r1836260088
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperatorTest.java:
##########
@@ -526,4 +526,39 @@ void testMetadataTransform() throws Exception {
new StreamRecord<>(
new CreateTableEvent(METADATA_TABLEID, EXPECTED_METADATA_SCHEMA)));
}
+
+ @Test
+ void testMultiTransform() throws Exception {
+ PreTransformOperator transform =
+ PreTransformOperator.newBuilder()
Review Comment:
Maybe we can also cover cases when projection is `*` or omitted here
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2436543491
This pull request has been automatically marked as stale because it has not had recent activity for 60 days. It will be closed in 30 days if no further activity occurs.
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2463886561
CI failure,I will recheck the code
--
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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2461897844
> @MOBIN-F Could you help rebase this PR?
@leonardBang 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-36093][transform] fix preTransformoperator wrongly filters columns when multiple transform [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #3572:
URL: https://github.com/apache/flink-cdc/pull/3572#issuecomment-2463611452
CI failure,It seems that this PR(https://github.com/apache/flink-cdc/pull/3696) should be merged first
--
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