You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by jo...@apache.org on 2025/04/16 10:49:08 UTC
(ignite-3) branch main updated: IGNITE-24795 Sql. Add PROJECT_CORRELATE rule to HEP push down list (#5437)
This is an automated email from the ASF dual-hosted git repository.
jooger pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new ac9dfee4ef0 IGNITE-24795 Sql. Add PROJECT_CORRELATE rule to HEP push down list (#5437)
ac9dfee4ef0 is described below
commit ac9dfee4ef06620253bd81c7548ad21667aa1ffe
Author: ygerzhedovich <41...@users.noreply.github.com>
AuthorDate: Wed Apr 16 13:49:03 2025 +0300
IGNITE-24795 Sql. Add PROJECT_CORRELATE rule to HEP push down list (#5437)
---
.../internal/sql/engine/prepare/PlannerPhase.java | 2 +
.../IgniteProjectCorrelateTransposeRule.java | 259 +++++++++++++++++++++
.../planner/CorrelatedSubqueryPlannerTest.java | 4 +-
.../sql/engine/planner/PartitionPruningTest.java | 2 +-
.../planner/PredicatePushDownPlannerTest.java | 8 +-
.../ProjectCorrelateTransposePlannerTest.java | 86 +++++++
.../sql/engine/planner/TpchQueryPlannerTest.java | 2 +-
.../src/test/resources/mapping/correlated.test | 98 +++++---
.../resources/mapping/test_partition_pruning.test | 54 +++--
.../src/test/resources/tpch/plan/q21.plan | 133 +++++++++++
10 files changed, 585 insertions(+), 63 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlannerPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlannerPhase.java
index c955e02e776..96da45f0e48 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlannerPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlannerPhase.java
@@ -73,6 +73,7 @@ import org.apache.ignite.internal.sql.engine.rule.logical.ExposeIndexRule;
import org.apache.ignite.internal.sql.engine.rule.logical.FilterScanMergeRule;
import org.apache.ignite.internal.sql.engine.rule.logical.IgniteJoinConditionPushRule;
import org.apache.ignite.internal.sql.engine.rule.logical.IgniteMultiJoinOptimizeBushyRule;
+import org.apache.ignite.internal.sql.engine.rule.logical.IgniteProjectCorrelateTransposeRule;
import org.apache.ignite.internal.sql.engine.rule.logical.LogicalOrToUnionRule;
import org.apache.ignite.internal.sql.engine.rule.logical.ProjectScanMergeRule;
import org.apache.ignite.internal.sql.engine.util.Commons;
@@ -131,6 +132,7 @@ public enum PlannerPhase {
"Heuristic phase to push down and merge projects",
ProjectScanMergeRule.TABLE_SCAN_SKIP_CORRELATED,
ProjectScanMergeRule.SYSTEM_VIEW_SCAN_SKIP_CORRELATED,
+ IgniteProjectCorrelateTransposeRule.INSTANCE,
CoreRules.JOIN_PUSH_EXPRESSIONS,
CoreRules.PROJECT_MERGE,
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/IgniteProjectCorrelateTransposeRule.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/IgniteProjectCorrelateTransposeRule.java
new file mode 100644
index 00000000000..0e732c52241
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/logical/IgniteProjectCorrelateTransposeRule.java
@@ -0,0 +1,259 @@
+/*
+ * 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.ignite.internal.sql.engine.rule.logical;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttleImpl;
+import org.apache.calcite.rel.core.Correlate;
+import org.apache.calcite.rel.core.CorrelationId;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.logical.LogicalTableFunctionScan;
+import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.rules.ProjectCorrelateTransposeRule;
+import org.apache.calcite.rel.rules.PushProjector;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexCorrelVariable;
+import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexOver;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.util.BitSets;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.internal.sql.engine.rule.logical.IgniteProjectCorrelateTransposeRule.IgniteProjectCorrelateTransposeRuleConfig;
+import org.immutables.value.Value;
+
+/**
+ * Planner rule that pushes a {@link Project} under {@link Correlate} to apply on Correlate's left and right inputs.
+ *
+ * <p>There are two differences between this rule and {@link ProjectCorrelateTransposeRule}:
+ * 1. Take into account RexCall parameters for functions in LogicalTableFunctionScan like a SYSTEM_RANGE.
+ * See RelNodesExprsHandler#visit(RelNode)
+ * 2. Fixed an issue for trying amend correlation for incorrect case. See RexFieldAccessReplacer#visitFieldAccess(RexFieldAccess)
+ *
+ * @see CoreRules#PROJECT_CORRELATE_TRANSPOSE
+ */
+@Value.Enclosing
+public class IgniteProjectCorrelateTransposeRule
+ extends RelRule<IgniteProjectCorrelateTransposeRuleConfig>
+ implements TransformationRule {
+ public static final RelOptRule INSTANCE = IgniteProjectCorrelateTransposeRuleConfig.DEFAULT.toRule();
+
+
+ /** Creates a ProjectCorrelateTransposeRule. */
+ protected IgniteProjectCorrelateTransposeRule(IgniteProjectCorrelateTransposeRuleConfig config) {
+ super(config);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ Project origProject = call.rel(0);
+ Correlate correlate = call.rel(1);
+
+ // locate all fields referenced in the projection
+ // determine which inputs are referenced in the projection;
+ // if all fields are being referenced and there are no
+ // special expressions, no point in proceeding any further
+ PushProjector pushProjector =
+ new PushProjector(origProject, call.builder().literal(true), correlate,
+ config.preserveExprCondition(), call.builder());
+ if (pushProjector.locateAllRefs()) {
+ return;
+ }
+
+ // create left and right projections, projecting only those
+ // fields referenced on each side
+ RelNode leftProject =
+ pushProjector.createProjectRefsAndExprs(
+ correlate.getLeft(),
+ true,
+ false);
+ RelNode rightProject =
+ pushProjector.createProjectRefsAndExprs(
+ correlate.getRight(),
+ true,
+ true);
+
+ Map<Integer, Integer> requiredColsMap = new HashMap<>();
+
+ // adjust requiredColumns that reference the projected columns
+ int[] adjustments = pushProjector.getAdjustments();
+ BitSet updatedBits = new BitSet();
+ for (Integer col : correlate.getRequiredColumns()) {
+ int newCol = col + adjustments[col];
+ updatedBits.set(newCol);
+ requiredColsMap.put(col, newCol);
+ }
+
+ RexBuilder rexBuilder = call.builder().getRexBuilder();
+
+ CorrelationId correlationId = correlate.getCluster().createCorrel();
+ RexCorrelVariable rexCorrel =
+ (RexCorrelVariable) rexBuilder.makeCorrel(
+ leftProject.getRowType(),
+ correlationId);
+
+ // updates RexCorrelVariable and sets actual RelDataType for RexFieldAccess
+ rightProject =
+ rightProject.accept(
+ new IgniteProjectCorrelateTransposeRule.RelNodesExprsHandler(
+ new IgniteProjectCorrelateTransposeRule.RexFieldAccessReplacer(correlate.getCorrelationId(),
+ rexCorrel, rexBuilder, requiredColsMap)));
+
+ // create a new correlate with the projected children
+ Correlate newCorrelate =
+ correlate.copy(
+ correlate.getTraitSet(),
+ leftProject,
+ rightProject,
+ correlationId,
+ ImmutableBitSet.of(BitSets.toIter(updatedBits)),
+ correlate.getJoinType());
+
+ // put the original project on top of the correlate, converting it to
+ // reference the modified projection list
+ RelNode topProject =
+ pushProjector.createNewProject(newCorrelate, adjustments);
+
+ call.transformTo(topProject);
+ }
+
+ /**
+ * Visitor for RexNodes which replaces {@link RexCorrelVariable} with specified.
+ */
+ public static class RexFieldAccessReplacer extends RexShuttle {
+ private final RexBuilder builder;
+ private final CorrelationId rexCorrelVariableToReplace;
+ private final RexCorrelVariable rexCorrelVariable;
+ private final Map<Integer, Integer> requiredColsMap;
+
+ /**
+ * Constructor.
+ */
+ public RexFieldAccessReplacer(
+ CorrelationId rexCorrelVariableToReplace,
+ RexCorrelVariable rexCorrelVariable,
+ RexBuilder builder,
+ Map<Integer, Integer> requiredColsMap) {
+ this.rexCorrelVariableToReplace = rexCorrelVariableToReplace;
+ this.rexCorrelVariable = rexCorrelVariable;
+ this.builder = builder;
+ this.requiredColsMap = requiredColsMap;
+ }
+
+ @Override
+ public RexNode visitCorrelVariable(RexCorrelVariable variable) {
+ if (variable.id.equals(rexCorrelVariableToReplace)) {
+ return rexCorrelVariable;
+ }
+ return variable;
+ }
+
+ @Override
+ public RexNode visitCall(RexCall call) {
+ // Over expression should not be pushed.
+ return super.visitCall(call);
+ }
+
+ @Override
+ public RexNode visitFieldAccess(RexFieldAccess fieldAccess) {
+ RexNode refExpr = fieldAccess.getReferenceExpr().accept(this);
+ // creates new RexFieldAccess instance for the case when referenceExpr was replaced.
+ // Otherwise calls super method.
+ if (fieldAccess.getReferenceExpr() != refExpr) {
+ if (refExpr == rexCorrelVariable) {
+ int fieldIndex = fieldAccess.getField().getIndex();
+ return builder.makeFieldAccess(
+ refExpr,
+ requireNonNull(requiredColsMap.get(fieldIndex),
+ () -> "no entry for field " + fieldIndex + " in " + requiredColsMap));
+ }
+ }
+ return super.visitFieldAccess(fieldAccess);
+ }
+ }
+
+ /**
+ * Visitor for RelNodes which applies specified {@link RexShuttle} visitor for every node in the tree.
+ */
+ public static class RelNodesExprsHandler extends RelShuttleImpl {
+ private final RexShuttle rexVisitor;
+
+ public RelNodesExprsHandler(RexShuttle rexVisitor) {
+ this.rexVisitor = rexVisitor;
+ }
+
+ @Override
+ protected RelNode visitChild(RelNode parent, int i,
+ RelNode input) {
+ return super.visitChild(parent, i, input.stripped())
+ .accept(rexVisitor);
+ }
+
+ @Override
+ public RelNode visit(RelNode other) {
+ if (other instanceof LogicalTableFunctionScan) {
+ List<RexNode> operands = ((RexCall) ((LogicalTableFunctionScan) other).getCall()).getOperands();
+ for (int i = 0; i < operands.size(); i++) {
+ operands.get(i).accept(rexVisitor);
+ }
+ }
+ return super.visit(other).accept(rexVisitor);
+ }
+ }
+
+ /** Rule configuration. */
+ @Value.Immutable(singleton = false)
+ public interface IgniteProjectCorrelateTransposeRuleConfig extends RelRule.Config {
+ IgniteProjectCorrelateTransposeRuleConfig DEFAULT = ImmutableIgniteProjectCorrelateTransposeRule
+ .IgniteProjectCorrelateTransposeRuleConfig.builder()
+ .withPreserveExprCondition(expr -> !(expr instanceof RexOver))
+ .build()
+ .withOperandFor(Project.class, Correlate.class);
+
+ @Override
+ default IgniteProjectCorrelateTransposeRule toRule() {
+ return new IgniteProjectCorrelateTransposeRule(this);
+ }
+
+ /** Defines when an expression should not be pushed. */
+ PushProjector.ExprCondition preserveExprCondition();
+
+ /** Sets {@link #preserveExprCondition()}. */
+ IgniteProjectCorrelateTransposeRuleConfig withPreserveExprCondition(PushProjector.ExprCondition condition);
+
+ /** Defines an operand tree for the given classes. */
+ default IgniteProjectCorrelateTransposeRuleConfig withOperandFor(Class<? extends Project> projectClass,
+ Class<? extends Correlate> correlateClass) {
+ return withOperandSupplier(b0 ->
+ b0.operand(projectClass).oneInput(b1 ->
+ b1.operand(correlateClass).anyInputs()))
+ .as(IgniteProjectCorrelateTransposeRuleConfig.class);
+ }
+ }
+}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/CorrelatedSubqueryPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/CorrelatedSubqueryPlannerTest.java
index 446d2c1f142..d88754f4c7f 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/CorrelatedSubqueryPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/CorrelatedSubqueryPlannerTest.java
@@ -68,7 +68,9 @@ public class CorrelatedSubqueryPlannerTest extends AbstractPlannerTest {
+ " OR c>d\n"
+ " ORDER BY 1;";
- IgniteRel rel = physicalPlan(sql, schema, "FilterTableScanMergeRule");
+ // ToDo: The rules %ScanMerge%Rule disabled for the test due to we have a bug https://issues.apache.org/jira/browse/IGNITE-25001
+ IgniteRel rel = physicalPlan(sql, schema, "FilterTableScanMergeRule", "ProjectTableScanMergeRule",
+ "ProjectTableScanMergeSkipCorrelatedRule");
IgniteFilter filter = findFirstNode(rel, byClass(IgniteFilter.class)
.and(f -> RexUtils.hasCorrelation(((Filter) f).getCondition())));
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PartitionPruningTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PartitionPruningTest.java
index 4b63c72cc64..4f757e3815f 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PartitionPruningTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PartitionPruningTest.java
@@ -233,7 +233,7 @@ public class PartitionPruningTest extends AbstractPlannerTest {
PartitionPruningColumns cols = actual.get(2);
assertNotNull(cols, "No metadata for source=2");
- assertEquals("[[0=$cor0.C1], [0=42]]", PartitionPruningColumns.canonicalForm(cols).toString());
+ assertEquals("[[0=$cor1.C1], [0=42]]", PartitionPruningColumns.canonicalForm(cols).toString());
}
private PartitionPruningMetadata extractMetadata(String query, IgniteTable... table) throws Exception {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PredicatePushDownPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PredicatePushDownPlannerTest.java
index 28eafcacb63..847f6f0110d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PredicatePushDownPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PredicatePushDownPlannerTest.java
@@ -47,10 +47,10 @@ public class PredicatePushDownPlannerTest extends AbstractPlannerTest {
+ " WHERE ot.c1 = it.c1"
+ " AND it.c2 > it.c3)";
- assertPlan(sql, schema, hasChildThat(isInstanceOf(Join.class)
- .and(input(0, isInstanceOf(ProjectableFilterableTableScan.class)
- .and(scan -> scan.condition().toString().contains(">($t1, 10)")))))
- );
+ assertPlan(sql, schema, isInstanceOf(Join.class)
+ .and(hasChildThat(isInstanceOf(ProjectableFilterableTableScan.class)
+ .and(scan -> scan.condition().toString().contains(">($t1, 10)")))));
+
}
private static IgniteTable createTable(String tableName) {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/ProjectCorrelateTransposePlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/ProjectCorrelateTransposePlannerTest.java
new file mode 100644
index 00000000000..ac2dc276dc7
--- /dev/null
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/ProjectCorrelateTransposePlannerTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.ignite.internal.sql.engine.planner;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Objects;
+import java.util.function.Predicate;
+import java.util.function.UnaryOperator;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.CorrelationId;
+import org.apache.calcite.rex.RexCorrelVariable;
+import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.ignite.internal.sql.engine.framework.TestBuilders.TableBuilder;
+import org.apache.ignite.internal.sql.engine.rel.IgniteCorrelatedNestedLoopJoin;
+import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
+import org.apache.ignite.internal.sql.engine.rel.IgniteTableFunctionScan;
+import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
+import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests IgniteProjectCorrelateTransposeRule.
+ */
+public class ProjectCorrelateTransposePlannerTest extends AbstractPlannerTest {
+ @Test
+ public void testProjectCorrelateTranspose() throws Exception {
+ IgniteSchema publicSchema = createSchemaFrom(
+ tableA("T0"),
+ tableA("T1"));
+
+ String sql = "select t0.id "
+ + "from t0 "
+ + "where exists (select * from t1 where t0.jid = t1.jid);";
+
+ Predicate<RelNode> check =
+ hasChildThat(isInstanceOf(IgniteCorrelatedNestedLoopJoin.class)
+ .and(input(0, isTableScan("T0").and(n -> n.requiredColumns() != null))));
+
+ assertPlan(sql, publicSchema, check);
+ }
+
+ /** Test works of system_range function with correlate. */
+ @Test
+ public void testSystemRangeFunctionWithCorrelate() throws Exception {
+ IgniteSchema schema = createSchemaFrom(tableA("T0"));
+ String sql = "SELECT t.jid FROM t0 t WHERE t.jid < 5 AND EXISTS "
+ + "(SELECT x FROM table(system_range(t.jid, t.jid)) WHERE mod(x, 2) = 0)";
+
+ IgniteRel rel = physicalPlan(sql, schema);
+
+ IgniteCorrelatedNestedLoopJoin join = findFirstNode(rel, byClass(IgniteCorrelatedNestedLoopJoin.class));
+ IgniteTableFunctionScan tableFunction = findFirstNode(rel, byClass(IgniteTableFunctionScan.class));
+
+ RexFieldAccess firstArgOfSystemRange = Objects.requireNonNull(findFirst(tableFunction.getCall(), RexFieldAccess.class));
+ CorrelationId correlationOnTableFunction = ((RexCorrelVariable) firstArgOfSystemRange.getReferenceExpr()).id;
+
+ assertTrue(join.getVariablesSet().contains(correlationOnTableFunction),
+ "Correlation " + correlationOnTableFunction + " should be in variables set " + join.getVariablesSet());
+ }
+
+ private static UnaryOperator<TableBuilder> tableA(String tableName) {
+ return tableBuilder -> tableBuilder
+ .name(tableName)
+ .addColumn("ID", NativeTypes.INT32)
+ .addColumn("JID", NativeTypes.INT32)
+ .addColumn("VAL", NativeTypes.STRING)
+ .distribution(IgniteDistributions.broadcast());
+ }
+}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/TpchQueryPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/TpchQueryPlannerTest.java
index 06c0e856db8..02a7173240d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/TpchQueryPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/TpchQueryPlannerTest.java
@@ -44,7 +44,7 @@ public class TpchQueryPlannerTest extends AbstractTpcQueryPlannerTest {
// HashJoin(condition=[... OR(=($10, _UTF-8'GERMANY'), =($14, _UTF-8'GERMANY')), OR(=($14, _UTF-8'FRANCE'), =($10, _UTF-8'FRANCE')))]
// vs
// HashJoin(condition=[... OR(=($14, _UTF-8'FRANCE'), =($10, _UTF-8'FRANCE')), OR(=($10, _UTF-8'GERMANY'), =($14, _UTF-8'GERMANY')))]
- @ValueSource(strings = {"1", "5", "7", "8", "9"})
+ @ValueSource(strings = {"1", "5", "7", "8", "9", "21"})
public void test(String queryId) {
validateQueryPlan(queryId);
}
diff --git a/modules/sql-engine/src/test/resources/mapping/correlated.test b/modules/sql-engine/src/test/resources/mapping/correlated.test
index cb42502015c..fd3b0185f96 100644
--- a/modules/sql-engine/src/test/resources/mapping/correlated.test
+++ b/modules/sql-engine/src/test/resources/mapping/correlated.test
@@ -165,10 +165,11 @@ Fragment#3
Project
CorrelatedNestedLoopJoin
TableScan(name=PUBLIC.CT_N1, source=1, partitions=1, distribution=single)
- ColocatedHashAggregate
- Project
- Filter
- TableFunctionScan(source=2, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Project
+ Filter
+ TableFunctionScan(source=2, distribution=single)
---
N0
@@ -183,10 +184,11 @@ Fragment#0 root
Project
CorrelatedNestedLoopJoin
Receiver(sourceFragment=1, exchange=1, distribution=single)
- ColocatedHashAggregate
- Project
- Filter
- TableFunctionScan(source=2, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Project
+ Filter
+ TableFunctionScan(source=2, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -195,7 +197,7 @@ Fragment#1
partitions: {N1=[0:1]}
tree:
Sender(targetFragment=0, exchange=1, distribution=single)
- TableScan(name=PUBLIC.T_N1, source=3, partitions=1, distribution=affinity[table: T_N1, columns: [ID]])
+ TableScan(name=PUBLIC.T_N1, source=3, partitions=1, distribution=random)
---
# Pass partition pruning metadata for correlated joins.
N0
@@ -204,13 +206,11 @@ SELECT * FROM t1_n0n1n2 as cor WHERE EXISTS (SELECT 1 FROM t2_n0n1n2 as t2 WHERE
Fragment#0 root
executionNodes: [N0]
remoteFragments: [1, 2]
- exchangeSourceNodes: {1=[N0, N1, N2], 2=[N0, N1, N2]}
+ exchangeSourceNodes: {1=[N0, N1, N2], 2=[N0]}
tree:
- Project
- CorrelatedNestedLoopJoin
- Receiver(sourceFragment=1, exchange=1, distribution=single)
- ColocatedHashAggregate
- Receiver(sourceFragment=2, exchange=2, distribution=single)
+ CorrelatedNestedLoopJoin
+ Receiver(sourceFragment=1, exchange=1, distribution=single)
+ Receiver(sourceFragment=2, exchange=2, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -219,17 +219,28 @@ Fragment#1
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
tree:
Sender(targetFragment=0, exchange=1, distribution=single)
- TableScan(name=PUBLIC.T1_N0N1N2, source=4, partitions=3, distribution=affinity[table: T1_N0N1N2, columns: [ID]])
+ TableScan(name=PUBLIC.T1_N0N1N2, source=5, partitions=3, distribution=affinity[table: T1_N0N1N2, columns: [ID]])
Fragment#2 correlated
+ targetNodes: [N0]
+ executionNodes: [N0]
+ remoteFragments: [3]
+ exchangeSourceNodes: {3=[N0, N1, N2]}
+ tree:
+ Sender(targetFragment=0, exchange=2, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Receiver(sourceFragment=3, exchange=3, distribution=hash[0])
+
+Fragment#3 correlated
targetNodes: [N0]
executionNodes: [N0, N1, N2]
tables: [T2_N0N1N2]
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
- pruningMetadata: [3=[{0=$cor1.ID}]]
+ pruningMetadata: [4=[{0=$cor3.ID}]]
tree:
- Sender(targetFragment=0, exchange=2, distribution=single)
- TableScan(name=PUBLIC.T2_N0N1N2, source=3, partitions=3, distribution=random)
+ Sender(targetFragment=2, exchange=3, distribution=hash[0])
+ TableScan(name=PUBLIC.T2_N0N1N2, source=4, partitions=3, distribution=random)
---
# Pass partition pruning metadata to correlated joins.
N0
@@ -239,17 +250,16 @@ WHERE EXISTS (SELECT * FROM t1_n0n1n2 as cor WHERE out.id = cor.id AND EXISTS (S
Fragment#0 root
executionNodes: [N0]
remoteFragments: [1, 2, 3]
- exchangeSourceNodes: {1=[N0, N1, N2], 2=[N0, N1, N2], 3=[N0, N1, N2]}
+ exchangeSourceNodes: {1=[N0, N1, N2], 2=[N0, N1, N2], 3=[N0]}
tree:
- Project
- CorrelatedNestedLoopJoin
- Receiver(sourceFragment=1, exchange=1, distribution=single)
+ CorrelatedNestedLoopJoin
+ Receiver(sourceFragment=1, exchange=1, distribution=single)
+ Project
ColocatedHashAggregate
Project
CorrelatedNestedLoopJoin
Receiver(sourceFragment=2, exchange=2, distribution=single)
- ColocatedHashAggregate
- Receiver(sourceFragment=3, exchange=3, distribution=single)
+ Receiver(sourceFragment=3, exchange=3, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -258,27 +268,38 @@ Fragment#1
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
tree:
Sender(targetFragment=0, exchange=1, distribution=single)
- TableScan(name=PUBLIC.T3_N0N1N2, source=6, partitions=3, distribution=affinity[table: T3_N0N1N2, columns: [ID]])
+ TableScan(name=PUBLIC.T3_N0N1N2, source=7, partitions=3, distribution=affinity[table: T3_N0N1N2, columns: [ID]])
Fragment#2 correlated
targetNodes: [N0]
executionNodes: [N0, N1, N2]
tables: [T1_N0N1N2]
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
- pruningMetadata: [5=[{0=$cor2.ID}]]
+ pruningMetadata: [6=[{0=$cor6.ID}]]
tree:
Sender(targetFragment=0, exchange=2, distribution=single)
- TableScan(name=PUBLIC.T1_N0N1N2, source=5, partitions=3, distribution=affinity[table: T1_N0N1N2, columns: [ID]])
+ TableScan(name=PUBLIC.T1_N0N1N2, source=6, partitions=3, distribution=affinity[table: T1_N0N1N2, columns: [ID]])
Fragment#3 correlated
+ targetNodes: [N0]
+ executionNodes: [N0]
+ remoteFragments: [4]
+ exchangeSourceNodes: {4=[N0, N1, N2]}
+ tree:
+ Sender(targetFragment=0, exchange=3, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Receiver(sourceFragment=4, exchange=4, distribution=hash[0])
+
+Fragment#4 correlated
targetNodes: [N0]
executionNodes: [N0, N1, N2]
tables: [T2_N0N1N2]
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
- pruningMetadata: [4=[{0=$cor3.ID}]]
+ pruningMetadata: [5=[{0=$cor7.ID}]]
tree:
- Sender(targetFragment=0, exchange=3, distribution=single)
- TableScan(name=PUBLIC.T2_N0N1N2, source=4, partitions=3, distribution=random)
+ Sender(targetFragment=3, exchange=4, distribution=hash[0])
+ TableScan(name=PUBLIC.T2_N0N1N2, source=5, partitions=3, distribution=random)
---
# Pass partition pruning metadata to correlated joins one layer deep.
N0
@@ -293,15 +314,16 @@ Fragment#0 root
remoteFragments: [1, 2, 3]
exchangeSourceNodes: {1=[N0, N1, N2], 2=[N0, N1, N2], 3=[N0, N1, N2]}
tree:
- Project
- CorrelatedNestedLoopJoin
- Receiver(sourceFragment=1, exchange=1, distribution=single)
+ CorrelatedNestedLoopJoin
+ Receiver(sourceFragment=1, exchange=1, distribution=single)
+ Project
ColocatedHashAggregate
Project
CorrelatedNestedLoopJoin
Receiver(sourceFragment=2, exchange=2, distribution=single)
- ReduceHashAggregate
- Receiver(sourceFragment=3, exchange=3, distribution=single)
+ Project
+ ReduceHashAggregate
+ Receiver(sourceFragment=3, exchange=3, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -317,7 +339,7 @@ Fragment#2 correlated
executionNodes: [N0, N1, N2]
tables: [T1_N0N1N2]
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
- pruningMetadata: [5=[{0=$cor3.ID}]]
+ pruningMetadata: [5=[{0=$cor8.ID}]]
tree:
Sender(targetFragment=0, exchange=2, distribution=single)
TableScan(name=PUBLIC.T1_N0N1N2, source=5, partitions=3, distribution=affinity[table: T1_N0N1N2, columns: [ID]])
@@ -327,7 +349,7 @@ Fragment#3 correlated
executionNodes: [N0, N1, N2]
tables: [T2_N0N1N2]
partitions: {N0=[0:3], N1=[1:3], N2=[2:3]}
- pruningMetadata: [4=[{0=$cor3.ID}, {0=$cor5.ID}]]
+ pruningMetadata: [4=[{0=$cor8.ID}, {0=$cor9.ID}]]
tree:
Sender(targetFragment=0, exchange=3, distribution=single)
MapHashAggregate
diff --git a/modules/sql-engine/src/test/resources/mapping/test_partition_pruning.test b/modules/sql-engine/src/test/resources/mapping/test_partition_pruning.test
index 8760c44ea29..81a75f2171f 100644
--- a/modules/sql-engine/src/test/resources/mapping/test_partition_pruning.test
+++ b/modules/sql-engine/src/test/resources/mapping/test_partition_pruning.test
@@ -121,11 +121,9 @@ Fragment#0 root
remoteFragments: [1, 2]
exchangeSourceNodes: {1=[N3], 2=[N1, N2, N3]}
tree:
- Project
- CorrelatedNestedLoopJoin
- Receiver(sourceFragment=1, exchange=1, distribution=single)
- ColocatedHashAggregate
- Receiver(sourceFragment=2, exchange=2, distribution=single)
+ CorrelatedNestedLoopJoin
+ Receiver(sourceFragment=1, exchange=1, distribution=single)
+ Receiver(sourceFragment=2, exchange=2, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -134,17 +132,28 @@ Fragment#1
partitions: {N3=[2:3]}
tree:
Sender(targetFragment=0, exchange=1, distribution=single)
- TableScan(name=PUBLIC.T1_N1N2N3, source=4, partitions=3, distribution=affinity[table: T1_N1N2N3, columns: [ID]])
+ TableScan(name=PUBLIC.T1_N1N2N3, source=5, partitions=3, distribution=affinity[table: T1_N1N2N3, columns: [ID]])
Fragment#2 correlated
targetNodes: [N0]
executionNodes: [N1, N2, N3]
+ remoteFragments: [3]
+ exchangeSourceNodes: {3=[N1, N2, N3]}
+ tree:
+ Sender(targetFragment=0, exchange=2, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Receiver(sourceFragment=3, exchange=3, distribution=hash[0])
+
+Fragment#3 correlated
+ targetNodes: [N1, N2, N3]
+ executionNodes: [N1, N2, N3]
tables: [T3_N1N2N3]
partitions: {N1=[0:3], N2=[1:3], N3=[2:3]}
- pruningMetadata: [3=[{0=$cor1.ID}]]
+ pruningMetadata: [4=[{0=$cor3.ID}]]
tree:
- Sender(targetFragment=0, exchange=2, distribution=single)
- TableScan(name=PUBLIC.T3_N1N2N3, source=3, partitions=3, distribution=random)
+ Sender(targetFragment=2, exchange=3, distribution=hash[0])
+ TableScan(name=PUBLIC.T3_N1N2N3, source=4, partitions=3, distribution=random)
---
# Correlated.
# Prune partitions from left arm statically, and pass meta to the right arm.
@@ -157,11 +166,9 @@ Fragment#0 root
remoteFragments: [1, 2]
exchangeSourceNodes: {1=[N3], 2=[N4, N5]}
tree:
- Project
- CorrelatedNestedLoopJoin
- Receiver(sourceFragment=1, exchange=1, distribution=single)
- ColocatedHashAggregate
- Receiver(sourceFragment=2, exchange=2, distribution=single)
+ CorrelatedNestedLoopJoin
+ Receiver(sourceFragment=1, exchange=1, distribution=single)
+ Receiver(sourceFragment=2, exchange=2, distribution=single)
Fragment#1
targetNodes: [N0]
@@ -170,15 +177,26 @@ Fragment#1
partitions: {N3=[2:3]}
tree:
Sender(targetFragment=0, exchange=1, distribution=single)
- TableScan(name=PUBLIC.T1_N1N2N3, source=4, partitions=3, distribution=affinity[table: T1_N1N2N3, columns: [ID]])
+ TableScan(name=PUBLIC.T1_N1N2N3, source=5, partitions=3, distribution=affinity[table: T1_N1N2N3, columns: [ID]])
Fragment#2 correlated
targetNodes: [N0]
executionNodes: [N4, N5]
+ remoteFragments: [3]
+ exchangeSourceNodes: {3=[N4, N5]}
+ tree:
+ Sender(targetFragment=0, exchange=2, distribution=single)
+ Project
+ ColocatedHashAggregate
+ Receiver(sourceFragment=3, exchange=3, distribution=hash[0])
+
+Fragment#3 correlated
+ targetNodes: [N4, N5]
+ executionNodes: [N4, N5]
tables: [T2_N4N5]
partitions: {N4=[0:2], N5=[1:2]}
- pruningMetadata: [3=[{0=$cor1.ID}]]
+ pruningMetadata: [4=[{0=$cor3.ID}]]
tree:
- Sender(targetFragment=0, exchange=2, distribution=single)
- TableScan(name=PUBLIC.T2_N4N5, source=3, partitions=2, distribution=random)
+ Sender(targetFragment=2, exchange=3, distribution=hash[0])
+ TableScan(name=PUBLIC.T2_N4N5, source=4, partitions=2, distribution=random)
---
diff --git a/modules/sql-engine/src/test/resources/tpch/plan/q21.plan b/modules/sql-engine/src/test/resources/tpch/plan/q21.plan
new file mode 100644
index 00000000000..1525b0dfb21
--- /dev/null
+++ b/modules/sql-engine/src/test/resources/tpch/plan/q21.plan
@@ -0,0 +1,133 @@
+Limit
+ fetch: 100
+ est. row count: 100
+ Sort
+ collation: [NUMWAIT DESC, S_NAME ASC]
+ fetch: 100
+ est. row count: 100
+ ColocatedSortAggregate
+ group: [S_NAME]
+ aggs: [COUNT()]
+ collation: [S_NAME ASC]
+ est. row count: 332203730726418
+ Project
+ fields: [S_NAME]
+ exprs: [S_NAME]
+ est. row count: 1661018653632089
+ Filter
+ condition: IS NULL(i)
+ est. row count: 1661018653632089
+ CorrelatedNestedLoopJoin
+ condition: true
+ joinType: left
+ variablesSet: [$cor5]
+ est. row count: 6644074614528357
+ CorrelatedNestedLoopJoin
+ condition: true
+ joinType: inner
+ variablesSet: [$cor6]
+ est. row count: 66493788407
+ Sort
+ collation: [S_NAME ASC]
+ est. row count: 332734
+ Project
+ fields: [S_NAME, L_ORDERKEY, L_SUPPKEY]
+ exprs: [S_NAME, L_ORDERKEY, L_SUPPKEY]
+ est. row count: 332734
+ HashJoin
+ condition: =(O_ORDERKEY, L_ORDERKEY)
+ joinType: inner
+ est. row count: 332734
+ HashJoin
+ condition: =(S_SUPPKEY, L_SUPPKEY)
+ joinType: inner
+ est. row count: 999202
+ Exchange
+ distribution: single
+ est. row count: 3000608
+ TableScan
+ table: [PUBLIC, LINEITEM]
+ filters: >(L_RECEIPTDATE, L_COMMITDATE)
+ fields: [L_ORDERKEY, L_SUPPKEY, L_COMMITDATE, L_RECEIPTDATE]
+ est. row count: 3000608
+ MergeJoin
+ condition: =(S_NATIONKEY, N_NATIONKEY)
+ joinType: inner
+ leftCollation: [S_NATIONKEY ASC]
+ rightCollation: [S_SUPPKEY ASC]
+ est. row count: 3330
+ Exchange
+ distribution: single
+ est. row count: 10000
+ IndexScan
+ table: [PUBLIC, SUPPLIER]
+ index: S_NK
+ type: SORTED
+ fields: [S_SUPPKEY, S_NAME, S_NATIONKEY]
+ collation: [S_NATIONKEY ASC]
+ est. row count: 10000
+ Exchange
+ distribution: single
+ est. row count: 8
+ Sort
+ collation: [N_NATIONKEY ASC]
+ est. row count: 8
+ TableScan
+ table: [PUBLIC, NATION]
+ filters: =(N_NAME, _UTF-8'SAUDI ARABIA')
+ fields: [N_NATIONKEY, N_NAME]
+ est. row count: 8
+ Exchange
+ distribution: single
+ est. row count: 499500
+ TableScan
+ table: [PUBLIC, ORDERS]
+ filters: =(O_ORDERSTATUS, _UTF-8'F')
+ fields: [O_ORDERKEY, O_ORDERSTATUS]
+ est. row count: 499500
+ Project
+ fields: []
+ exprs: []
+ est. row count: 199840
+ ReduceHashAggregate
+ rowType: RecordType(BOOLEAN i)
+ group: [$f0]
+ est. row count: 199840
+ Exchange
+ distribution: single
+ est. row count: 199840
+ MapHashAggregate
+ group: [$f0]
+ aggs: []
+ est. row count: 199840
+ IndexScan
+ table: [PUBLIC, LINEITEM]
+ index: L_OK
+ type: SORTED
+ searchBounds: [ExactBounds [bound=$cor6.L_ORDERKEY]]
+ filters: AND(=(L_ORDERKEY, $cor6.L_ORDERKEY), <>(L_SUPPKEY, $cor6.L_SUPPKEY))
+ fields: [$f0]
+ projects: [true]
+ collation: []
+ est. row count: 999202
+ ReduceHashAggregate
+ rowType: RecordType(BOOLEAN i)
+ group: [$f0]
+ est. row count: 99920
+ Exchange
+ distribution: single
+ est. row count: 99920
+ MapHashAggregate
+ group: [$f0]
+ aggs: []
+ est. row count: 99920
+ IndexScan
+ table: [PUBLIC, LINEITEM]
+ index: L_OK
+ type: SORTED
+ searchBounds: [ExactBounds [bound=$cor5.L_ORDERKEY]]
+ filters: AND(=(L_ORDERKEY, $cor5.L_ORDERKEY), <>(L_SUPPKEY, $cor5.L_SUPPKEY), >(L_RECEIPTDATE, L_COMMITDATE))
+ fields: [$f0]
+ projects: [true]
+ collation: []
+ est. row count: 499601