[FLINK-20887][table-planner] Disable project merge during sql2rel phase by default to avoid incorrectly project merge #22827

lincoln-lil · 2023-06-19T15:07:18Z

What is the purpose of the change

FLINK-30841 fixes incorrect project merge in the optimization phase (prevents projects containing non-deterministic calls from being merged with 'wrong results' that are not expected by users, also FLINK-15366 & FLINK-30006 had taken some efforts), but didn't fix the problem completely, as in the case described in this issue, we need to turn off the relevant optimizations in the sql2rel phase(add internal config and turn off by default) to fix it completely.

some more detailed explanation of the relate case, if we keep project merge in sql2rel phase, we won't have any chance to fix it because the projects are been merge already, see the original relnode tree after sql2rel:

== Abstract Syntax Tree ==
LogicalProject(exprs=[[-(+(RAND(), 7), +(RAND(), 5))]])
+- LogicalTableScan(table=[[default_catalog, default_database, SmallTable3]])

Note: this is not a new feature(the newly added config option is not recommended to use, only used as a rollback configuration when users need to restore the behavior of an older version)

Brief change log

1. pre-step: turn off the project merge optimization in the sql2rel phase by default
1. fix incorrectly calc merge (but need to change related rules)
1. fix the fail case due to incorrect structured type's nullability (more details can see FLINK-31830)
1. fix the incorrectly aggregate project merge for LogicalWindowAggregate

Verifying this change

existing and newly added cases

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): (yes)
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)

flinkbot · 2023-06-19T15:13:51Z

CI report:

a6bdb84 Azure: SUCCESS
4f95c14 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

swuferhong

Thanks for the PR. It generally LGTM. Just leave some minor comments.

swuferhong · 2023-06-21T06:00:53Z

...rg/apache/flink/table/planner/plan/rules/logical/PushProjectIntoTableSourceScanRuleTest.java

+    private void replaceProgramWithProjectMergeRule() {
+        FlinkChainedProgram programs = new FlinkChainedProgram<BatchOptimizeContext>();
+        programs.addLast(
+                "rules",
+                FlinkHepRuleSetProgramBuilder.<BatchOptimizeContext>newBuilder()
+                        .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE())
+                        .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+                        .add(
+                                RuleSets.ofList(
+                                        CoreRules.PROJECT_MERGE,
+                                        PushProjectIntoTableSourceScanRule.INSTANCE))
+                        .build());
+        util().replaceBatchProgram(programs);
+    }


Why do we need to add this method in the pre-step commit？ Maybe it need to move to the Step2.

the ast changes after we disable project merge during sql2rel phase by default in the 1st commit, and this rule test will fail, so it should stay in 1st commit

swuferhong · 2023-06-21T06:17:37Z

...r/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectMergeRule.java

