[CALCITE-6891] Implement IntersectReorderRule

Author: xiedeyantu

core/src/main/java/org/apache/calcite/rel/rules/CoreRules.java

@@ -349,6 +349,11 @@ private CoreRules() {}
   public static final UnionMergeRule INTERSECT_MERGE =
       UnionMergeRule.Config.INTERSECT.toRule();
 
+  /** Planner rule that reorders inputs of an {@link Intersect} to put smaller inputs first.
+   * This helps reduce the size of intermediate results. */
+  public static final IntersectReorderRule INTERSECT_REORDER =
+      IntersectReorderRule.Config.DEFAULT.toRule();
+
   /** Rule that translates a distinct
    * {@link Intersect} into a group of operators
    * composed of {@link Union}, {@link Aggregate}, etc. */

core/src/main/java/org/apache/calcite/rel/rules/IntersectReorderRule.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.rel.rules;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Intersect;
+import org.apache.calcite.rel.logical.LogicalIntersect;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.Pair;
+
+import org.immutables.value.Value;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Planner rule that reorders inputs of an {@link Intersect} to put smaller inputs first.
+ * This helps reduce the size of intermediate results.
+ *
+ * <p>Intersect(A, B, ...) where B is smallest will reorder to Intersect(B, A, ...)
+ */
+@Value.Enclosing
+public class IntersectReorderRule extends RelRule<IntersectReorderRule.Config>
+  implements SubstitutionRule {
+  /** Creates an IntersectReorderRule. */
+  protected IntersectReorderRule(Config config) {
+    super(config);
+  }
+
+  @Override public void onMatch(RelOptRuleCall call) {
+    final Intersect intersect = call.rel(0);
+    final RelMetadataQuery mq = call.getMetadataQuery();
+    final List<RelNode> inputs = intersect.getInputs();
+
+    final List<Pair<RelNode, Double>> inputsWithRowCounts = new ArrayList<>();
+
+    for (RelNode input : inputs) {
+      Double rowCount = mq.getRowCount(input);
+      inputsWithRowCounts.add(Pair.of(input, rowCount));
+    }
+
+    inputsWithRowCounts.sort((a, b) -> Double.compare(a.right, b.right));
+
+    boolean needsReorder = false;
+    for (int i = 0; i < inputs.size(); i++) {
+      if (inputs.get(i) != inputsWithRowCounts.get(i).left) {
+        needsReorder = true;
+        break;
+      }
+    }
+
+    if (!needsReorder) {
+      return;
+    }
+
+    final List<RelNode> newInputs = inputsWithRowCounts.stream()
+        .map(pair -> pair.left)
+        .collect(Collectors.toList());
+
+    final RelBuilder relBuilder = call.builder();
+    relBuilder.pushAll(newInputs);
+    relBuilder.intersect(intersect.all, newInputs.size());
+
+    call.transformTo(relBuilder.build());
+  }
+
+  /** Rule configuration. */
+  @Value.Immutable
+  public interface Config extends RelRule.Config {
+    Config DEFAULT = ImmutableIntersectReorderRule.Config.of()
+        .withOperandSupplier(b0 ->
+            b0.operand(LogicalIntersect.class)
+                .predicate(intersect -> intersect.getInputs().size() > 1)
+                .anyInputs())
+        .withDescription("IntersectReorderRule");
+
+    @Override default IntersectReorderRule toRule() {
+      return new IntersectReorderRule(this);
+    }
+  }
+}

core/src/test/java/org/apache/calcite/test/RelOptRulesTest.java

@@ -10048,4 +10048,35 @@ private void checkLoptOptimizeJoinRule(LoptOptimizeJoinRule rule) {
     sql(sql).withRule(CoreRules.INTERSECT_TO_EXISTS)
         .checkUnchanged();
   }
