You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by gv...@apache.org on 2019/10/04 19:58:28 UTC
[ignite] branch ignite-12248 updated: IGNITE-12248: Apache Calcite
based query execution engine.
This is an automated email from the ASF dual-hosted git repository.
gvvinblade pushed a commit to branch ignite-12248
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/ignite-12248 by this push:
new 6bee086 IGNITE-12248: Apache Calcite based query execution engine.
6bee086 is described below
commit 6bee0863fa35b51a0ef486ebeeaa9cfef97839f1
Author: Igor Seliverstov <gv...@gmail.com>
AuthorDate: Fri Oct 4 22:58:14 2019 +0300
IGNITE-12248: Apache Calcite based query execution engine.
exchanges.
---
.../apache/calcite/plan/volcano/VolcanoUtil.java | 28 +++++
.../calcite/metadata/IgniteMdDistribution.java | 122 +++++++++++++++++++++
.../query/calcite/metadata/IgniteMetadata.java | 53 +++++++++
.../query/calcite/prepare/IgnitePlanner.java | 8 +-
.../calcite/rel/logical/IgniteLogicalExchange.java | 65 +++++++++++
.../processors/query/calcite/rule/IgniteRules.java | 4 +-
.../calcite/rule/logical/IgniteFilterRule.java | 32 ++++--
.../query/calcite/rule/logical/IgniteJoinRule.java | 31 +++++-
.../calcite/rule/logical/IgniteProjectRule.java | 53 +++++++--
.../query/calcite/schema/IgniteTable.java | 8 +-
.../query/calcite/trait/IgniteDistribution.java | 87 +++++++++++++++
.../calcite/trait/IgniteDistributionImpl.java | 78 +++++++++++++
.../calcite/trait/IgniteDistributionTraitDef.java | 69 ++++++++++++
.../query/calcite/trait/IgniteDistributions.java | 44 ++++++++
.../processors/query/calcite/util/Commons.java | 7 ++
.../query/calcite/CalciteQueryProcessorTest.java | 16 ++-
16 files changed, 664 insertions(+), 41 deletions(-)
diff --git a/modules/calcite/src/main/java/org/apache/calcite/plan/volcano/VolcanoUtil.java b/modules/calcite/src/main/java/org/apache/calcite/plan/volcano/VolcanoUtil.java
new file mode 100644
index 0000000..8d3030d
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/calcite/plan/volcano/VolcanoUtil.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.plan.volcano;
+
+import org.apache.calcite.rel.RelNode;
+
+/**
+ *
+ */
+public class VolcanoUtil {
+ public static void fireRules(VolcanoPlanner planner, RelNode rel, boolean deferred) {
+ planner.fireRules(rel, deferred);
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java
new file mode 100644
index 0000000..838b37a
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.metadata;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.metadata.MetadataDef;
+import org.apache.calcite.rel.metadata.MetadataHandler;
+import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexSlot;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+
+import static org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution.DistributionType.HASH;
+
+/**
+ *
+ */
+public class IgniteMdDistribution implements MetadataHandler<IgniteMetadata.Distribution> {
+ public static final RelMetadataProvider PROVIDER =
+ ReflectiveRelMetadataProvider.reflectiveSource(IgniteMetadata.Distribution.METHOD, new IgniteMdDistribution());
+
+ @Override public MetadataDef<IgniteMetadata.Distribution> getDef() {
+ return IgniteMetadata.Distribution.DEF;
+ }
+
+ public IgniteDistribution getDistribution(RelNode rel, RelMetadataQuery mq) {
+ return rel.getTraitSet().getTrait(IgniteDistributionTraitDef.INSTANCE);
+ }
+
+ public IgniteDistribution getDistribution(Filter filter, RelMetadataQuery mq) {
+ return filter(mq, filter.getInput(), filter.getCondition());
+ }
+
+ public IgniteDistribution getDistribution(Project project, RelMetadataQuery mq) {
+ return project(mq, project.getInput(), project.getProjects());
+ }
+
+ public IgniteDistribution getDistribution(Join join, RelMetadataQuery mq) {
+ return join(mq, join.getLeft(), join.getRight(), join.getCondition());
+ }
+
+ public static IgniteDistribution project(RelMetadataQuery mq, RelNode input, List<RexNode> projects) {
+ IgniteDistribution trait = distribution_(input, mq);
+
+ if (trait.type() == HASH) {
+ ImmutableIntList keys = trait.keys();
+
+ if (keys.size() > projects.size())
+ return IgniteDistributions.random(trait.sources());
+
+ Map<Integer, Integer> m = new HashMap<>(projects.size());
+
+ for (Ord<RexNode> node : Ord.zip(projects)) {
+ if (node.e instanceof RexInputRef)
+ m.put( ((RexSlot) node.e).getIndex(), node.i);
+ else if (node.e.isA(SqlKind.CAST)) {
+ RexNode operand = ((RexCall) node.e).getOperands().get(0);
+
+ if (operand instanceof RexInputRef)
+ m.put(((RexSlot) operand).getIndex(), node.i);
+ }
+ }
+
+ List<Integer> newKeys = new ArrayList<>(keys.size());
+
+ for (Integer key : keys) {
+ Integer mapped = m.get(key);
+
+ if (mapped == null)
+ return IgniteDistributions.random(trait.sources());
+
+ newKeys.add(mapped);
+ }
+
+ return IgniteDistributions.hash(newKeys, trait.sources());
+ }
+
+ return trait;
+ }
+
+ public static IgniteDistribution filter(RelMetadataQuery mq, RelNode input, RexNode condition) {
+ return distribution_(input, mq);
+ }
+
+ public static IgniteDistribution join(RelMetadataQuery mq, RelNode left, RelNode right, RexNode condition) {
+ return distribution_(left, mq);
+ }
+
+ private static IgniteDistribution distribution_(RelNode rel, RelMetadataQuery mq) {
+ return rel.metadata(IgniteMetadata.Distribution.class, mq).getDistribution();
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMetadata.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMetadata.java
new file mode 100644
index 0000000..155b9ce
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMetadata.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.metadata;
+
+import com.google.common.collect.ImmutableList;
+import java.lang.reflect.Method;
+import org.apache.calcite.linq4j.tree.Types;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.metadata.ChainedRelMetadataProvider;
+import org.apache.calcite.rel.metadata.DefaultRelMetadataProvider;
+import org.apache.calcite.rel.metadata.Metadata;
+import org.apache.calcite.rel.metadata.MetadataDef;
+import org.apache.calcite.rel.metadata.MetadataHandler;
+import org.apache.calcite.rel.metadata.RelMetadataProvider;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+
+/**
+ *
+ */
+public class IgniteMetadata {
+ public static final RelMetadataProvider METADATA_PROVIDER = ChainedRelMetadataProvider.of(ImmutableList.of(
+ DefaultRelMetadataProvider.INSTANCE,
+ IgniteMdDistribution.PROVIDER
+ ));
+
+ public interface Distribution extends Metadata {
+ Method METHOD = Types.lookupMethod(Distribution.class, "getDistribution");
+ MetadataDef<Distribution> DEF = MetadataDef.of(Distribution.class, Distribution.Handler.class, METHOD);
+
+ /** Determines how the rows are distributed. */
+ IgniteDistribution getDistribution();
+
+ /** Handler API. */
+ interface Handler extends MetadataHandler<Distribution> {
+ IgniteDistribution getDistribution(RelNode r, RelMetadataQuery mq);
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgnitePlanner.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgnitePlanner.java
index b9bb2ed..f568e63 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgnitePlanner.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgnitePlanner.java
@@ -45,7 +45,7 @@ import org.apache.calcite.rel.RelShuttleImpl;
import org.apache.calcite.rel.core.TableFunctionScan;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.logical.LogicalValues;
-import org.apache.calcite.rel.metadata.DefaultRelMetadataProvider;
+import org.apache.calcite.rel.metadata.CachingRelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataProvider;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeSystem;
@@ -69,6 +69,7 @@ import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RuleSet;
import org.apache.calcite.tools.ValidationException;
import org.apache.calcite.util.Pair;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMetadata;
import org.apache.ignite.internal.processors.query.calcite.rule.PlannerPhase;
import org.apache.ignite.internal.processors.query.calcite.rule.PlannerType;
@@ -88,11 +89,11 @@ public class IgnitePlanner implements Planner, RelOptTable.ViewExpander {
private final RexExecutor executor;
private final SchemaPlus defaultSchema;
private final JavaTypeFactory typeFactory;
- private final RelMetadataProvider metadataProvider;
private boolean open;
private RelOptPlanner planner;
+ private RelMetadataProvider metadataProvider;
private SqlValidator validator;
/**
@@ -110,7 +111,6 @@ public class IgnitePlanner implements Planner, RelOptTable.ViewExpander {
executor = config.getExecutor();
context = config.getContext();
connectionConfig = connConfig();
- metadataProvider = DefaultRelMetadataProvider.INSTANCE; // TODO: right costs
RelDataTypeSystem typeSystem = connectionConfig
.typeSystem(RelDataTypeSystem.class, RelDataTypeSystem.DEFAULT);
@@ -144,6 +144,7 @@ public class IgnitePlanner implements Planner, RelOptTable.ViewExpander {
/** {@inheritDoc} */
@Override public void reset() {
planner = null;
+ metadataProvider = null;
validator = null;
open = false;
@@ -153,6 +154,7 @@ public class IgnitePlanner implements Planner, RelOptTable.ViewExpander {
if (!open) {
planner = new VolcanoPlanner(frameworkConfig.getCostFactory(), context);
planner.setExecutor(executor);
+ metadataProvider = new CachingRelMetadataProvider(IgniteMetadata.METADATA_PROVIDER, planner);
validator = new IgniteSqlValidator(operatorTable, createCatalogReader(), typeFactory, conformance());
validator.setIdentifierExpansion(true);
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalExchange.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalExchange.java
new file mode 100644
index 0000000..03c4682
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/logical/IgniteLogicalExchange.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.rel.logical;
+
+import java.util.List;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.util.Util;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
+
+/**
+ *
+ */
+public final class IgniteLogicalExchange extends SingleRel implements IgniteRel {
+ /**
+ * Creates a <code>SingleRel</code>.
+ *
+ * @param cluster Cluster this relational expression belongs to
+ * @param traits Node traits.
+ * @param input Input relational expression
+ */
+ public IgniteLogicalExchange(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
+ super(cluster, traits, input);
+ }
+
+ @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ double rowCount = mq.getRowCount(this);
+ double bytesPerRow = getRowType().getFieldCount() * 4;
+ return planner.getCostFactory().makeCost(
+ Util.nLogN(rowCount) * bytesPerRow, rowCount, 0);
+ }
+
+ public IgniteDistribution sourceDistribution() {
+ return getInput().getTraitSet().getTrait(IgniteDistributionTraitDef.INSTANCE);
+ }
+
+ public IgniteDistribution targetDistribution() {
+ return getTraitSet().getTrait(IgniteDistributionTraitDef.INSTANCE);
+ }
+
+ @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ return new IgniteLogicalExchange(getCluster(), traitSet, sole(inputs));
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/IgniteRules.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/IgniteRules.java
index 477b747..d8e50ab 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/IgniteRules.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/IgniteRules.java
@@ -166,8 +166,8 @@ public class IgniteRules {
public static List<RelOptRule> logicalRules(Context ctx) {
return ImmutableList.<RelOptRule>builder()
- .addAll(BASE_RULES)
- .addAll(ABSTRACT_RULES)
+// .addAll(BASE_RULES)
+// .addAll(ABSTRACT_RULES)
.addAll(ABSTRACT_RELATIONAL_RULES)
.addAll(IGNITE_BASE_RULES)
.build();
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteFilterRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteFilterRule.java
index bdda595..32f99fe 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteFilterRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteFilterRule.java
@@ -17,19 +17,22 @@
package org.apache.ignite.internal.processors.query.calcite.rule.logical;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.calcite.plan.Convention;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.RelFactories;
import org.apache.calcite.rel.logical.LogicalFilter;
-import org.apache.calcite.rel.metadata.RelMdCollation;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalFilter;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
/**
*
@@ -46,11 +49,24 @@ public class IgniteFilterRule extends RelOptRule {
final RelNode input = filter.getInput();
final RelOptCluster cluster = input.getCluster();
final RelMetadataQuery mq = cluster.getMetadataQuery();
- final RelTraitSet traitSet = cluster.traitSet()
- .replace(IgniteRel.LOGICAL_CONVENTION)
- .replaceIfs(RelCollationTraitDef.INSTANCE,
- () -> RelMdCollation.filter(mq, input));
- RelNode convertedInput = convert(input, traitSet);
- call.transformTo(new IgniteLogicalFilter(cluster, traitSet, convertedInput, filter.getCondition(), filter.getVariablesSet()));
+
+ Set<RelNode> transformed = new HashSet<>();
+
+ RelSubset subset = (RelSubset) input;
+
+ for (RelNode relNode : subset.getRelList()) {
+ if (!(relNode instanceof IgniteRel))
+ continue;
+
+ final RelTraitSet traitSet = cluster.traitSet()
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteMdDistribution.filter(mq, relNode, filter.getCondition()));
+
+ RelNode converted = convert(relNode, traitSet);
+
+ if (transformed.add(converted))
+ call.transformTo(new IgniteLogicalFilter(cluster, traitSet, converted, filter.getCondition(), filter.getVariablesSet()));
+ }
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteJoinRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteJoinRule.java
index 44009a8..ff203d3 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteJoinRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteJoinRule.java
@@ -25,8 +25,13 @@ import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.RelFactories;
import org.apache.calcite.rel.logical.LogicalJoin;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalJoin;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
/**
*
@@ -41,10 +46,30 @@ public class IgniteJoinRule extends RelOptRule {
@Override public void onMatch(RelOptRuleCall call) {
LogicalJoin join = call.rel(0);
RelOptCluster cluster = join.getCluster();
+
+ IgniteDistribution leftDist = join.getLeft().getTraitSet().getTrait(IgniteDistributionTraitDef.INSTANCE);
+ IgniteDistribution rightDist = join.getRight().getTraitSet().getTrait(IgniteDistributionTraitDef.INSTANCE);
+
+ RelTraitSet leftTraits = join.getLeft().getTraitSet()
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteDistributions.hash(join.analyzeCondition().leftKeys, leftDist.sources()));
+
+ RelTraitSet rightTraits = cluster.traitSet()
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteDistributions.hash(join.analyzeCondition().rightKeys, rightDist.sources()));
+
+ RelNode left = convert(join.getLeft(), leftTraits);
+ RelNode right = convert(join.getRight(), rightTraits);
+
+ RelMetadataQuery mq = cluster.getMetadataQuery();
+
RelTraitSet traitSet = cluster.traitSet()
- .replace(IgniteRel.LOGICAL_CONVENTION);
- RelNode left = convert(join.getLeft(), traitSet);
- RelNode right = convert(join.getRight(), traitSet);
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteMdDistribution.join(mq, left, right, join.getCondition()));
+
call.transformTo(new IgniteLogicalJoin(cluster, traitSet, left, right,
join.getCondition(), join.getVariablesSet(), join.getJoinType(), join.isSemiJoinDone()));
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteProjectRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteProjectRule.java
index 043085f..f1582ba 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteProjectRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/logical/IgniteProjectRule.java
@@ -17,21 +17,24 @@
package org.apache.ignite.internal.processors.query.calcite.rule.logical;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.calcite.plan.Convention;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.RelFactories;
import org.apache.calcite.rel.logical.LogicalProject;
-import org.apache.calcite.rel.metadata.RelMdCollation;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rex.RexNode;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalProject;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
/**
*
@@ -47,16 +50,42 @@ public class IgniteProjectRule extends RelOptRule {
LogicalProject project = call.rel(0);
final RelNode input = project.getInput();
final RelOptCluster cluster = input.getCluster();
- final List<? extends RexNode> projects = project.getProjects();
+ final List<RexNode> projects = project.getProjects();
final RelMetadataQuery mq = cluster.getMetadataQuery();
- final RelTraitSet traitSet =
- cluster.traitSet()
- .replace(IgniteRel.LOGICAL_CONVENTION)
-// .replaceIf(RelDistributionTraitDef.INSTANCE,
-// () -> RelMdDistribution.project(mq, input, projects))
- .replaceIfs(RelCollationTraitDef.INSTANCE,
- () -> RelMdCollation.project(mq, input, projects));
- RelNode convertedInput = convert(input, traitSet);
- call.transformTo(new IgniteLogicalProject(cluster, traitSet, convertedInput, projects, project.getRowType()));
+
+ boolean done = false;
+
+ Set<RelNode> transformed = new HashSet<>();
+
+ for (RelNode relNode : ((RelSubset) input).getRelList()) {
+ if (!(relNode instanceof IgniteRel))
+ continue;
+
+ final RelTraitSet traitSet =
+ cluster.traitSet()
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteMdDistribution.project(mq, relNode, projects));
+
+ RelNode converted = convert(relNode, traitSet);
+
+ if (transformed.add(converted)) {
+ call.transformTo(new IgniteLogicalProject(cluster, traitSet, converted, projects, project.getRowType()));
+
+ done = true;
+ }
+ }
+
+ if (!done) {
+ final RelTraitSet traitSet =
+ cluster.traitSet()
+ .replace(IgniteRel.LOGICAL_CONVENTION)
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE,
+ () -> IgniteMdDistribution.project(mq, input, projects));
+
+ RelNode converted = convert(input, traitSet);
+
+ call.transformTo(new IgniteLogicalProject(cluster, converted.getTraitSet(), converted, projects, project.getRowType()));
+ }
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/IgniteTable.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/IgniteTable.java
index 0bd4e51..ef1606f 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/IgniteTable.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/IgniteTable.java
@@ -21,8 +21,6 @@ import java.util.function.Function;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelDistributionTraitDef;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
@@ -30,8 +28,11 @@ import org.apache.calcite.schema.Statistic;
import org.apache.calcite.schema.Statistics;
import org.apache.calcite.schema.TranslatableTable;
import org.apache.calcite.schema.impl.AbstractTable;
+import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalTableScan;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.jetbrains.annotations.Nullable;
/** */
@@ -80,8 +81,7 @@ public class IgniteTable extends AbstractTable implements TranslatableTable {
RelOptCluster cluster = context.getCluster();
RelTraitSet traitSet = cluster.traitSet()
.replace(IgniteRel.LOGICAL_CONVENTION)
- .replaceIf(RelDistributionTraitDef.INSTANCE, () -> getStatistic().getDistribution())
- .replaceIfs(RelCollationTraitDef.INSTANCE, () -> getStatistic().getCollations());
+ .replaceIf(IgniteDistributionTraitDef.INSTANCE, () -> IgniteDistributions.hash(ImmutableIntList.of(0), ImmutableIntList.of()));
return new IgniteLogicalTableScan(cluster, traitSet, relOptTable);
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistribution.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistribution.java
new file mode 100644
index 0000000..f80fb2f
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistribution.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.trait;
+
+import org.apache.calcite.plan.RelTrait;
+import org.apache.calcite.plan.RelTraitDef;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Util;
+
+/**
+ *
+ */
+public interface IgniteDistribution extends RelTrait {
+ public enum DistributionType {
+ HASH("hash"),
+ RANDOM("random"),
+ BROADCAST("broadcast"),
+ SINGLE("single"),
+ ANY("any");
+
+ /** */
+ private final String description;
+
+ /** */
+ DistributionType(String description) {
+ this.description = description;
+ }
+
+ /** */
+ @Override public String toString() {
+ return description;
+ }
+ }
+
+ IgniteDistribution ANY = new IgniteDistributionImpl(DistributionType.ANY, ImmutableIntList.of());
+
+ DistributionType type();
+
+ @Override default RelTraitDef getTraitDef() {
+ return IgniteDistributionTraitDef.INSTANCE;
+ }
+
+ @Override default boolean satisfies(RelTrait trait) {
+ if (trait == this)
+ return true;
+
+ if (!(trait instanceof IgniteDistribution))
+ return false;
+
+ IgniteDistribution other = (IgniteDistribution) trait;
+
+ if (!Util.startsWith(other.sources(), sources()))
+ return false;
+
+ if (other.type() == DistributionType.ANY)
+ return true;
+
+ if (type() == other.type())
+ return type() != DistributionType.HASH || keys().equals(other.keys());
+
+ return other.type() == DistributionType.RANDOM && type() == DistributionType.HASH;
+ }
+
+ /**
+ * @return Hash distribution columns ordinals or empty list otherwise.
+ */
+ ImmutableIntList keys();
+
+ /**
+ * @return Sources (indexes of sorted by node order nodes list for particular topology).
+ */
+ ImmutableIntList sources();
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionImpl.java
new file mode 100644
index 0000000..fdc43a1
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionImpl.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.trait;
+
+import java.util.Objects;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.util.ImmutableIntList;
+
+/**
+ *
+ */
+public class IgniteDistributionImpl implements IgniteDistribution {
+ private final DistributionType type;
+ private final ImmutableIntList keys;
+
+ private ImmutableIntList sources;
+
+ public IgniteDistributionImpl(DistributionType type, ImmutableIntList keys) {
+ this(type, keys, ImmutableIntList.of());
+ }
+
+ public IgniteDistributionImpl(DistributionType type, ImmutableIntList keys, ImmutableIntList sources) {
+ this.type = type;
+ this.keys = keys;
+ this.sources = sources;
+ }
+
+ @Override public DistributionType type() {
+ return type;
+ }
+
+ @Override public ImmutableIntList keys() {
+ return keys;
+ }
+
+ @Override public ImmutableIntList sources() {
+ return sources;
+ }
+
+ @Override public void register(RelOptPlanner planner) {}
+
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o instanceof IgniteDistribution) {
+ IgniteDistribution that = (IgniteDistribution) o;
+
+ return type == that.type()
+ && keys.equals(that.keys())
+ && sources.equals(that.sources());
+ }
+
+ return false;
+ }
+
+ @Override public int hashCode() {
+ return Objects.hash(type, keys, sources);
+ }
+
+ @Override public String toString() {
+ return type + (type == DistributionType.HASH ? keys.toString() : "") + sources;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionTraitDef.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionTraitDef.java
new file mode 100644
index 0000000..85e1999
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributionTraitDef.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.trait;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelTraitDef;
+import org.apache.calcite.rel.RelNode;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
+import org.apache.ignite.internal.processors.query.calcite.rel.logical.IgniteLogicalExchange;
+
+/**
+ *
+ */
+public class IgniteDistributionTraitDef extends RelTraitDef<IgniteDistribution> {
+ /** */
+ public static final IgniteDistributionTraitDef INSTANCE = new IgniteDistributionTraitDef();
+
+ @Override public Class<IgniteDistribution> getTraitClass() {
+ return IgniteDistribution.class;
+ }
+
+ @Override public String getSimpleName() {
+ return "distr";
+ }
+
+ @Override public RelNode convert(RelOptPlanner planner, RelNode rel, IgniteDistribution targetDist, boolean allowInfiniteCostConverters) {
+ IgniteDistribution srcDist = rel.getTraitSet().getTrait(INSTANCE);
+
+ // Source and Target have the same trait.
+ if (srcDist.equals(targetDist))
+ return rel;
+
+ if (rel.getConvention() != IgniteRel.LOGICAL_CONVENTION)
+ return null;
+
+ switch(targetDist.type()){
+ case HASH:
+ case BROADCAST:
+ case SINGLE:
+ return new IgniteLogicalExchange(rel.getCluster(), rel.getTraitSet().replace(targetDist), rel);
+ case ANY:
+ return rel;
+ default:
+ return null;
+ }
+ }
+
+ @Override public boolean canConvert(RelOptPlanner planner, IgniteDistribution fromTrait, IgniteDistribution toTrait) {
+ return true;
+ }
+
+ @Override public IgniteDistribution getDefault() {
+ return IgniteDistribution.ANY;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java
new file mode 100644
index 0000000..91f2176
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019 GridGain Systems, Inc. and Contributors.
+ *
+ * Licensed under the GridGain Community Edition License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+ *
+ * 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.trait;
+
+import java.util.List;
+import org.apache.calcite.util.ImmutableIntList;
+
+import static org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution.DistributionType.HASH;
+import static org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution.DistributionType.RANDOM;
+import static org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution.DistributionType.SINGLE;
+
+/**
+ *
+ */
+public class IgniteDistributions {
+ /** */
+ private static final IgniteDistributionTraitDef traitDef = IgniteDistributionTraitDef.INSTANCE;
+
+ public static IgniteDistribution random(List<Integer> sources) {
+ return traitDef.canonize(new IgniteDistributionImpl(RANDOM, ImmutableIntList.of(), ImmutableIntList.copyOf(sources)));
+ }
+
+ public static IgniteDistribution hash(List<Integer> keys, List<Integer> sources) {
+ return traitDef.canonize(new IgniteDistributionImpl(HASH, ImmutableIntList.copyOf(keys), ImmutableIntList.copyOf(sources)));
+ }
+
+ public static IgniteDistribution single(List<Integer> sources) {
+ return traitDef.canonize(new IgniteDistributionImpl(SINGLE, ImmutableIntList.of(), ImmutableIntList.copyOf(sources)));
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
index 8758eb2..26ec6a0 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
@@ -22,6 +22,8 @@ import java.util.function.Function;
import java.util.function.Predicate;
import org.apache.calcite.plan.Context;
import org.apache.calcite.plan.Contexts;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
@@ -56,4 +58,9 @@ public final class Commons {
return builder.build();
};
}
+
+ public static boolean transformSubset(RelOptRuleCall call, RelTraitSet traits) {
+
+ return false;
+ }
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
index 76da7d3..25298e4 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
@@ -23,15 +23,13 @@ import org.apache.calcite.plan.Contexts;
import org.apache.calcite.plan.ConventionTraitDef;
import org.apache.calcite.plan.RelTraitDef;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelCollationTraitDef;
-import org.apache.calcite.rel.RelDistributionTraitDef;
-import org.apache.calcite.rel.RelDistributions;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.tools.Frameworks;
+import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.internal.processors.query.calcite.prepare.IgnitePlanner;
import org.apache.ignite.internal.processors.query.calcite.prepare.Query;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
@@ -39,6 +37,8 @@ import org.apache.ignite.internal.processors.query.calcite.rule.PlannerPhase;
import org.apache.ignite.internal.processors.query.calcite.rule.PlannerType;
import org.apache.ignite.internal.processors.query.calcite.schema.IgniteSchema;
import org.apache.ignite.internal.processors.query.calcite.schema.IgniteTable;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributionTraitDef;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -135,9 +135,8 @@ public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
assertNotNull(ctx);
RelTraitDef[] traitDefs = {
- RelDistributionTraitDef.INSTANCE,
- ConventionTraitDef.INSTANCE,
- RelCollationTraitDef.INSTANCE
+ IgniteDistributionTraitDef.INSTANCE,
+ ConventionTraitDef.INSTANCE
};
RelRoot relRoot;
@@ -163,10 +162,9 @@ public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
// Transformation chain
rel = planner.transform(PlannerType.HEP, PlannerPhase.SUBQUERY_REWRITE, rel, rel.getTraitSet());
- RelTraitSet desired = rel.getTraitSet()
- .replace(relRoot.collation)
+ RelTraitSet desired = rel.getCluster().traitSet()
.replace(IgniteRel.LOGICAL_CONVENTION)
- .replace(RelDistributions.ANY)
+ .replace(IgniteDistributions.single(ImmutableIntList.of()))
.simplify();
rel = planner.transform(PlannerType.VOLCANO, PlannerPhase.LOGICAL, rel, desired);