[FLINK-30597] Refactor FlinkProjectJoinTransposeRule and FlinkPruneEm…

…ptyRules
apache · Jan 9, 2023 · 4673fa7 · 4673fa7
1 parent 8a4ee95
commit 4673fa7
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 10 deletions.
diff --git a/...java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectJoinTransposeRule.java b/...java/org/apache/flink/table/planner/plan/rules/logical/FlinkProjectJoinTransposeRule.java
@@ -17,7 +17,8 @@
 
 package org.apache.flink.table.planner.plan.rules.logical;
 
-import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.rules.ProjectJoinTransposeRule;
 import org.apache.calcite.rel.rules.PushProjector;
 
@@ -26,11 +27,24 @@
  * org.apache.calcite.rel.core.Join} by splitting the projection into a projection on top of each
  * child of the join.
  */
-public class FlinkProjectJoinTransposeRule {
-    public static final ProjectJoinTransposeRule.Config DEFAULT =
-            ProjectJoinTransposeRule.Config.DEFAULT
-                    .withPreserveExprCondition(PushProjector.ExprCondition.FALSE)
-                    .withRelBuilderFactory(RelFactories.LOGICAL_BUILDER)
-                    .withDescription("FlinkProjectJoinTransposeRule")
-                    .as(ProjectJoinTransposeRule.Config.class);
+public class FlinkProjectJoinTransposeRule extends ProjectJoinTransposeRule {
+    public static ProjectJoinTransposeRule INSTANCE =
+            new FlinkProjectJoinTransposeRule(
+                    Config.DEFAULT
+                            .withPreserveExprCondition(PushProjector.ExprCondition.FALSE)
+                            .withDescription("FlinkProjectJoinTransposeRule")
+                            .as(Config.class));
+
+    private FlinkProjectJoinTransposeRule(Config config) {
+        super(config);
+    }
+
+    @Override
+    public void onMatch(RelOptRuleCall call) {
+        final Join join = call.rel(1);
+        if (!join.getJoinType().projectsRight()) {
+            return; // TODO: support SEMI/ANTI join later
+        }
+        super.onMatch(call);
+    }
 }
diff --git a/...planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala b/...planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -189,7 +189,7 @@ object FlinkBatchRuleSets {
     CoreRules.PROJECT_FILTER_TRANSPOSE,
     // push a projection to the children of a non semi/anti join
     // push all expressions to handle the time indicator correctly
-    FlinkProjectJoinTransposeRule.DEFAULT.toRule,
+    FlinkProjectJoinTransposeRule.INSTANCE,
     // push a projection to the children of a semi/anti Join
     ProjectSemiAntiJoinTransposeRule.INSTANCE,
     // merge projections

diff --git a/...lanner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/...lanner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -193,7 +193,7 @@ object FlinkStreamRuleSets {
     CoreRules.PROJECT_FILTER_TRANSPOSE,
     // push a projection to the children of a non semi/anti join
     // push all expressions to handle the time indicator correctly
-    FlinkProjectJoinTransposeRule.DEFAULT.toRule,
+    FlinkProjectJoinTransposeRule.INSTANCE,
     // push a projection to the children of a semi/anti Join
     ProjectSemiAntiJoinTransposeRule.INSTANCE,
     // merge projections