You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "snuyanzin (via GitHub)" <gi...@apache.org> on 2025/11/18 23:11:50 UTC
[PR] [FLINK-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
snuyanzin opened a new pull request, #27250:
URL: https://github.com/apache/flink/pull/27250
## What is the purpose of the change
The current issue is that Changelog normalize uses Calcite's CNF in order to calculate common conditions between several changelog normalize nodes if they share same table scan. In case of query with huge amount of conditions connected with `OR` it might lead to OOM even if another changelog normalize nodes doesn't have conditions.
This code removes usage of CNF in marking program and fixes the behavior for the case of other changelog normalize nodes have no conditions.
The rest part of the issue will be solved separately
## Brief change log
FlinkMarkChangelogNormalizeProgram.java
## Verifying this change
1. existing tests
2. I have locally query leading to OOM, not sure if a huge query (about 1k lines including table/view creation) should be a part of test...
3. just check that there is no occurrences of `toCnf` in that class anymore and existing tests continue passing
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): ( no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable )
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #27250:
URL: https://github.com/apache/flink/pull/27250#discussion_r2545598342
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java:
##########
@@ -149,27 +154,47 @@ public RelNode optimize(RelNode root, StreamOptimizeContext context) {
continue;
}
- final Set<RexNode> common = calculateCommonCondition(changelogNormalizeContexts);
+ final List<RexNode> commons =
+ calculateCommonCondition(rexBuilder, changelogNormalizeContexts);
for (ChangelogNormalizeContext ctx : changelogNormalizeContexts) {
ctx.getChangelogNormalize().markSourceReuse();
- if (!common.isEmpty()) {
- ctx.getChangelogNormalize().setCommonFilter(common.toArray(new RexNode[0]));
+ if (!commons.isEmpty()) {
+ ctx.getChangelogNormalize().setCommonFilter(commons.toArray(new RexNode[0]));
}
}
}
return root;
}
- private Set<RexNode> calculateCommonCondition(
- List<ChangelogNormalizeContext> changelogNormalizeContexts) {
- changelogNormalizeContexts.sort(Comparator.comparingInt(o -> o.getConditions().size()));
- final Set<RexNode> common =
- new HashSet<>(changelogNormalizeContexts.get(0).getConditions());
+ private List<RexNode> calculateCommonCondition(
Review Comment:
can we have a unit test for this change please.
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27250:
URL: https://github.com/apache/flink/pull/27250#discussion_r2545646914
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java:
##########
@@ -149,27 +154,47 @@ public RelNode optimize(RelNode root, StreamOptimizeContext context) {
continue;
}
- final Set<RexNode> common = calculateCommonCondition(changelogNormalizeContexts);
+ final List<RexNode> commons =
+ calculateCommonCondition(rexBuilder, changelogNormalizeContexts);
for (ChangelogNormalizeContext ctx : changelogNormalizeContexts) {
ctx.getChangelogNormalize().markSourceReuse();
- if (!common.isEmpty()) {
- ctx.getChangelogNormalize().setCommonFilter(common.toArray(new RexNode[0]));
+ if (!commons.isEmpty()) {
+ ctx.getChangelogNormalize().setCommonFilter(commons.toArray(new RexNode[0]));
}
}
}
return root;
}
- private Set<RexNode> calculateCommonCondition(
- List<ChangelogNormalizeContext> changelogNormalizeContexts) {
- changelogNormalizeContexts.sort(Comparator.comparingInt(o -> o.getConditions().size()));
- final Set<RexNode> common =
- new HashSet<>(changelogNormalizeContexts.get(0).getConditions());
+ private List<RexNode> calculateCommonCondition(
Review Comment:
the only reason to refactor this code is memory issue in case of query with huge list of conditions connected by `OR`
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin merged PR #27250:
URL: https://github.com/apache/flink/pull/27250
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27250:
URL: https://github.com/apache/flink/pull/27250#discussion_r2539926404
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java:
##########
@@ -182,12 +207,24 @@ private void gatherTableScanToChangelogNormalizeMap(
(StreamPhysicalChangelogNormalize) input;
if (curRelNode instanceof StreamPhysicalCalc) {
StreamPhysicalCalc calc = (StreamPhysicalCalc) curRelNode;
- final List<RexNode> conditions =
- FlinkRexUtil.extractConjunctiveConditions(
- rexBuilder, calc.getProgram());
Review Comment:
this is the problem code leading to OOM
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #27250:
URL: https://github.com/apache/flink/pull/27250#issuecomment-3551791988
something wrong happened with CI
all PR's CIs are now failing with same issue
```
2025-11-19T09:01:35.2198281Z #6 ERROR: process "/bin/sh -c apt-get update && apt-get install build-essential -y" did not complete successfully: exit code: 100
```
https://dev.azure.com/apache-flink/apache-flink/_build?definitionId=2&_a=summary
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #27250:
URL: https://github.com/apache/flink/pull/27250#issuecomment-3549840685
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9058051562c0ccc478035d95a4cbdfc73b70dfb7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9058051562c0ccc478035d95a4cbdfc73b70dfb7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9058051562c0ccc478035d95a4cbdfc73b70dfb7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27250:
URL: https://github.com/apache/flink/pull/27250#discussion_r2539930231
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java:
##########
@@ -149,27 +154,47 @@ public RelNode optimize(RelNode root, StreamOptimizeContext context) {
continue;
}
- final Set<RexNode> common = calculateCommonCondition(changelogNormalizeContexts);
+ final List<RexNode> commons =
+ calculateCommonCondition(rexBuilder, changelogNormalizeContexts);
for (ChangelogNormalizeContext ctx : changelogNormalizeContexts) {
ctx.getChangelogNormalize().markSourceReuse();
- if (!common.isEmpty()) {
- ctx.getChangelogNormalize().setCommonFilter(common.toArray(new RexNode[0]));
+ if (!commons.isEmpty()) {
+ ctx.getChangelogNormalize().setCommonFilter(commons.toArray(new RexNode[0]));
}
}
}
return root;
}
- private Set<RexNode> calculateCommonCondition(
- List<ChangelogNormalizeContext> changelogNormalizeContexts) {
- changelogNormalizeContexts.sort(Comparator.comparingInt(o -> o.getConditions().size()));
- final Set<RexNode> common =
- new HashSet<>(changelogNormalizeContexts.get(0).getConditions());
+ private List<RexNode> calculateCommonCondition(
+ RexBuilder rexBuilder, List<ChangelogNormalizeContext> changelogNormalizeContexts) {
+ if (changelogNormalizeContexts.stream()
+ .map(ChangelogNormalizeContext::getConditions)
+ .anyMatch(Objects::isNull)) {
+ return List.of();
+ }
+
+ final RexNode or =
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.OR,
+ changelogNormalizeContexts.stream()
+ .map(ChangelogNormalizeContext::getConditions)
+ .collect(Collectors.toList()));
+ final RexCall factors = (RexCall) RexUtil.pullFactors(rexBuilder, or);
- for (int i = 1; i < changelogNormalizeContexts.size() && !common.isEmpty(); i++) {
- common.retainAll(changelogNormalizeContexts.get(i).getConditions());
+ final List<RexNode> commonCondition = new ArrayList<>();
+ // Since we are interested in factors only then look for AND
+ if (factors.getKind() == SqlKind.AND) {
+ for (RexNode node : factors.getOperands()) {
+ // If there is OR on top level then it is not a common factor anymore
+ if (node.getKind() == SqlKind.OR) {
+ break;
+ }
+ commonCondition.addAll(RelOptUtil.conjunctions(RexUtil.toCnf(rexBuilder, node)));
Review Comment:
it is ok to run it on small pieces like factors, however running on huge conditions connected with `OR` leads to memory issues
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #27250:
URL: https://github.com/apache/flink/pull/27250#discussion_r2545643628
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java:
##########
@@ -149,27 +154,47 @@ public RelNode optimize(RelNode root, StreamOptimizeContext context) {
continue;
}
- final Set<RexNode> common = calculateCommonCondition(changelogNormalizeContexts);
+ final List<RexNode> commons =
+ calculateCommonCondition(rexBuilder, changelogNormalizeContexts);
for (ChangelogNormalizeContext ctx : changelogNormalizeContexts) {
ctx.getChangelogNormalize().markSourceReuse();
- if (!common.isEmpty()) {
- ctx.getChangelogNormalize().setCommonFilter(common.toArray(new RexNode[0]));
+ if (!commons.isEmpty()) {
+ ctx.getChangelogNormalize().setCommonFilter(commons.toArray(new RexNode[0]));
}
}
}
return root;
}
- private Set<RexNode> calculateCommonCondition(
- List<ChangelogNormalizeContext> changelogNormalizeContexts) {
- changelogNormalizeContexts.sort(Comparator.comparingInt(o -> o.getConditions().size()));
- final Set<RexNode> common =
- new HashSet<>(changelogNormalizeContexts.get(0).getConditions());
+ private List<RexNode> calculateCommonCondition(
Review Comment:
as mentioned above it is already covered by unit tests from https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java
--
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-38693][table] Changelog normalize shouldn't call CNF for marking program [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #27250:
URL: https://github.com/apache/flink/pull/27250#issuecomment-3558435880
rebased to have green ci after FLINK-38700 is fixed
--
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