+import org.apache.calcite.rel.rules.ProjectMergeRule;
+
+/**
+ * Extends calcite's FilterCalcMergeRule for streaming scenario, modification: does not merge the


Extends calcite's ProjectMergeRule

swuferhong · 2023-06-21T06:35:06Z

...c/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectCalcMergeRule.java

+    @Override
+    public void onMatch(RelOptRuleCall call) {
+        LogicalProject project = call.rel(0);
+        LogicalCalc calc = call.rel(1);
+
+        List<RexNode> expandProjects =
+                calc.getProgram().getProjectList().stream()
+                        .map(p -> calc.getProgram().expandLocalRef(p))
+                        .collect(Collectors.toList());
+        InputRefVisitor inputRefVisitor = new InputRefVisitor();
+        project.getProjects().forEach(p -> p.accept(inputRefVisitor));
+        boolean existNonDeterministicRef =
+                Arrays.stream(inputRefVisitor.getFields())
+                        .anyMatch(i -> !RexUtil.isDeterministic(expandProjects.get(i)));
+
+        if (!existNonDeterministicRef) {
+            super.onMatch(call);
+        }
+    }
+}


is it more reasonable to re-implement matches() method here, which to be consistent with ProjectMergeRule and CalcMergeRule.

make sense, I'll update it.

swuferhong · 2023-06-21T06:37:36Z

...nk-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

+    }
+  }
+
+  private def mergeable(


Add some commets?

swuferhong · 2023-06-21T06:39:19Z

...nk-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

+      bottomProgram.getProjectList
+        .map(bottomProgram.expandLocalRef)
+        .toList)
+  }


Nits: To achieve the goal of scala-free, would it be better to put these utils methods in a Java class?

Ok, this encourage me creating a new class FlinkRelUtil which is more appropriately than the FlinkRexUtil here

swuferhong · 2023-06-21T06:40:23Z

...nk-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

+   * @param deep
+   * @param refCounts


Missing Parameter Description

swuferhong · 2023-06-21T06:45:53Z

...c/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectCalcMergeRule.java

+        project.getProjects().forEach(p -> p.accept(inputRefVisitor));
+        boolean existNonDeterministicRef =
+                Arrays.stream(inputRefVisitor.getFields())
+                        .anyMatch(i -> !RexUtil.isDeterministic(expandProjects.get(i)));


Can resue FlinkRexUtil.Mergeable()?

From a maintainability point of view, it is indeed better to use a unified reusable logic here

swuferhong · 2023-06-21T06:51:23Z

...r/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectMergeRule.java

+        if (FlinkRexUtil.isMergeable(topProject, bottomProject)) {
+            super.onMatch(call);
+        }
+    }


Ditto: is it more reasonable to re-implement matches() method here ?

yes, will update

lincoln-lil

@swuferhong thank you for reviewing this! I've addressed your comments and updated the pr

lincoln-lil · 2023-06-21T09:36:20Z

...c/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectCalcMergeRule.java

+    @Override
+    public void onMatch(RelOptRuleCall call) {
+        LogicalProject project = call.rel(0);
+        LogicalCalc calc = call.rel(1);
+
+        List<RexNode> expandProjects =
+                calc.getProgram().getProjectList().stream()
+                        .map(p -> calc.getProgram().expandLocalRef(p))
+                        .collect(Collectors.toList());
+        InputRefVisitor inputRefVisitor = new InputRefVisitor();
+        project.getProjects().forEach(p -> p.accept(inputRefVisitor));
+        boolean existNonDeterministicRef =
+                Arrays.stream(inputRefVisitor.getFields())
+                        .anyMatch(i -> !RexUtil.isDeterministic(expandProjects.get(i)));
+
+        if (!existNonDeterministicRef) {
+            super.onMatch(call);
+        }
+    }
+}


make sense, I'll update it.

lincoln-lil · 2023-06-21T09:51:26Z

...r/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectMergeRule.java

+        if (FlinkRexUtil.isMergeable(topProject, bottomProject)) {
+            super.onMatch(call);
+        }
+    }


yes, will update

lincoln-lil · 2023-06-21T09:51:54Z

...nk-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

+    }
+  }
+
+  private def mergeable(


lincoln-lil · 2023-06-21T09:54:39Z

...rg/apache/flink/table/planner/plan/rules/logical/PushProjectIntoTableSourceScanRuleTest.java

+    private void replaceProgramWithProjectMergeRule() {
+        FlinkChainedProgram programs = new FlinkChainedProgram<BatchOptimizeContext>();
+        programs.addLast(
+                "rules",
+                FlinkHepRuleSetProgramBuilder.<BatchOptimizeContext>newBuilder()
+                        .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE())
+                        .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+                        .add(
+                                RuleSets.ofList(
+                                        CoreRules.PROJECT_MERGE,
+                                        PushProjectIntoTableSourceScanRule.INSTANCE))
+                        .build());
+        util().replaceBatchProgram(programs);
+    }


the ast changes after we disable project merge during sql2rel phase by default in the 1st commit, and this rule test will fail, so it should stay in 1st commit

lincoln-lil · 2023-06-21T13:33:36Z

...c/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectCalcMergeRule.java

+        project.getProjects().forEach(p -> p.accept(inputRefVisitor));
+        boolean existNonDeterministicRef =
+                Arrays.stream(inputRefVisitor.getFields())
+                        .anyMatch(i -> !RexUtil.isDeterministic(expandProjects.get(i)));


From a maintainability point of view, it is indeed better to use a unified reusable logic here

lincoln-lil · 2023-06-21T13:37:22Z

...nk-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/FlinkRexUtil.scala

+      bottomProgram.getProjectList
+        .map(bottomProgram.expandLocalRef)
+        .toList)
+  }


Ok, this encourage me creating a new class FlinkRelUtil which is more appropriately than the FlinkRexUtil here

swuferhong

Thanks, @lincoln-lil . This PR LGTM now.

…ql2rel phase by default

…ated rules

…bility of structured type

…r LogicalWindowAggregate

lincoln-lil · 2023-06-25T05:54:19Z

The squashed commits after rebased master branch succeed in my private pipeline: https://dev.azure.com/lincoln86xy/flink/_build/results?buildId=537&view=results

lincoln-lil force-pushed the FLINK-20887 branch from 0f1b59c to 590a207 Compare June 20, 2023 02:58

lincoln-lil force-pushed the FLINK-20887 branch 3 times, most recently from e284871 to f774338 Compare June 21, 2023 06:42

swuferhong reviewed Jun 21, 2023

View reviewed changes

lincoln-lil commented Jun 21, 2023

View reviewed changes

swuferhong approved these changes Jun 25, 2023

View reviewed changes

lincoln-lil added 4 commits June 25, 2023 09:23

[FLINK-20887][table-planner] Pre-step: disable project merge during s…

380db9a

…ql2rel phase by default

[FLINK-20887][table-planner] Step2: fix incorrect calc merge via rele…

a286d50

…ated rules

[FLINK-20887][table-planner] Fix the fail case due to incorrect nulla…

bb18576

…bility of structured type

[FLINK-20887][table-planner] Fix incorrect aggregate project merge fo…

4f95c14

…r LogicalWindowAggregate

lincoln-lil force-pushed the FLINK-20887 branch from a6bdb84 to 4f95c14 Compare June 25, 2023 01:27

lincoln-lil closed this Jun 25, 2023

flinkbot added the component=TableSQL/Planner label Apr 4, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-20887][table-planner] Disable project merge during sql2rel phase by default to avoid incorrectly project merge #22827

[FLINK-20887][table-planner] Disable project merge during sql2rel phase by default to avoid incorrectly project merge #22827

lincoln-lil commented Jun 19, 2023 •

edited

flinkbot commented Jun 19, 2023 •

edited

swuferhong left a comment

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong Jun 21, 2023

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong Jun 21, 2023

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong Jun 21, 2023

lincoln-lil Jun 21, 2023

lincoln-lil left a comment

lincoln-lil Jun 21, 2023

lincoln-lil Jun 21, 2023

lincoln-lil Jun 21, 2023

lincoln-lil Jun 21, 2023

lincoln-lil Jun 21, 2023

lincoln-lil Jun 21, 2023

swuferhong left a comment

lincoln-lil commented Jun 25, 2023

[FLINK-20887][table-planner] Disable project merge during sql2rel phase by default to avoid incorrectly project merge #22827

[FLINK-20887][table-planner] Disable project merge during sql2rel phase by default to avoid incorrectly project merge #22827

Conversation

lincoln-lil commented Jun 19, 2023 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jun 19, 2023 • edited

CI report:

swuferhong left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lincoln-lil left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

swuferhong left a comment

Choose a reason for hiding this comment

lincoln-lil commented Jun 25, 2023

lincoln-lil commented Jun 19, 2023 •

edited

flinkbot commented Jun 19, 2023 •

edited