+
+  /** Test case for <a href="https://issues.apache.org/jira/browse/CALCITE-6891">
+   * [CALCITE-6891] Implement IntersectReorderRule</a>. */
+  @Test void testIntersectReorderRule() {
+    final String sql = "select deptno from emp\n"
+        + "intersect\n"
+        + "select deptno from dept\n";
+
+    sql(sql).withVolcanoPlanner(true, p -> {
+      p.addRule(CoreRules.INTERSECT_REORDER);
+      p.addRule(EnumerableRules.ENUMERABLE_TABLE_SCAN_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_INTERSECT_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_FILTER_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_PROJECT_RULE);
+    }).check();
+  }
+
+  /** Test case for <a href="https://issues.apache.org/jira/browse/CALCITE-6891">
+   * [CALCITE-6891] Implement IntersectReorderRule</a>. */
+  @Test void testIntersectReorderRuleSameRowCount() {
+    final String sql = "select deptno from emp where deptno > 10\n"
+        + "intersect\n"
+        + "select deptno from emp where deptno > 5\n";
+    sql(sql).withVolcanoPlanner(true, p -> {
+      p.addRule(CoreRules.INTERSECT_REORDER);
+      p.addRule(EnumerableRules.ENUMERABLE_TABLE_SCAN_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_INTERSECT_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_FILTER_RULE);
+      p.addRule(EnumerableRules.ENUMERABLE_PROJECT_RULE);
+    }).check();
+  }
 }

core/src/test/resources/org/apache/calcite/test/RelOptRulesTest.xml

@@ -5978,6 +5978,62 @@ LogicalProject(EMPNO=[$0], ENAME=[$1], T=[$10])
   LogicalProject(EMPNO=[$0], ENAME=[$1], JOB=[$2], MGR=[$3], HIREDATE=[$4], SAL=[$5], COMM=[$6], DEPTNO=[$7], SLACKER=[$8], $f9=[7934], $f10=[CURRENT_TIMESTAMP])
     LogicalFilter(condition=[=($0, 7934)])
       LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testIntersectReorderRule">
+    <Resource name="sql">
+      <![CDATA[select deptno from emp
+intersect
+select deptno from dept
+]]>
+    </Resource>
+    <Resource name="planBefore">
+      <![CDATA[
+LogicalIntersect(all=[false])
+  LogicalProject(DEPTNO=[$7])
+    LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+  LogicalProject(DEPTNO=[$0])
+    LogicalTableScan(table=[[CATALOG, SALES, DEPT]])
+]]>
+    </Resource>
+    <Resource name="planAfter">
+      <![CDATA[
+EnumerableIntersect(all=[false])
+  EnumerableProject(DEPTNO=[$0])
+    EnumerableTableScan(table=[[CATALOG, SALES, DEPT]])
+  EnumerableProject(DEPTNO=[$7])
+    EnumerableTableScan(table=[[CATALOG, SALES, EMP]])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testIntersectReorderRuleSameRowCount">
+    <Resource name="sql">
+      <![CDATA[select deptno from emp where deptno > 10
+intersect
+select deptno from emp where deptno > 5
+]]>
+    </Resource>
+    <Resource name="planBefore">
+      <![CDATA[
+LogicalIntersect(all=[false])
+  LogicalProject(DEPTNO=[$7])
+    LogicalFilter(condition=[>($7, 10)])
+      LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+  LogicalProject(DEPTNO=[$7])
+    LogicalFilter(condition=[>($7, 5)])
+      LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+]]>
+    </Resource>
+    <Resource name="planAfter">
+      <![CDATA[
+EnumerableIntersect(all=[false])
+  EnumerableProject(DEPTNO=[$7])
+    EnumerableFilter(condition=[>($7, 10)])
+      EnumerableTableScan(table=[[CATALOG, SALES, EMP]])
+  EnumerableProject(DEPTNO=[$7])
+    EnumerableFilter(condition=[>($7, 5)])
+      EnumerableTableScan(table=[[CATALOG, SALES, EMP]])
 ]]>
     </Resource>
   </TestCase>