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 2020/04/16 21:10:17 UTC
[ignite] 02/02: IGNITE-12864: Calcite integration. UNION support.
This closes #7673
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
commit f23fadbaa08237c50bceb356af438e2fecafeb37
Author: Igor Seliverstov <gv...@gmail.com>
AuthorDate: Fri Apr 17 00:09:13 2020 +0300
IGNITE-12864: Calcite integration. UNION support. This closes #7673
---
.../query/calcite/exec/ExecutionServiceImpl.java | 1 +
.../query/calcite/exec/LogicalRelImplementor.java | 9 +
.../query/calcite/exec/PhysicalRelImplementor.java | 10 +
.../query/calcite/exec/rel/UnionAllNode.java | 98 +++++++
.../metadata/IgniteMdDerivedDistribution.java | 85 +++---
.../calcite/metadata/IgniteMdDistribution.java | 73 +----
.../calcite/metadata/IgniteMdFragmentInfo.java | 27 ++
.../processors/query/calcite/prepare/Cloner.java | 11 +-
.../query/calcite/prepare/FragmentSplitter.java | 319 ++++++++++++---------
.../query/calcite/prepare/PlannerPhase.java | 8 +-
.../processors/query/calcite/prepare/Splitter.java | 8 +-
.../query/calcite/rel/IgniteReceiver.java | 9 +-
.../query/calcite/rel/IgniteReduceAggregate.java | 5 +-
.../query/calcite/rel/IgniteRelVisitor.java | 5 +
.../processors/query/calcite/rel/IgniteSender.java | 14 +-
.../{IgniteReceiver.java => IgniteUnionAll.java} | 47 ++-
.../rule/AggregateTraitsPropagationRule.java | 6 +-
.../processors/query/calcite/rule/RuleUtils.java | 36 +--
.../query/calcite/rule/UnionConverterRule.java | 64 +++++
.../rule/UnionTraitsTraitsPropagationRule.java | 69 +++++
.../calcite/serialize/PhysicalRelVisitor.java | 5 +
.../calcite/serialize/RelToPhysicalConverter.java | 13 +
.../calcite/serialize/UnionAllPhysicalRel.java | 67 +++++
.../query/calcite/trait/DistributionTrait.java | 38 ++-
.../query/calcite/trait/IgniteDistributions.java | 313 +++++++-------------
.../processors/query/calcite/util/Commons.java | 27 ++
.../query/calcite/CalciteQueryProcessorTest.java | 95 ++++++
.../processors/query/calcite/PlannerTest.java | 282 ++++++++++++++++++
.../query/calcite/exec/rel/ExecutionTest.java | 44 +++
29 files changed, 1269 insertions(+), 519 deletions(-)
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
index ebd153f..00c39c4 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
@@ -502,6 +502,7 @@ public class ExecutionServiceImpl extends AbstractService implements ExecutionSe
switch (sqlNode.getKind()) {
case SELECT:
+ case UNION:
return prepareQuery(sqlNode, ctx);
case INSERT:
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
index c340d2e..8845b16 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
@@ -41,6 +41,7 @@ import org.apache.ignite.internal.processors.query.calcite.exec.rel.Node;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Outbox;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.UnionAllNode;
import org.apache.ignite.internal.processors.query.calcite.metadata.PartitionService;
import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
import org.apache.ignite.internal.processors.query.calcite.prepare.RelTarget;
@@ -58,6 +59,7 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
import org.apache.ignite.internal.processors.query.calcite.schema.TableDescriptor;
import org.apache.ignite.internal.processors.query.calcite.trait.Destination;
@@ -174,6 +176,13 @@ public class LogicalRelImplementor implements IgniteRelVisitor<Node<Object[]>> {
}
/** {@inheritDoc} */
+ @Override public Node<Object[]> visit(IgniteUnionAll rel) {
+ UnionAllNode<Object[]> node = new UnionAllNode<>(ctx);
+ node.register(Commons.transform(rel.getInputs(), this::visit));
+ return node;
+ }
+
+ /** {@inheritDoc} */
@Override public Node<Object[]> visit(IgniteTableModify rel) {
switch (rel.getOperation()){
case INSERT:
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
index d1c2282..e090b38 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.processors.query.calcite.exec.rel.Node;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Outbox;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.UnionAllNode;
import org.apache.ignite.internal.processors.query.calcite.metadata.PartitionService;
import org.apache.ignite.internal.processors.query.calcite.schema.TableDescriptor;
import org.apache.ignite.internal.processors.query.calcite.serialize.AggregatePhysicalRel;
@@ -51,10 +52,12 @@ import org.apache.ignite.internal.processors.query.calcite.serialize.SenderPhysi
import org.apache.ignite.internal.processors.query.calcite.serialize.TableModifyPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.TableScanPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.TrimExchangePhysicalRel;
+import org.apache.ignite.internal.processors.query.calcite.serialize.UnionAllPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.ValuesPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.trait.Destination;
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFunction;
import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
/**
@@ -185,6 +188,13 @@ public class PhysicalRelImplementor implements PhysicalRelVisitor<Node<Object[]>
}
/** {@inheritDoc} */
+ @Override public Node<Object[]> visit(UnionAllPhysicalRel rel) {
+ UnionAllNode<Object[]> node = new UnionAllNode<>(ctx);
+ node.register(Commons.transform(rel.inputs(), this::visit));
+ return node;
+ }
+
+ /** {@inheritDoc} */
@Override public Node<Object[]> visit(AggregatePhysicalRel rel) {
AggregateNode.AggregateType type;
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/UnionAllNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/UnionAllNode.java
new file mode 100644
index 0000000..759c868
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/UnionAllNode.java
@@ -0,0 +1,98 @@
+/*
+ * 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.exec.rel;
+
+import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ *
+ */
+public class UnionAllNode<T> extends AbstractNode<T> implements Downstream<T> {
+ /** */
+ private int curSrc;
+
+ /** */
+ private int waiting;
+
+ /**
+ * @param ctx Execution context.
+ */
+ public UnionAllNode(ExecutionContext ctx) {
+ super(ctx);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Downstream<T> requestDownstream(int idx) {
+ assert sources != null;
+ assert idx >= 0 && idx < sources.size();
+
+ return this;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void request(int rowsCount) {
+ checkThread();
+
+ assert !F.isEmpty(sources);
+ assert rowsCount > 0 && waiting == 0;
+
+ source().request(waiting = rowsCount);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void push(T row) {
+ checkThread();
+
+ assert downstream != null;
+ assert waiting > 0;
+
+ waiting--;
+
+ downstream.push(row);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void end() {
+ checkThread();
+
+ assert downstream != null;
+ assert waiting > 0;
+
+ if (++curSrc < sources.size())
+ source().request(waiting);
+ else {
+ waiting = -1;
+ downstream.end();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onError(Throwable e) {
+ checkThread();
+
+ assert downstream != null;
+
+ downstream.onError(e);
+ }
+
+ /** */
+ private Node<T> source() {
+ return sources.get(curSrc);
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
index 82ca615..46a7d8a 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.query.calcite.metadata;
+import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -32,19 +33,20 @@ import org.apache.calcite.rel.logical.LogicalAggregate;
import org.apache.calcite.rel.logical.LogicalFilter;
import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.logical.LogicalProject;
-import org.apache.calcite.rel.logical.LogicalTableScan;
-import org.apache.calcite.rel.logical.LogicalValues;
+import org.apache.calcite.rel.logical.LogicalUnion;
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.util.ImmutableBitSet;
import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMetadata.DerivedDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
-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.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.BiSuggestion;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.Suggestion;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.processors.query.calcite.util.IgniteMethod;
import org.apache.ignite.internal.util.typedef.F;
@@ -83,42 +85,6 @@ public class IgniteMdDerivedDistribution implements MetadataHandler<DerivedDistr
}
/**
- * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
- */
- public List<IgniteDistribution> deriveDistributions(IgniteRel rel, RelMetadataQuery mq) {
- return F.asList(IgniteMdDistribution._distribution(rel, mq));
- }
-
- /**
- * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
- */
- public List<IgniteDistribution> deriveDistributions(LogicalTableScan rel, RelMetadataQuery mq) {
- return F.asList(IgniteMdDistribution._distribution(rel, mq));
- }
-
- /**
- * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
- */
- public List<IgniteDistribution> deriveDistributions(LogicalValues rel, RelMetadataQuery mq) {
- return F.asList(IgniteMdDistribution._distribution(rel, mq));
- }
-
- /**
- * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
- */
- public List<IgniteDistribution> deriveDistributions(LogicalProject rel, RelMetadataQuery mq) {
- HashSet<IgniteDistribution> res = new HashSet<>();
-
- Mappings.TargetMapping mapping =
- Project.getPartialMapping(rel.getInput().getRowType().getFieldCount(), rel.getProjects());
-
- for (IgniteDistribution inDistr : _deriveDistributions(rel.getInput(), mq))
- res.add(inDistr.apply(mapping));
-
- return new ArrayList<>(res);
- }
-
- /**
* Here we trying to get physical nodes and request distribution types from them, in case there is no physical
* nodes, we get logical ones and derive possible distribution types they may satisfy with.
*
@@ -163,21 +129,54 @@ public class IgniteMdDerivedDistribution implements MetadataHandler<DerivedDistr
/**
* See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
*/
+ public List<IgniteDistribution> deriveDistributions(LogicalProject rel, RelMetadataQuery mq) {
+ HashSet<IgniteDistribution> res = new HashSet<>();
+
+ Mappings.TargetMapping mapping =
+ Project.getPartialMapping(rel.getInput().getRowType().getFieldCount(), rel.getProjects());
+
+ for (IgniteDistribution inDistr : _deriveDistributions(rel.getInput(), mq))
+ res.add(inDistr.apply(mapping));
+
+ return new ArrayList<>(res);
+ }
+
+ /**
+ * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
+ */
public List<IgniteDistribution> deriveDistributions(LogicalAggregate rel, RelMetadataQuery mq) {
- List<IgniteDistributions.Suggestion> suggestions = IgniteDistributions.suggestAggregate(
+ List<Suggestion> suggestions = IgniteDistributions.suggestAggregate(
mq, rel.getInput(), rel.getGroupSet(), rel.getGroupSets());
- return Commons.transform(suggestions, IgniteDistributions.Suggestion::out);
+ return Commons.transform(suggestions, Suggestion::out);
}
/**
* See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
*/
public List<IgniteDistribution> deriveDistributions(LogicalJoin rel, RelMetadataQuery mq) {
- List<IgniteDistributions.BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
+ List<BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
mq, rel.getLeft(), rel.getRight(), rel.analyzeCondition(), rel.getJoinType());
- return Commons.transform(suggestions, IgniteDistributions.BiSuggestion::out);
+ return Commons.transform(suggestions, BiSuggestion::out);
+ }
+
+ /**
+ * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
+ */
+ public List<IgniteDistribution> deriveDistributions(LogicalUnion rel, RelMetadataQuery mq) {
+ List<IgniteDistribution> distributions = Commons.transform(
+ IgniteDistributions.suggestUnionAll(mq, rel.getInputs()), Suggestion::out);
+
+ if (rel.all)
+ return distributions;
+
+ // UNION becomes DISTINCT + UNION ALL where DISTINCT is implemented as an Aggregate
+ ImmutableBitSet groupSet = ImmutableBitSet.range(rel.getRowType().getFieldCount());
+ List<ImmutableBitSet> groupSets = ImmutableList.of(groupSet);
+
+ return Commons.transform(
+ IgniteDistributions.suggestAggregate(distributions, groupSet, groupSets), Suggestion::out);
}
/** */
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
index 5e7f7f2..8c040a1 100644
--- 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
@@ -17,18 +17,12 @@
package org.apache.ignite.internal.processors.query.calcite.metadata;
-import java.util.List;
import org.apache.calcite.plan.hep.HepRelVertex;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.logical.LogicalAggregate;
-import org.apache.calcite.rel.logical.LogicalExchange;
-import org.apache.calcite.rel.logical.LogicalFilter;
-import org.apache.calcite.rel.logical.LogicalJoin;
-import org.apache.calcite.rel.logical.LogicalProject;
-import org.apache.calcite.rel.logical.LogicalTableModify;
-import org.apache.calcite.rel.logical.LogicalTableScan;
-import org.apache.calcite.rel.logical.LogicalValues;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.core.Values;
import org.apache.calcite.rel.metadata.BuiltInMetadata;
import org.apache.calcite.rel.metadata.MetadataDef;
import org.apache.calcite.rel.metadata.MetadataHandler;
@@ -41,7 +35,6 @@ import org.apache.ignite.internal.processors.query.calcite.schema.DistributedTab
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionTraitDef;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
-import org.apache.ignite.internal.util.typedef.F;
/**
* Implementation class for {@link RelMetadataQuery#distribution(RelNode)} method call.
@@ -66,7 +59,7 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
* @return Distribution type of the given relational node.
*/
public IgniteDistribution distribution(RelNode rel, RelMetadataQuery mq) {
- return DistributionTraitDef.INSTANCE.getDefault();
+ return rel.getTraitSet().getTrait(DistributionTraitDef.INSTANCE);
}
/**
@@ -79,47 +72,22 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
- public IgniteDistribution distribution(LogicalTableModify rel, RelMetadataQuery mq) {
- return IgniteDistributions.single(); // TODO skip reducer
- }
-
- /**
- * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
- */
- public IgniteDistribution distribution(LogicalFilter rel, RelMetadataQuery mq) {
- return _distribution(rel.getInput(), mq);
- }
-
- /**
- * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
- */
- public IgniteDistribution distribution(LogicalProject rel, RelMetadataQuery mq) {
- return IgniteDistributions.project(mq, rel.getInput(), rel.getProjects());
+ public IgniteDistribution distribution(TableScan rel, RelMetadataQuery mq) {
+ return rel.getTable().unwrap(DistributedTable.class).distribution();
}
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
- public IgniteDistribution distribution(LogicalJoin rel, RelMetadataQuery mq) {
- IgniteDistribution leftIn = _distribution(rel.getLeft(), mq);
- IgniteDistribution rightIn = _distribution(rel.getRight(), mq);
-
- List<IgniteDistributions.BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
- leftIn, rightIn, rel.analyzeCondition(), rel.getJoinType());
-
- return F.first(suggestions).out();
+ public IgniteDistribution distribution(Values rel, RelMetadataQuery mq) {
+ return IgniteDistributions.broadcast();
}
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
- public IgniteDistribution distribution(LogicalAggregate rel, RelMetadataQuery mq) {
- IgniteDistribution inDistr = _distribution(rel.getInput(), mq);
-
- List<IgniteDistributions.Suggestion> suggestions =
- IgniteDistributions.suggestAggregate(inDistr, rel.getGroupSet(), rel.getGroupSets());
-
- return F.first(suggestions).out();
+ public IgniteDistribution distribution(Exchange rel, RelMetadataQuery mq) {
+ return (IgniteDistribution) rel.distribution;
}
/**
@@ -132,27 +100,6 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
- public IgniteDistribution distribution(LogicalTableScan rel, RelMetadataQuery mq) {
- return rel.getTable().unwrap(DistributedTable.class).distribution();
- }
-
- /**
- * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
- */
- public IgniteDistribution distribution(LogicalValues rel, RelMetadataQuery mq) {
- return IgniteDistributions.broadcast();
- }
-
- /**
- * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
- */
- public IgniteDistribution distribution(LogicalExchange rel, RelMetadataQuery mq) {
- return (IgniteDistribution) rel.distribution;
- }
-
- /**
- * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
- */
public IgniteDistribution distribution(HepRelVertex rel, RelMetadataQuery mq) {
return _distribution(rel.getCurrentRel(), mq);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
index 9488866..b860711 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
@@ -23,6 +23,7 @@ import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.BiRel;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.core.SetOp;
import org.apache.calcite.rel.metadata.MetadataDef;
import org.apache.calcite.rel.metadata.MetadataHandler;
import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
@@ -118,6 +119,32 @@ public class IgniteMdFragmentInfo implements MetadataHandler<FragmentMetadata> {
/**
* See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
*
+ * {@link LocationMappingException} may be thrown on two children nodes locations merge. This means
+ * that the fragment (which part the parent node is) cannot be executed on any node and additional exchange
+ * is needed. This case we throw {@link OptimisticPlanningException} with an edge, where we need the additional
+ * exchange. After the exchange is put into the fragment and the fragment is split into two ones, fragment meta
+ * information will be recalculated for all fragments.
+ */
+ public FragmentInfo fragmentInfo(SetOp rel, RelMetadataQuery mq) {
+ FragmentInfo res = new FragmentInfo();
+
+ for (RelNode input : rel.getInputs()) {
+ FragmentInfo inputInfo = _fragmentInfo(input, mq);
+
+ try {
+ res = res.merge(inputInfo);
+ }
+ catch (LocationMappingException e) {
+ throw new OptimisticPlanningException("Failed to calculate physical distribution", input, e);
+ }
+ }
+
+ return res;
+ }
+
+ /**
+ * See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
+ *
* Prunes involved partitions (hence nodes, involved in query execution) if possible.
*/
public FragmentInfo fragmentInfo(IgniteFilter rel, RelMetadataQuery mq) {
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
index 238de49..b13f99a 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
@@ -37,7 +37,9 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
/** */
@@ -136,11 +138,18 @@ class Cloner implements IgniteRelVisitor<IgniteRel> {
}
/** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteUnionAll rel) {
+ List<RelNode> inputs = Commons.transform(rel.getInputs(), rel0 -> visit((IgniteRel) rel0));
+
+ return new IgniteUnionAll(cluster, rel.getTraitSet(), inputs);
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteRel visit(IgniteReceiver rel) {
Fragment fragment = rel.source();
fragments.add(fragment = new Fragment(fragment.fragmentId(), visit(fragment.root())));
- return new IgniteReceiver(cluster, rel.getTraitSet(), rel.getRowType(), fragment);
+ return new IgniteReceiver(cluster, rel.getTraitSet(), fragment);
}
/** {@inheritDoc} */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
index 18655bf..3083feb 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
@@ -21,12 +21,11 @@ import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Objects;
import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTrait;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
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.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
@@ -41,9 +40,12 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
+import org.apache.ignite.internal.processors.query.calcite.trait.DistributionTrait;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -84,191 +86,230 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteSender rel) {
- // a split may happen on BiRel inputs merge. A sender node cannot be a BiRel input.
- assert rel != cutPoint;
assert cutPoint != null;
+ assert rel != cutPoint;
+
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
+
+ assert input != newInput;
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ checkTraitsEqual(input, newInput);
+
+ RelTraitSet traits = rel.getTraitSet();
+
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteFilter rel) {
+ if (cutPoint == rel)
+ return split(rel);
+
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
if (input == newInput)
return rel;
- return (IgniteRel) rel.copy(rel.getTraitSet(), ImmutableList.of(newInput));
+ checkTraits(input, newInput);
+
+ RelTraitSet traits = rel.getTraitSet()
+ .replace(newInput.distribution());
+
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteAggregate rel) {
- boolean split = cutPoint(rel);
+ @Override public IgniteRel visit(IgniteTrimExchange rel) {
+ if (cutPoint == rel)
+ return split(rel);
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ if (cutPoint == null)
+ return (IgniteRel) rel.getInput();
- if (input != newInput) {
- checkDistributionEqual(input, newInput);
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
- rel = (IgniteAggregate) rel.copy(rel.getTraitSet(), ImmutableList.of(newInput));
- }
+ if (input == newInput)
+ return rel;
+
+ checkTraitsEqual(input, newInput);
+
+ RelTraitSet traits = rel.getTraitSet();
- return split ? split(rel) : rel;
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteMapAggregate rel) {
- // a split may happen on BiRel inputs merge. An map aggregate node cannot be a BiRel input.
- assert rel != cutPoint;
- assert cutPoint != null;
+ @Override public IgniteRel visit(IgniteProject rel) {
+ if (cutPoint == rel)
+ return split(rel);
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
if (input == newInput)
return rel;
- IgniteDistribution newDistr = IgniteDistributions.mapAggregate(mq,
- newInput, rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
-
- assert newDistr != null;
+ checkTraits(input, newInput);
RelTraitSet traits = rel.getTraitSet()
- .replace(newDistr);
+ .replace(IgniteDistributions.project(mq, newInput, rel.getProjects()));
- return (IgniteRel) rel.copy(traits, ImmutableList.of(newInput));
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteReduceAggregate rel) {
- // a split may happen on BiRel inputs merge. An reduce aggregate node doesn't have a
- // physical mapping (because it always goes after receiver node), so, its merge with
- // any input cannot cause the split.
- assert cutPoint == null;
+ @Override public IgniteRel visit(IgniteJoin rel) {
+ if (cutPoint == null)
+ return rel; // No need to check deeper
- if (U.assertionsEnabled()) {
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ if (cutPoint == rel)
+ return split(rel);
- assert input == newInput;
- }
+ IgniteRel left = (IgniteRel) rel.getLeft();
+ IgniteRel newLeft = visit(left);
+ IgniteRel right = (IgniteRel) rel.getRight();
+ IgniteRel newRight = visit(right);
- return rel;
+ if (left == newLeft && right == newRight)
+ return rel;
+
+ checkTraitsEqual(left, newLeft);
+ checkTraitsEqual(right, newRight);
+
+ RelTraitSet traits = rel.getTraitSet();
+
+ return (IgniteRel) rel.copy(traits, F.asList(newLeft, newRight));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteFilter rel) {
- boolean split = cutPoint(rel);
+ @Override public IgniteRel visit(IgniteTableScan rel) {
+ if (rel == cutPoint)
+ return split(rel);
- RelNode input = rel.getInput();
- IgniteRel newInput = visit((IgniteRel) input);
+ assert cutPoint == null;
- if (input != newInput) {
- RelTraitSet traits = rel.getTraitSet()
- .replace(newInput.distribution());
+ return rel;
+ }
- rel = (IgniteFilter) rel.copy(traits, ImmutableList.of(newInput));
- }
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteReceiver rel) {
+ assert cutPoint == null;
- return split ? split(rel) : rel;
+ return rel;
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteTrimExchange rel) {
- boolean split = cutPoint(rel);
+ @Override public IgniteRel visit(IgniteExchange rel) {
+ throw new AssertionError();
+ }
- // after split we need to remove the exchange, return an original
- // distribution and propagate the distribution to a fragment root.
- if (split)
- return split((IgniteRel) rel.getInput());
- else if (cutPoint == null)
- return (IgniteRel) rel.getInput();
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteAggregate rel) {
+ if (rel == cutPoint)
+ return split(rel);
- RelNode input = rel.getInput();
- IgniteRel newInput = visit((IgniteRel) input);
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
if (input == newInput)
return rel;
- return (IgniteRel) rel.copy(rel.getTraitSet(), F.asList(newInput));
- }
-
- /** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteProject rel) {
- boolean split = cutPoint(rel);
+ checkTraits(input, newInput);
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ IgniteDistribution distr = IgniteDistributions.aggregate(mq,
+ newInput, rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
- if (input != newInput) {
- RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteDistributions.project(mq, newInput, rel.getProjects()));
+ assert distr != null;
- rel = (IgniteProject) rel.copy(traits, ImmutableList.of(newInput));
- }
+ RelTraitSet traits = rel.getTraitSet().replace(distr);
- return split ? split(rel) : rel;
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteJoin rel) {
- if (cutPoint == null)
- return rel; // no need to check deeper
+ @Override public IgniteRel visit(IgniteMapAggregate rel) {
+ assert rel != cutPoint;
- boolean split = cutPoint(rel);
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
- RelNode left = rel.getLeft();
- RelNode right = rel.getRight();
+ if (input == newInput)
+ return rel;
- RelNode newLeft = visit((IgniteRel) left);
- RelNode newRight = visit((IgniteRel) right);
+ checkTraitsEqual(input, newInput);
- if (left != newLeft || right != newRight) {
- // Join requires input distribution and produces its own one.
- // It cannot change the distribution on a child node change.
- checkDistributionEqual(left, newLeft);
- checkDistributionEqual(right, newRight);
+ RelTraitSet traits = rel.getTraitSet();
- rel = (IgniteJoin) rel.copy(rel.getTraitSet(), ImmutableList.of(newLeft, newRight));
- }
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteReduceAggregate rel) {
+ assert cutPoint == null;
- return split ? split(rel) : rel;
+ return rel;
}
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteTableModify rel) {
- boolean split = cutPoint(rel);
+ if (cutPoint == null)
+ return rel; // No need to check deeper
- RelNode input = rel.getInput();
- RelNode newInput = visit((IgniteRel) input);
+ if (rel == cutPoint)
+ return split(rel);
- if (input != newInput) {
- checkDistributionEqual(input, newInput);
+ IgniteRel input = (IgniteRel) rel.getInput();
+ IgniteRel newInput = visit(input);
- rel = (IgniteTableModify) rel.copy(rel.getTraitSet(), ImmutableList.of(newInput));
- }
+ if (input == newInput)
+ return rel;
- return split ? split(rel) : rel;
- }
+ checkTraitsEqual(input, newInput);
- /** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteTableScan rel) {
- return cutPoint(rel) ? split(rel) : rel;
+ RelTraitSet traits = rel.getTraitSet();
+
+ return (IgniteRel) rel.copy(traits, F.asList(newInput));
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteReceiver rel) {
- // a split may happen on BiRel inputs merge. A receiver doesn't have a
- // physical mapping, so, its merge with any input cannot cause the split.
+ @Override public IgniteRel visit(IgniteValues rel) {
assert cutPoint == null;
return rel;
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteValues rel) {
- // a split may happen on BiRel inputs merge. A values node doesn't have a
- // physical mapping, so, its merge with any input cannot cause the split.
- assert cutPoint == null;
+ @Override public IgniteRel visit(IgniteUnionAll rel) {
+ if (cutPoint == null)
+ return rel; // No need to check deeper
- return rel;
+ if (rel == cutPoint)
+ return split(rel);
+
+ boolean inputChanged = false;
+
+ List<IgniteRel> inputs = Commons.cast(rel.getInputs());
+ List<IgniteRel> newInputs = new ArrayList<>(inputs.size());
+
+ for (RelNode input : inputs) {
+ IgniteRel newInput = visit(input);
+
+ if (input != newInput) {
+ inputChanged = true;
+
+ checkTraitsEqual((IgniteRel) input, newInput);
+ }
+
+ newInputs.add(newInput);
+ }
+
+ if (!inputChanged)
+ return rel;
+
+ return (IgniteRel) rel.copy(rel.getTraitSet(), Commons.cast(newInputs));
}
/** {@inheritDoc} */
@@ -276,41 +317,61 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
return rel.accept(this);
}
- /** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteExchange rel) {
- throw new AssertionError();
+ /** */
+ private IgniteRel visit(RelNode rel) {
+ return visit((IgniteRel) rel);
}
/** */
- public boolean cutPoint(RelNode rel) {
- boolean res = rel == cutPoint;
+ private void checkTraits(IgniteRel orig, IgniteRel newRel) {
+ if (!U.assertionsEnabled() || orig == newRel)
+ return;
- if (res)
- cutPoint = null;
+ RelTraitSet origTraits = orig.getTraitSet();
+ RelTraitSet newTraits = newRel.getTraitSet();
- return res;
+ ImmutableList<RelTrait> difference = origTraits.difference(newTraits);
+
+ // Only distribution trait may be changed after split
+ if (difference.isEmpty() || difference.size() == 1 && F.first(difference) instanceof DistributionTrait)
+ return;
+
+ assert false : difference;
}
/** */
- private IgniteRel split(IgniteRel input) {
- RelOptCluster cluster = input.getCluster();
- RelTraitSet traits = input.getTraitSet();
+ private void checkTraitsEqual(IgniteRel orig, IgniteRel newRel) {
+ if (!U.assertionsEnabled() || orig == newRel)
+ return;
- Fragment fragment = new Fragment(new IgniteSender(cluster, traits, input));
+ RelTraitSet origTraits = orig.getTraitSet();
+ RelTraitSet newTraits = newRel.getTraitSet();
- fragments.add(fragment);
+ ImmutableList<RelTrait> difference = origTraits.difference(newTraits);
+
+ if (difference.isEmpty())
+ return;
- return new IgniteReceiver(cluster, traits, input.getRowType(), fragment);
+ assert false : difference;
}
/** */
- private void checkDistributionEqual(RelNode rel1, RelNode rel2) {
- if (!U.assertionsEnabled())
- return;
+ private IgniteRel split(IgniteRel rel) {
+ cutPoint = null;
+
+ IgniteRel newRel = visit(rel);
+
+ checkTraits(rel, newRel);
+
+ RelOptCluster cluster = rel.getCluster();
+ RelTraitSet traits = rel.getTraitSet();
- IgniteDistribution distr1 = IgniteMdDistribution._distribution(rel1, mq);
- IgniteDistribution distr2 = IgniteMdDistribution._distribution(rel2, mq);
+ IgniteSender sender = new IgniteSender(cluster, traits, newRel);
+
+ Fragment fragment = new Fragment(sender);
+
+ fragments.add(fragment);
- assert Objects.equals(distr1, distr2) : "distr1=" + distr1 + "; distr2=" + distr2;
+ return new IgniteReceiver(cluster, traits, fragment);
}
}
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 551aa99..50be506 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
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.query.calcite.prepare;
import org.apache.calcite.rel.rules.SubQueryRemoveRule;
+import org.apache.calcite.rel.rules.UnionMergeRule;
import org.apache.calcite.tools.Program;
import org.apache.calcite.tools.RuleSet;
import org.apache.calcite.tools.RuleSets;
@@ -30,6 +31,8 @@ import org.apache.ignite.internal.processors.query.calcite.rule.JoinTraitsPropag
import org.apache.ignite.internal.processors.query.calcite.rule.ProjectConverterRule;
import org.apache.ignite.internal.processors.query.calcite.rule.ProjectTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.TableModifyConverterRule;
+import org.apache.ignite.internal.processors.query.calcite.rule.UnionConverterRule;
+import org.apache.ignite.internal.processors.query.calcite.rule.UnionTraitsTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.ValuesConverterRule;
import static org.apache.ignite.internal.processors.query.calcite.prepare.IgnitePrograms.cbo;
@@ -71,7 +74,10 @@ public enum PlannerPhase {
ProjectTraitsPropagationRule.INSTANCE,
FilterConverterRule.INSTANCE,
FilterTraitsPropagationRule.INSTANCE,
- TableModifyConverterRule.INSTANCE);
+ TableModifyConverterRule.INSTANCE,
+ UnionMergeRule.INSTANCE,
+ UnionConverterRule.INSTANCE,
+ UnionTraitsTraitsPropagationRule.INSTANCE);
}
/** {@inheritDoc} */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
index 1cbfc9e..4cdf985 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
/**
@@ -112,6 +113,11 @@ public class Splitter implements IgniteRelVisitor<IgniteRel> {
}
/** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteUnionAll rel) {
+ return visitChildren(rel);
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteRel visit(IgniteRel rel) {
return rel.accept(this);
}
@@ -135,7 +141,7 @@ public class Splitter implements IgniteRelVisitor<IgniteRel> {
fragments.add(fragment);
- return new IgniteReceiver(cluster, rel.getTraitSet(), input.getRowType(), fragment);
+ return new IgniteReceiver(cluster, rel.getTraitSet(), fragment);
}
/**
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java
index ae86e74..46cad1d 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java
@@ -22,7 +22,6 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.AbstractRelNode;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
/**
@@ -37,19 +36,19 @@ public class IgniteReceiver extends AbstractRelNode implements IgniteRel {
*
* @param cluster Cluster that this relational expression belongs to
* @param traits Traits of this relational expression
- * @param rowType Output row type
* @param source Source fragment.
*/
- public IgniteReceiver(RelOptCluster cluster, RelTraitSet traits, RelDataType rowType, Fragment source) {
+ public IgniteReceiver(RelOptCluster cluster, RelTraitSet traits, Fragment source) {
super(cluster, traits);
- this.rowType = rowType;
this.source = source;
+
+ rowType = source.root().getRowType();
}
/** {@inheritDoc} */
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return new IgniteReceiver(getCluster(), traitSet, rowType, source);
+ return new IgniteReceiver(getCluster(), traitSet, source);
}
/** {@inheritDoc} */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
index 3595a9b..4c9a4ab 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
@@ -30,6 +30,7 @@ import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Util;
+import org.apache.ignite.internal.processors.query.calcite.trait.DistributionTraitDef;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
@@ -46,14 +47,14 @@ public class IgniteReduceAggregate extends SingleRel implements IgniteRel {
/** */
private final List<AggregateCall> aggCalls;
- /***/
+ /** */
public IgniteReduceAggregate(RelOptCluster cluster, RelTraitSet traits, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls, RelDataType rowType) {
super(cluster, traits, input);
assert rowType != null;
assert RelOptUtil.areRowTypesEqual(input.getRowType(),
Commons.aggregationDataRowType(getCluster().getTypeFactory()), true);
- assert getInput().getTraitSet().contains(IgniteDistributions.single());
+ assert traits.getTrait(DistributionTraitDef.INSTANCE).satisfies(IgniteDistributions.single());
this.groupSet = groupSet;
this.groupSets = groupSets;
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
index cc178ce..0cd1275 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
@@ -87,6 +87,11 @@ public interface IgniteRelVisitor<T> {
T visit(IgniteValues rel);
/**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
+ T visit(IgniteUnionAll rel);
+
+ /**
* Visits a relational node and calculates a result on the basis of node meta information.
* @param rel Relational node.
* @return Visit result.
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSender.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSender.java
index b95af15..4e4d84e 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSender.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteSender.java
@@ -41,12 +41,9 @@ public class IgniteSender extends SingleRel implements IgniteRel, RelTargetAware
* @param cluster Cluster that this relational expression belongs to
* @param traits Traits of this relational expression
* @param input input relational expression
- * @param target Remote targets information
*/
- public IgniteSender(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelTarget target) {
- super(cluster, traits, input);
-
- this.target = target;
+ public IgniteSender(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
+ this(cluster, traits, input, null);
}
/**
@@ -55,9 +52,12 @@ public class IgniteSender extends SingleRel implements IgniteRel, RelTargetAware
* @param cluster Cluster that this relational expression belongs to
* @param traits Traits of this relational expression
* @param input input relational expression
+ * @param target Remote targets information
*/
- public IgniteSender(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- this(cluster, traits, input, null);
+ private IgniteSender(RelOptCluster cluster, RelTraitSet traits, RelNode input, RelTarget target) {
+ super(cluster, traits, input);
+
+ this.target = target;
}
/** {@inheritDoc} */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteUnionAll.java
similarity index 52%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteUnionAll.java
index ae86e74..b184493 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReceiver.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteUnionAll.java
@@ -19,37 +19,29 @@ package org.apache.ignite.internal.processors.query.calcite.rel;
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.AbstractRelNode;
+import org.apache.calcite.rel.RelDistribution;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
+import org.apache.calcite.rel.core.SetOp;
+import org.apache.calcite.rel.core.Union;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
/**
- * Relational expression that receives elements from remote {@link IgniteSender}
+ *
*/
-public class IgniteReceiver extends AbstractRelNode implements IgniteRel {
+public class IgniteUnionAll extends Union implements IgniteRel {
/** */
- private final Fragment source;
-
- /**
- * Creates a Receiver
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param traits Traits of this relational expression
- * @param rowType Output row type
- * @param source Source fragment.
- */
- public IgniteReceiver(RelOptCluster cluster, RelTraitSet traits, RelDataType rowType, Fragment source) {
- super(cluster, traits);
-
- this.rowType = rowType;
- this.source = source;
+ public IgniteUnionAll(RelOptCluster cluster, RelTraitSet traits, List<RelNode> inputs) {
+ super(cluster, traits, inputs, true);
}
/** {@inheritDoc} */
- @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return new IgniteReceiver(getCluster(), traitSet, rowType, source);
+ @Override public SetOp copy(RelTraitSet traitSet, List<RelNode> inputs, boolean all) {
+ assert all;
+
+ return new IgniteUnionAll(getCluster(), traitSet, inputs);
}
/** {@inheritDoc} */
@@ -57,10 +49,11 @@ public class IgniteReceiver extends AbstractRelNode implements IgniteRel {
return visitor.visit(this);
}
- /**
- * @return Source fragment.
- */
- public Fragment source() {
- return source;
+ /** {@inheritDoc} */
+ @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ if (distribution().getType() == RelDistribution.Type.ANY)
+ return planner.getCostFactory().makeInfiniteCost();
+
+ return super.computeSelfCost(planner, mq);
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
index 8bd7a6c..b7be4f3 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.query.calcite.rule;
import java.util.ArrayList;
import java.util.List;
-import java.util.Objects;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
@@ -34,6 +33,7 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregat
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.util.typedef.F;
/**
*
@@ -62,7 +62,7 @@ public class AggregateTraitsPropagationRule extends RelOptRule {
IgniteDistribution inDistr = IgniteMdDistribution._distribution(input, mq);
List<IgniteDistributions.Suggestion> suggestions =
- IgniteDistributions.suggestAggregate(inDistr, groupSet, groupSets);
+ IgniteDistributions.suggestAggregate(F.asList(inDistr), groupSet, groupSets);
List<RelNode> newRels = new ArrayList<>(suggestions.size());
@@ -88,7 +88,7 @@ public class AggregateTraitsPropagationRule extends RelOptRule {
/** */
private boolean isMapReduce(IgniteDistributions.Suggestion suggestion) {
- return Objects.equals(suggestion.out(), IgniteDistributions.single())
+ return suggestion.out().satisfies(IgniteDistributions.single())
&& suggestion.in().satisfies(IgniteDistributions.random());
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
index 1131ac2..dc1e484 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
@@ -21,6 +21,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.util.List;
import java.util.Map;
+import org.apache.calcite.plan.Convention;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptRuleOperand;
@@ -34,6 +35,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.NotNull;
import static org.apache.calcite.plan.RelOptRule.any;
@@ -86,28 +88,16 @@ public class RuleUtils {
/** */
public static void transformTo(RelOptRuleCall call, List<RelNode> newRels) {
- transformTo(call, newRels, ImmutableMap.of());
- }
-
- /** */
- public static void transformTo(RelOptRuleCall call, List<RelNode> newRels, Map<RelNode, RelNode> additional) {
RelNode orig = call.rel(0);
- if (F.isEmpty(newRels)) {
- if (!F.isEmpty(additional))
- // small trick to register the additional equivalence map entries only, we pass
- // the original rel as transformed one, which will be skipped by the planner.
- call.transformTo(orig, additional);
-
+ if (F.isEmpty(newRels))
return;
- }
if (isRoot(orig))
newRels = Commons.transform(newRels, RuleUtils::changeToRootTraits);
RelNode first = F.first(newRels);
- List<RelNode> remaining = newRels.subList(1, newRels.size());
- Map<RelNode, RelNode> equivMap = equivMap(orig, remaining, additional);
+ Map<RelNode, RelNode> equivMap = equivMap(orig, newRels.subList(1, newRels.size()));
call.transformTo(first, equivMap);
}
@@ -181,27 +171,27 @@ public class RuleUtils {
/** */
private static RelNode changeToRootTraits(RelNode rel) {
+ if (rel.getConvention() == Convention.NONE)
+ return rel; // logical rels will be converted after to-ignite transformation
+
RelTraitSet rootTraits = rel.getCluster().getPlanner().getRoot().getTraitSet();
return changeTraits(rel, rootTraits);
}
/** */
- private static @NotNull Map<RelNode, RelNode> equivMap(RelNode orig, List<RelNode> equivList, Map<RelNode, RelNode> additional) {
+ private static @NotNull Map<RelNode, RelNode> equivMap(RelNode orig, List<RelNode> equivList) {
assert orig != null;
assert equivList != null;
- assert additional != null;
- if(F.isEmpty(equivList))
- return additional;
+ if (F.isEmpty(equivList))
+ return ImmutableMap.of();
- ImmutableMap.Builder<RelNode, RelNode> b = ImmutableMap.builder();
+ Map<RelNode, RelNode> res = U.newHashMap(equivList.size());
for (RelNode equiv : equivList)
- b.put(equiv, orig);
-
- b.putAll(additional);
+ res.put(equiv, orig);
- return b.build();
+ return res;
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionConverterRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionConverterRule.java
new file mode 100644
index 0000000..f2f378b
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionConverterRule.java
@@ -0,0 +1,64 @@
+/*
+ * 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 java.util.List;
+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.RelNode;
+import org.apache.calcite.rel.logical.LogicalUnion;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/**
+ *
+ */
+public class UnionConverterRule extends RelOptRule {
+ /** */
+ public static final RelOptRule INSTANCE = new UnionConverterRule();
+
+ /** */
+ public UnionConverterRule() {
+ super(operand(LogicalUnion.class, any()));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMatch(RelOptRuleCall call) {
+ LogicalUnion rel = call.rel(0);
+
+ RelOptCluster cluster = rel.getCluster();
+ List<RelNode> inputs = Commons.transform(rel.getInputs(),
+ rel0 -> convert(rel0, IgniteConvention.INSTANCE));
+ RelTraitSet traits = rel.getTraitSet()
+ .replace(IgniteConvention.INSTANCE);
+
+ RelNode res = new IgniteUnionAll(cluster, traits, inputs);
+
+ if (!rel.all) {
+ RelBuilder b = relBuilderFactory.create(rel.getCluster(), null);
+
+ res = b.push(res).aggregate(b.groupKey(b.fields())).build();
+ }
+
+ RuleUtils.transformTo(call, res);
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionTraitsTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionTraitsTraitsPropagationRule.java
new file mode 100644
index 0000000..4cca3db
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/UnionTraitsTraitsPropagationRule.java
@@ -0,0 +1,69 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+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.RelNode;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions.Suggestion;
+
+/**
+ *
+ */
+public class UnionTraitsTraitsPropagationRule extends RelOptRule {
+ /** */
+ public static final RelOptRule INSTANCE = new UnionTraitsTraitsPropagationRule();
+
+ /** */
+ public UnionTraitsTraitsPropagationRule() {
+ super(RuleUtils.traitPropagationOperand(IgniteUnionAll.class));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMatch(RelOptRuleCall call) {
+ IgniteUnionAll rel = call.rel(0);
+ List<RelNode> inputs = rel.getInputs();
+
+ RelOptCluster cluster = rel.getCluster();
+ RelMetadataQuery mq = call.getMetadataQuery();
+
+ List<Suggestion> suggestions =
+ IgniteDistributions.suggestUnionAll(mq, inputs);
+
+ List<RelNode> newRels = new ArrayList<>(suggestions.size());
+
+ for (Suggestion suggestion : suggestions) {
+ RelTraitSet traits = rel.getTraitSet().replace(suggestion.out());
+ List<RelNode> inputs0 = new ArrayList<>(inputs.size());
+
+ for (RelNode input : inputs)
+ inputs0.add(RuleUtils.changeTraits(input, suggestion.in()));
+
+ newRels.add(new IgniteUnionAll(cluster, traits, inputs0));
+ }
+
+ RuleUtils.transformTo(call, newRels);
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
index 028daa0..0c1aee1 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
@@ -72,6 +72,11 @@ public interface PhysicalRelVisitor<T> {
T visit(TableModifyPhysicalRel rel);
/**
+ * See {@link PhysicalRelVisitor#visit(PhysicalRel)}
+ */
+ T visit(UnionAllPhysicalRel rel);
+
+ /**
* Processes given physical rel.
*
* @param rel Physical rel.
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
index 0466219..b07453f 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
@@ -18,8 +18,10 @@
package org.apache.ignite.internal.processors.query.calcite.serialize;
import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
import java.util.List;
import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
@@ -43,6 +45,7 @@ import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTrimExchange;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteUnionAll;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFunction;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
@@ -174,6 +177,16 @@ public class RelToPhysicalConverter implements IgniteRelVisitor<PhysicalRel> {
}
/** {@inheritDoc} */
+ @Override public PhysicalRel visit(IgniteUnionAll rel) {
+ List<PhysicalRel> inputs = new ArrayList<>(rel.getInputs().size());
+
+ for (RelNode input : rel.getInputs())
+ inputs.add(visit((IgniteRel) input));
+
+ return new UnionAllPhysicalRel(inputs);
+ }
+
+ /** {@inheritDoc} */
@Override public PhysicalRel visit(IgniteExchange rel) {
throw new AssertionError();
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/UnionAllPhysicalRel.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/UnionAllPhysicalRel.java
new file mode 100644
index 0000000..ae39973
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/UnionAllPhysicalRel.java
@@ -0,0 +1,67 @@
+/*
+ * 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.serialize;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.List;
+
+/** */
+public class UnionAllPhysicalRel implements PhysicalRel {
+ /** */
+ private List<PhysicalRel> inputs;
+
+ /** */
+ public UnionAllPhysicalRel(List<PhysicalRel> inputs) {
+ this.inputs = inputs;
+ }
+
+ /** */
+ public UnionAllPhysicalRel() {
+ }
+
+ /** */
+ public List<PhysicalRel> inputs() {
+ return inputs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T accept(PhysicalRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(inputs.size());
+ for (PhysicalRel input : inputs)
+ out.writeObject(input);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ int size = in.readInt();
+ List<PhysicalRel> inputs = new ArrayList<>(size);
+
+ for (int i = 0; i < size; i++)
+ inputs.add((PhysicalRel) in.readObject());
+
+ this.inputs = inputs;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
index d85dd91..27dcbdf 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/DistributionTrait.java
@@ -41,10 +41,10 @@ public final class DistributionTrait implements IgniteDistribution {
Ordering.<Integer>natural().lexicographical();
/** */
- private DistributionFunction function;
+ private final DistributionFunction function;
/** */
- private ImmutableIntList keys;
+ private final ImmutableIntList keys;
/**
* @param function Distribution function.
@@ -139,14 +139,38 @@ public final class DistributionTrait implements IgniteDistribution {
/** {@inheritDoc} */
@Override public IgniteDistribution apply(Mappings.TargetMapping mapping) {
- if (keys.isEmpty())
+ if (getType() != HASH_DISTRIBUTED)
return this;
- assert getType() == HASH_DISTRIBUTED;
+ if (mapping.getTargetCount() < keys.size())
+ IgniteDistributions.random();
- ImmutableIntList newKeys = IgniteDistributions.projectDistributionKeys(mapping, keys);
+ int[] map = new int[mapping.getSourceCount()];
+ int[] res = new int[keys.size()];
- return newKeys.isEmpty() ? IgniteDistributions.random() : IgniteDistributions.hash(newKeys, function);
+ for (int i = 0; i < keys.size(); i++)
+ map[keys.getInt(i)] = i + 1;
+
+ for (int i = 0, found = 0; i < mapping.getTargetCount(); i++) {
+ int source = mapping.getSourceOpt(i);
+
+ if (source == -1)
+ continue;
+
+ int keyPos = map[source] - 1;
+
+ if (keyPos == -1)
+ continue;
+
+ res[keyPos] = i;
+
+ if (++found == keys.size())
+ return IgniteDistributions.hash(ImmutableIntList.of(res), function);
+
+ map[source] = 0;
+ }
+
+ return IgniteDistributions.random();
}
/** {@inheritDoc} */
@@ -156,8 +180,6 @@ public final class DistributionTrait implements IgniteDistribution {
/** {@inheritDoc} */
@Override public int compareTo(RelMultipleTrait o) {
- // TODO is this method really needed??
-
final IgniteDistribution distribution = (IgniteDistribution) o;
if (getType() == distribution.getType() && getType() == Type.HASH_DISTRIBUTED) {
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
index 1e4bb02..d504307 100644
--- 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
@@ -19,11 +19,10 @@ package org.apache.ignite.internal.processors.query.calcite.trait;
import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import org.apache.calcite.plan.RelTraitDef;
import org.apache.calcite.rel.RelDistribution;
import org.apache.calcite.rel.RelNode;
@@ -39,7 +38,6 @@ import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
import org.apache.calcite.util.mapping.Mappings;
-import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDerivedDistribution;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
@@ -52,8 +50,6 @@ import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFun
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.NotNull;
import static org.apache.calcite.rel.core.JoinRelType.INNER;
import static org.apache.calcite.rel.core.JoinRelType.LEFT;
@@ -64,12 +60,6 @@ import static org.apache.calcite.rel.core.JoinRelType.RIGHT;
*/
public class IgniteDistributions {
/** */
- private static final int BEST_CNT = IgniteSystemProperties.getInteger("IGNITE_CALCITE_JOIN_SUGGESTS_COUNT", 0);
-
- /** */
- private static final Integer[] INTS = new Integer[]{0, 1, 2};
-
- /** */
private static final IgniteDistribution BROADCAST = canonize(new DistributionTrait(BroadcastDistribution.INSTANCE));
/** */
@@ -147,111 +137,107 @@ public class IgniteDistributions {
}
/**
- * Suggests possible aggregate distributions.
+ * Suggests possible union distributions.
*
- * @param inDistr Input node distribution.
- * @param groupSet Aggregate group set.
- * @param groupSets Aggregate group sets.
- * @return Array of possible distributions, sorted by their efficiency (cheaper first).
+ * @param mq Metadata query.
+ * @param inputs Input nodes.
+ * @return Array of possible distributions.
*/
- public static List<Suggestion> suggestAggregate(IgniteDistribution inDistr, ImmutableBitSet groupSet,
- List<ImmutableBitSet> groupSets) {
- return sorted(suggestAggregate0(new HashMap<>(), inDistr, groupSet, groupSets));
+ public static List<Suggestion> suggestUnionAll(RelMetadataQuery mq, List<RelNode> inputs) {
+ Set<Suggestion> suggestions = new HashSet<>();
+
+ suggestions.add(new Suggestion(random(), random()));
+ suggestions.add(new Suggestion(single(), single()));
+ suggestions.add(new Suggestion(broadcast(), broadcast()));
+
+ for (RelNode input : inputs) {
+ for (IgniteDistribution distribution : IgniteMdDerivedDistribution._deriveDistributions(input, mq)) {
+ if (distribution.getType() == RelDistribution.Type.HASH_DISTRIBUTED)
+ suggestions.add(new Suggestion(distribution, distribution));
+ }
+ }
+
+ return new ArrayList<>(suggestions);
}
/**
* Suggests possible aggregate distributions.
*
* @param mq Metadata query.
- * @param input Input node.
+ * @param input Input rel.
* @param groupSet Aggregate group set.
* @param groupSets Aggregate group sets.
* @return Array of possible distributions, sorted by their efficiency (cheaper first).
*/
public static List<Suggestion> suggestAggregate(RelMetadataQuery mq, RelNode input, ImmutableBitSet groupSet,
List<ImmutableBitSet> groupSets) {
- List<IgniteDistribution> inDistrs = IgniteMdDerivedDistribution._deriveDistributions(input, mq);
+ Set<Suggestion> suggestions = new HashSet<>();
- Map<Suggestion, Integer> suggestions = new HashMap<>();
+ if (!groupSet.isEmpty() && simpleAggregate(groupSet, groupSets))
+ processHashAgg(suggestions, IgniteMdDerivedDistribution._deriveDistributions(input, mq), groupSet);
- for (IgniteDistribution inDistr : inDistrs)
- suggestions = suggestAggregate0(suggestions, inDistr, groupSet, groupSets);
+ suggestions.add(new Suggestion(single(), random()));
+ suggestions.add(new Suggestion(broadcast(), broadcast()));
+ suggestions.add(new Suggestion(single(), single()));
- return sorted(suggestions);
+ return new ArrayList<>(suggestions);
}
/**
- * Suggests possible aggregate distributions based on next principles:
- * <li>Simple aggregate may be done in a distributed form in case its input is hash distributed by aggregate group keys</li>
- * <li>Any aggregate may be done without map-reduce processing in case its input is singleton or broadcast distributed</li>
- * <li>Any aggregate may be done in map-reduce style in case its input is hash or random distributed</li>
+ * Suggests possible aggregate distributions.
+ *
+ * @param inDistrs Input node distributions.
+ * @param groupSet Aggregate group set.
+ * @param groupSets Aggregate group sets.
+ * @return Array of possible distributions, sorted by their efficiency (cheaper first).
*/
- private static Map<Suggestion,Integer> suggestAggregate0(Map<Suggestion, Integer> dst, IgniteDistribution inDistr,
- ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets) {
+ public static List<Suggestion> suggestAggregate(List<IgniteDistribution> inDistrs, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets) {
+ Set<Suggestion> suggestions = new HashSet<>();
- IgniteDistribution out, in;
+ if (!groupSet.isEmpty() && simpleAggregate(groupSet, groupSets))
+ processHashAgg(suggestions, inDistrs, groupSet);
- if (simpleAggregate(groupSet, groupSets) && !groupSet.isEmpty()) {
- // re-hash by group keys
- in = hash(groupSet.asList());
- out = hash(ImmutableIntList.range(0, groupSet.cardinality()));
- dst = add(dst, out, inDistr, in);
- }
+ suggestions.add(new Suggestion(single(), single()));
+ suggestions.add(new Suggestion(broadcast(), broadcast()));
- if (inDistr.satisfies(random()))
- // map-reduce case
- dst = add(dst, new Suggestion(single(), inDistr), BEST_CNT > 0 ? INTS[1] : INTS[0]);
+ // Map-reduce cases
+ suggestions.add(new Suggestion(single(), random()));
+ suggestions.add(new Suggestion(broadcast(), random()));
- out = broadcast(); in = broadcast();
- dst = add(dst, out, inDistr, in);
+ return new ArrayList<>(suggestions);
+ }
- out = single(); in = single();
- dst = add(dst, out, inDistr, in);
+ /** */
+ private static void processHashAgg(Set<Suggestion> dst, List<IgniteDistribution> inDistrs, ImmutableBitSet groupSet) {
+ int cardinality = groupSet.cardinality();
- return dst;
- }
+ // Here we check whether all distribution keys contain in group set
+ for (IgniteDistribution in : inDistrs) {
+ if (in.getType() != RelDistribution.Type.HASH_DISTRIBUTED)
+ continue;
- /**
- * Suggests possible join distributions.
- *
- * @param mq Metadata query.
- * @param left Left node.
- * @param right Right node.
- * @param joinInfo Join info.
- * @param joinType Join type.
- * @return Array of possible distributions, sorted by their efficiency (cheaper first).
- */
- public static List<BiSuggestion> suggestJoin(RelMetadataQuery mq, RelNode left, RelNode right, JoinInfo joinInfo,
- JoinRelType joinType) {
- List<IgniteDistribution> leftIn = IgniteMdDerivedDistribution._deriveDistributions(left, mq);
- List<IgniteDistribution> rightIn = IgniteMdDerivedDistribution._deriveDistributions(right, mq);
+ ImmutableIntList keys = in.getKeys();
- Map<BiSuggestion, Integer> suggestions = new HashMap<>();
+ if (keys.size() != cardinality)
+ continue;
- for (IgniteDistribution leftIn0 : leftIn)
- for (IgniteDistribution rightIn0 : rightIn)
- suggestions = suggestJoin0(suggestions, leftIn0, rightIn0, joinInfo, joinType);
+ Mappings.TargetMapping mapping = IgniteAggregate.partialMapping(
+ Math.max(Commons.max(keys), cardinality), cardinality, groupSet);
- return sorted(suggestions);
- }
+ IgniteDistribution out = in.apply(mapping);
- /**
- * Suggests possible join distributions.
- *
- * @param leftIn Left distribution.
- * @param rightIn Right distribution.
- * @param joinInfo Join info.
- * @param joinType Join type.
- * @return Array of possible distributions, sorted by their efficiency (cheaper first).
- */
- public static List<BiSuggestion> suggestJoin(IgniteDistribution leftIn, IgniteDistribution rightIn,
- JoinInfo joinInfo, JoinRelType joinType) {
- return sorted(suggestJoin0(new HashMap<>(), leftIn, rightIn, joinInfo, joinType));
+ if (out.getType() != RelDistribution.Type.RANDOM_DISTRIBUTED)
+ dst.add(new Suggestion(out, in));
+ }
+
+ IgniteDistribution in = hash(groupSet.asList());
+ IgniteDistribution out = hash(ImmutableIntList.range(0, cardinality));
+ dst.add(new Suggestion(out, in));
}
/**
- * Suggests possible join distributions using inputs distributions and join info
- * based on next distributions table:
+ * Suggests possible join distributions based on next table:
* <p/><table>
* <tr><th>===============INNER JOIN==============</th></tr>
* <tr><td>hash + hash = hash</td></tr>
@@ -273,149 +259,64 @@ public class IgniteDistributions {
* <tr><td>broadcast + broadcast = broadcast</td></tr>
* <tr><td>single + single = single</td></tr>
* </table>
- * <p/>others require redistribution.
+ *
+ * @param mq Metadata query.
+ * @param left Left node.
+ * @param right Right node.
+ * @param joinInfo Join info.
+ * @param joinType Join type.
+ * @return Array of possible distributions.
*/
- private static Map<BiSuggestion, Integer> suggestJoin0(Map<BiSuggestion, Integer> dst, IgniteDistribution leftIn,
- IgniteDistribution rightIn, JoinInfo joinInfo, JoinRelType joinType) {
- IgniteDistribution out, left, right;
+ public static List<BiSuggestion> suggestJoin(RelMetadataQuery mq, RelNode left, RelNode right, JoinInfo joinInfo,
+ JoinRelType joinType) {
+ Set<BiSuggestion> suggestions = new HashSet<>();
+
+ suggestions.add(new BiSuggestion(broadcast(), broadcast(), broadcast()));
+ suggestions.add(new BiSuggestion(single(), single(), single()));
if (joinType == LEFT || joinType == RIGHT || (joinType == INNER && !F.isEmpty(joinInfo.pairs()))) {
- HashSet<DistributionFunction> factories = U.newHashSet(3);
+ Set<DistributionFunction> functions = new HashSet<>();
+
+ for (IgniteDistribution leftIn : IgniteMdDerivedDistribution._deriveDistributions(left, mq)) {
+ if (leftIn.getType() == RelDistribution.Type.HASH_DISTRIBUTED
+ && Objects.equals(joinInfo.leftKeys, leftIn.getKeys()))
+ functions.add(leftIn.function());
+ }
- if (leftIn.getType() == RelDistribution.Type.HASH_DISTRIBUTED
- && Objects.equals(joinInfo.leftKeys, leftIn.getKeys()))
- factories.add(leftIn.function());
+ for (IgniteDistribution rightIn : IgniteMdDerivedDistribution._deriveDistributions(right, mq)) {
+ if (rightIn.getType() == RelDistribution.Type.HASH_DISTRIBUTED
+ && Objects.equals(joinInfo.rightKeys, rightIn.getKeys()))
+ functions.add(rightIn.function());
+ }
- if (rightIn.getType() == RelDistribution.Type.HASH_DISTRIBUTED
- && Objects.equals(joinInfo.rightKeys, rightIn.getKeys()))
- factories.add(rightIn.function());
+ functions.add(HashDistribution.INSTANCE);
- factories.add(HashDistribution.INSTANCE);
+ IgniteDistribution out, left0, right0;
- for (DistributionFunction factory : factories) {
+ for (DistributionFunction factory : functions) {
out = hash(joinInfo.leftKeys, factory);
+ left0 = hash(joinInfo.leftKeys, factory);
+ right0 = hash(joinInfo.rightKeys, factory);
- left = hash(joinInfo.leftKeys, factory); right = hash(joinInfo.rightKeys, factory);
- dst = add(dst, out, leftIn, rightIn, left, right);
+ suggestions.add(new BiSuggestion(out, left0, right0));
if (joinType == INNER || joinType == LEFT) {
- left = hash(joinInfo.leftKeys, factory); right = broadcast();
- dst = add(dst, out, leftIn, rightIn, left, right);
- }
+ left0 = hash(joinInfo.leftKeys, factory);
+ right0 = broadcast();
- if (joinType == INNER || joinType == RIGHT) {
- left = broadcast(); right = hash(joinInfo.rightKeys, factory);
- dst = add(dst, out, leftIn, rightIn, left, right);
+ suggestions.add(new BiSuggestion(out, left0, right0));
}
- }
- }
-
- out = left = right = broadcast();
- dst = add(dst, out, leftIn, rightIn, left, right);
-
- out = left = right = single();
- dst = add(dst, out, leftIn, rightIn, left, right);
-
- return dst;
- }
-
- /** */
- private static Map<Suggestion, Integer> add(Map<Suggestion, Integer> dst, IgniteDistribution out,
- IgniteDistribution in, IgniteDistribution newIn) {
- if (BEST_CNT > 0) {
- int exch = 0;
-
- if (!in.satisfies(newIn))
- exch++;
-
- return add(dst, new Suggestion(out, newIn), INTS[exch]);
- }
- else
- return add(dst, new Suggestion(out, newIn), INTS[0]);
- }
-
- /** */
- private static Map<BiSuggestion, Integer> add(Map<BiSuggestion, Integer> dst, IgniteDistribution out,
- IgniteDistribution left, IgniteDistribution right, IgniteDistribution newLeft, IgniteDistribution newRight) {
- if (BEST_CNT > 0) {
- int exch = 0;
-
- if (!left.satisfies(newLeft))
- exch++;
-
- if (!right.satisfies(newRight))
- exch++;
-
- return add(dst, new BiSuggestion(out, newLeft, newRight), INTS[exch]);
- }
- else
- return add(dst, new BiSuggestion(out, newLeft, newRight), INTS[0]);
- }
-
- /** */
- private static <T> Map<T, Integer> add(Map<T, Integer> dst, T suggest, Integer exchCnt) {
- if (dst == null)
- dst = new HashMap<>();
-
- dst.merge(suggest, exchCnt, IgniteDistributions::min);
-
- return dst;
- }
-
- /** */
- private static <T> List<T> sorted(Map<T, Integer> src) {
- if (BEST_CNT > 0) {
- List<Map.Entry<T, Integer>> entries = new ArrayList<>(src.entrySet());
-
- entries.sort(Map.Entry.comparingByValue());
-
- if (entries.size() >= BEST_CNT)
- entries = entries.subList(0, BEST_CNT);
-
- return Commons.transform(entries, Map.Entry::getKey);
- }
-
- return new ArrayList<>(src.keySet());
- }
-
- /** */
- private static @NotNull Integer min(@NotNull Integer i1, @NotNull Integer i2) {
- return i1 < i2 ? i1 : i2;
- }
-
- /**
- * Projects distribution keys using target mapping.
- * Returns empty collection in case any of distribution keys is lost.
- *
- * @param mapping Target mapping.
- * @param keys Distribution keys.
- * @return New distribution keys.
- */
- public static ImmutableIntList projectDistributionKeys(Mappings.TargetMapping mapping, ImmutableIntList keys) {
- if (mapping.getTargetCount() < keys.size())
- return ImmutableIntList.of();
- int[] resKeys = new int[keys.size()];
-
- for (int i = 0; i < keys.size(); i++) {
- boolean found = false;
- int key = keys.getInt(i);
-
- for (int j = 0; j < mapping.getTargetCount(); j++) {
- if (mapping.getSourceOpt(j) != key)
- continue;
-
- found = true;
- resKeys[i] = j;
+ if (joinType == INNER || joinType == RIGHT) {
+ left0 = broadcast();
+ right0 = hash(joinInfo.rightKeys, factory);
- break;
+ suggestions.add(new BiSuggestion(out, left0, right0));
+ }
}
-
- if (!found)
- return ImmutableIntList.of();
}
- return ImmutableIntList.of(resKeys);
+ return new ArrayList<>(suggestions);
}
/**
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 f66fb54..f6e9a19 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
@@ -37,6 +37,7 @@ import org.apache.calcite.rel.externalize.RelWriterImpl;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridKernalContext;
@@ -252,6 +253,32 @@ public final class Commons {
}
/** */
+ public static int max(ImmutableIntList list) {
+ if (list.isEmpty())
+ throw new UnsupportedOperationException();
+
+ int res = list.getInt(0);
+
+ for (int i = 1; i < list.size(); i++)
+ res = Math.max(res, list.getInt(i));
+
+ return res;
+ }
+
+ /** */
+ public static int min(ImmutableIntList list) {
+ if (list.isEmpty())
+ throw new UnsupportedOperationException();
+
+ int res = list.getInt(0);
+
+ for (int i = 1; i < list.size(); i++)
+ res = Math.min(res, list.getInt(i));
+
+ return res;
+ }
+
+ /** */
public static RelDataType aggregationDataRowType(RelDataTypeFactory typeFactory) {
assert typeFactory instanceof IgniteTypeFactory;
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 aef2231..64ab800 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
@@ -22,6 +22,7 @@ import java.util.Arrays;
import java.util.List;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.affinity.AffinityKeyMapped;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
@@ -56,6 +57,100 @@ public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
}
@Test
+ public void unionAll() throws Exception {
+ IgniteCache<Integer, Employer> employer1 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer1")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER1")))
+ .setBackups(1)
+ );
+
+ IgniteCache<Integer, Employer> employer2 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer2")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER2")))
+ .setBackups(2)
+ );
+
+ IgniteCache<Integer, Employer> employer3 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer3")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER3")))
+ .setBackups(3)
+ );
+
+ employer1.put(1, new Employer("Igor", 10d));
+ employer2.put(1, new Employer("Roman", 15d));
+ employer3.put(1, new Employer("Nikolay", 20d));
+ employer1.put(2, new Employer("Igor", 10d));
+ employer2.put(2, new Employer("Roman", 15d));
+ employer3.put(3, new Employer("Nikolay", 20d));
+
+ waitForReadyTopology(internalCache(employer1).context().topology(), new AffinityTopologyVersion(5, 4));
+
+ QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
+
+ List<FieldsQueryCursor<List<?>>> query = engine.query(null, "PUBLIC",
+ "SELECT * FROM employer1 " +
+ "UNION ALL " +
+ "SELECT * FROM employer2 " +
+ "UNION ALL " +
+ "SELECT * FROM employer3 ");
+
+ assertEquals(1, query.size());
+
+ List<List<?>> rows = query.get(0).getAll();
+ assertEquals(6, rows.size());
+ }
+
+ @Test
+ public void union() throws Exception {
+ IgniteCache<Integer, Employer> employer1 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer1")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER1")))
+ .setBackups(1)
+ );
+
+ IgniteCache<Integer, Employer> employer2 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer2")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER2")))
+ .setBackups(2)
+ );
+
+ IgniteCache<Integer, Employer> employer3 = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer3")
+ .setSqlSchema("PUBLIC")
+ .setQueryEntities(F.asList(new QueryEntity(Integer.class, Employer.class).setTableName("EMPLOYER3")))
+ .setBackups(3)
+ );
+
+ employer1.put(1, new Employer("Igor", 10d));
+ employer2.put(1, new Employer("Roman", 15d));
+ employer3.put(1, new Employer("Nikolay", 20d));
+ employer1.put(2, new Employer("Igor", 10d));
+ employer2.put(2, new Employer("Roman", 15d));
+ employer3.put(3, new Employer("Nikolay", 20d));
+
+ waitForReadyTopology(internalCache(employer1).context().topology(), new AffinityTopologyVersion(5, 4));
+
+ QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
+
+ List<FieldsQueryCursor<List<?>>> query = engine.query(null, "PUBLIC",
+ "SELECT * FROM employer1 " +
+ "UNION " +
+ "SELECT * FROM employer2 " +
+ "UNION " +
+ "SELECT * FROM employer3 ");
+
+ assertEquals(1, query.size());
+
+ List<List<?>> rows = query.get(0).getAll();
+ assertEquals(3, rows.size());
+ }
+
+ @Test
public void aggregate() throws Exception {
IgniteCache<Integer, Employer> employer = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
.setName("employer")
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 d47b4f0..0896669 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
@@ -368,6 +368,288 @@ public class PlannerTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
@Test
+ public void testUnion() throws Exception {
+ IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
+
+ TestTable table1 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table1", "hash");
+ }
+ };
+
+ TestTable table2 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table2", "hash");
+ }
+ };
+
+ TestTable table3 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table3", "hash");
+ }
+ };
+
+ IgniteSchema publicSchema = new IgniteSchema("PUBLIC");
+
+ publicSchema.addTable("TABLE1", table1);
+ publicSchema.addTable("TABLE2", table2);
+ publicSchema.addTable("TABLE3", table3);
+
+ SchemaPlus schema = createRootSchema(false)
+ .add("PUBLIC", publicSchema);
+
+ String sql = "" +
+ "SELECT * FROM table1 " +
+ "UNION " +
+ "SELECT * FROM table2 " +
+ "UNION " +
+ "SELECT * FROM table3 ";
+
+ 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)
+ .parameters(new Object[]{2})
+ .topologyVersion(AffinityTopologyVersion.NONE)
+ .build();
+
+ RelNode root;
+
+ 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
+ root = planner.convert(sqlNode);
+
+ // Transformation chain
+ root = planner.transform(PlannerPhase.HEURISTIC_OPTIMIZATION, root.getTraitSet(), root);
+
+ // Transformation chain
+ RelTraitSet desired = root.getCluster().traitSet()
+ .replace(IgniteConvention.INSTANCE)
+ .replace(IgniteDistributions.single())
+ .simplify();
+
+ root = planner.transform(PlannerPhase.OPTIMIZATION, desired, root);
+ }
+
+ assertNotNull(root);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testUnionAll() throws Exception {
+ IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
+
+ TestTable table1 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table1", "hash");
+ }
+ };
+
+ TestTable table2 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table2", "hash");
+ }
+ };
+
+ TestTable table3 = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Table3", "hash");
+ }
+ };
+
+ IgniteSchema publicSchema = new IgniteSchema("PUBLIC");
+
+ publicSchema.addTable("TABLE1", table1);
+ publicSchema.addTable("TABLE2", table2);
+ publicSchema.addTable("TABLE3", table3);
+
+ SchemaPlus schema = createRootSchema(false)
+ .add("PUBLIC", publicSchema);
+
+ String sql = "" +
+ "SELECT * FROM table1 " +
+ "UNION ALL " +
+ "SELECT * FROM table2 " +
+ "UNION ALL " +
+ "SELECT * FROM table3 ";
+
+ 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)
+ .parameters(new Object[]{2})
+ .topologyVersion(AffinityTopologyVersion.NONE)
+ .build();
+
+ RelNode root;
+
+ 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
+ root = planner.convert(sqlNode);
+
+ // Transformation chain
+ root = planner.transform(PlannerPhase.HEURISTIC_OPTIMIZATION, root.getTraitSet(), root);
+
+ // Transformation chain
+ RelTraitSet desired = root.getCluster().traitSet()
+ .replace(IgniteConvention.INSTANCE)
+ .replace(IgniteDistributions.single())
+ .simplify();
+
+ root = planner.transform(PlannerPhase.OPTIMIZATION, desired, root);
+ }
+
+ assertNotNull(root);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
public void testAggregate() throws Exception {
IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
index 6b98fda..86430f5 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.query.calcite.exec.rel;
import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import java.util.UUID;
import org.apache.calcite.rel.RelCollations;
import org.apache.calcite.rel.core.AggregateCall;
@@ -106,6 +107,49 @@ public class ExecutionTest extends AbstractExecutionTest {
}
@Test
+ public void testUnionAll() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan1 = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ ScanNode scan2 = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ ScanNode scan3 = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ UnionAllNode<Object[]> union = new UnionAllNode<>(ctx);
+ union.register(F.asList(scan1, scan2, scan3));
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(union);
+
+ assertTrue(root.hasNext());
+
+ List<Object[]> res = new ArrayList<>();
+
+ while (root.hasNext())
+ res.add(root.next());
+
+ assertEquals(12, res.size());
+ }
+
+ @Test
public void testAggregateMapReduceAvg() throws Exception {
ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);