You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2020/05/14 18:03:22 UTC
[ignite] 01/01: Add filter pushdown rule for join.
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch ignite-12915
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit 56133ce8bf7c1a99cb44905c1204e4141e5eb5f6
Author: Andrey V. Mashenkov <an...@gmail.com>
AuthorDate: Thu May 14 21:02:24 2020 +0300
Add filter pushdown rule for join.
---
.../query/calcite/prepare/PlannerPhase.java | 5 +
.../query/calcite/rule/FilterJoinRule.java | 75 +++++++++++++++
.../calcite/rule/JoinPushExpressionsRule.java | 35 +++++++
.../processors/query/calcite/PlannerTest.java | 106 +++++++++++++++++++++
4 files changed, 221 insertions(+)
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
index 50be506..79600a4 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
@@ -25,8 +25,10 @@ import org.apache.calcite.tools.RuleSets;
import org.apache.ignite.internal.processors.query.calcite.rule.AggregateConverterRule;
import org.apache.ignite.internal.processors.query.calcite.rule.AggregateTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.FilterConverterRule;
+import org.apache.ignite.internal.processors.query.calcite.rule.FilterJoinRule;
import org.apache.ignite.internal.processors.query.calcite.rule.FilterTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.JoinConverterRule;
+import org.apache.ignite.internal.processors.query.calcite.rule.JoinPushExpressionsRule;
import org.apache.ignite.internal.processors.query.calcite.rule.JoinTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.ProjectConverterRule;
import org.apache.ignite.internal.processors.query.calcite.rule.ProjectTraitsPropagationRule;
@@ -69,6 +71,9 @@ public enum PlannerPhase {
AggregateConverterRule.INSTANCE,
AggregateTraitsPropagationRule.INSTANCE,
JoinConverterRule.INSTANCE,
+ JoinPushExpressionsRule.INSTANCE,
+ FilterJoinRule.FILTER_ON_JOIN,
+// FilterJoinRule.JOIN,
JoinTraitsPropagationRule.INSTANCE,
ProjectConverterRule.INSTANCE,
ProjectTraitsPropagationRule.INSTANCE,
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterJoinRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterJoinRule.java
new file mode 100644
index 0000000..b6440f7
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterJoinRule.java
@@ -0,0 +1,75 @@
+/*
+ * 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.processors.query.calcite.rule;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalJoin;
+
+public class FilterJoinRule {
+ /** Rule that pushes predicates from a Filter into the Join below them. */
+ public static final FilterIntoJoinRule FILTER_ON_JOIN = new FilterIntoJoinRule();
+
+ /** Rule that pushes predicates in a Join into the inputs to the join. */
+ public static final JoinConditionPushRule JOIN = new JoinConditionPushRule();
+
+ /**
+ * Rule that tries to push filter expressions into a join condition and into the inputs of the join.
+ */
+ public static class FilterIntoJoinRule extends org.apache.calcite.rel.rules.FilterJoinRule {
+ /**
+ * Creates a converter.
+ */
+ public FilterIntoJoinRule() {
+ super(
+ operand(LogicalFilter.class,
+ operand(LogicalJoin.class, RelOptRule.any())),
+ "FilterJoinRule:filter", true, RelFactories.LOGICAL_BUILDER,
+ org.apache.calcite.rel.rules.FilterJoinRule.TRUE_PREDICATE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMatch(RelOptRuleCall call) {
+ Filter filter = call.rel(0);
+ Join join = call.rel(1);
+ perform(call, filter, join);
+ }
+ }
+
+ /** Rule that pushes parts of the join condition to its inputs. */
+ public static class JoinConditionPushRule extends org.apache.calcite.rel.rules.FilterJoinRule {
+ /**
+ * Creates a converter.
+ */
+ public JoinConditionPushRule() {
+ super(RelOptRule.operand(LogicalJoin.class, RelOptRule.any()),
+ "FilterJoinRule:no-filter", true, RelFactories.LOGICAL_BUILDER,
+ org.apache.calcite.rel.rules.FilterJoinRule.TRUE_PREDICATE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMatch(RelOptRuleCall call) {
+ Join join = call.rel(0);
+ perform(call, null, join);
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinPushExpressionsRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinPushExpressionsRule.java
new file mode 100644
index 0000000..01e0438
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinPushExpressionsRule.java
@@ -0,0 +1,35 @@
+/*
+ * 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.processors.query.calcite.rule;
+
+import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.rel.logical.LogicalJoin;
+
+/**
+ * Planner rule that pushes down expressions in "equal" join condition.
+ *
+ * <p>For example, given
+ * "emp JOIN dept ON emp.deptno + 1 = dept.deptno", adds a project above "emp" that computes the expression "emp.deptno
+ * + 1". The resulting join condition is a simple combination of AND, equals, and input fields, plus the remaining
+ * non-equal conditions.
+ */
+public class JoinPushExpressionsRule {
+ /** */
+ public static final org.apache.calcite.rel.rules.JoinPushExpressionsRule INSTANCE =
+ new org.apache.calcite.rel.rules.JoinPushExpressionsRule(LogicalJoin.class, RelFactories.LOGICAL_BUILDER);
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
index 382a5db..894dd43 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
@@ -29,6 +29,7 @@ import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.ConventionTraitDef;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.RelTraitDef;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelCollation;
@@ -2323,6 +2324,111 @@ public class PlannerTest extends GridCommonAbstractTest {
assertNotNull(nodes);
}
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testJoinPushExpressionRule() throws Exception {
+ IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
+
+ TestTable emp = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("DEPTNO", f.createJavaType(Integer.class))
+ .build()) {
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.broadcast();
+ }
+ };
+
+ TestTable dept = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("DEPTNO", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .build()) {
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.broadcast();
+ }
+ };
+
+ IgniteSchema publicSchema = new IgniteSchema("PUBLIC");
+
+ publicSchema.addTable("EMP", emp);
+ publicSchema.addTable("DEPT", dept);
+
+ SchemaPlus schema = createRootSchema(false)
+ .add("PUBLIC", publicSchema);
+
+ String sql = "select d.deptno, e.deptno " +
+ "from dept d, emp e " +
+ "where d.deptno + 10 = e.deptno * 2";
+
+ RelTraitDef<?>[] traitDefs = {
+ DistributionTraitDef.INSTANCE,
+ ConventionTraitDef.INSTANCE
+ };
+
+ PlanningContext ctx = PlanningContext.builder()
+ .localNodeId(F.first(nodes))
+ .originatingNodeId(F.first(nodes))
+ .parentContext(Contexts.empty())
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(schema)
+ .traitDefs(traitDefs)
+ .build())
+ .logger(log)
+ .query(sql)
+ .topologyVersion(AffinityTopologyVersion.NONE)
+ .build();
+
+ RelRoot relRoot;
+
+ try (IgnitePlanner planner = ctx.planner()) {
+ assertNotNull(planner);
+
+ String query = ctx.query();
+
+ assertNotNull(query);
+
+ // Parse
+ SqlNode sqlNode = planner.parse(query);
+
+ // Validate
+ sqlNode = planner.validate(sqlNode);
+
+ // Convert to Relational operators graph
+ relRoot = planner.rel(sqlNode);
+
+ RelNode rel = relRoot.rel;
+
+ assertNotNull(rel);
+ assertEquals("LogicalProject(DEPTNO=[$0], DEPTNO0=[$4])\n" +
+ " LogicalFilter(condition=[=(+($0, 10), *($4, 2))])\n" +
+ " LogicalJoin(condition=[true], joinType=[inner])\n" +
+ " IgniteTableScan(table=[[PUBLIC, DEPT]])\n" +
+ " IgniteTableScan(table=[[PUBLIC, EMP]])\n",
+ RelOptUtil.toString(rel));
+
+ // Transformation chain
+ RelTraitSet desired = rel.getCluster().traitSet()
+ .replace(IgniteConvention.INSTANCE)
+ .replace(IgniteDistributions.single())
+ .simplify();
+
+ RelNode phys = planner.transform(PlannerPhase.OPTIMIZATION, desired, rel);
+
+ assertNotNull(phys);
+ assertEquals("IgniteProject(DEPTNO=[$0], DEPTNO0=[$4])\n" +
+ " IgniteJoin(condition=[=(+($0, 10), *($4, 2))], joinType=[inner])\n" +
+ " IgniteTableScan(table=[[PUBLIC, DEPT]])\n" +
+ " IgniteTableScan(table=[[PUBLIC, EMP]])\n",
+ RelOptUtil.toString(phys));
+ }
+ }
+
/** */
private NodesMapping intermediateMapping(@NotNull AffinityTopologyVersion topVer, int desiredCount, @Nullable Predicate<ClusterNode> filter) {
List<UUID> nodes = desiredCount == 1 ? select(this.nodes, 0) : select(this.nodes, 0, 1, 2, 3);