This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/logical_to_distributed
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to
refs/heads/xingtanzjr/logical_to_distributed by this push:
new c217c74 complete exchange node add
c217c74 is described below
commit c217c74218a75b3605917fc9ae1a26e47f537727
Author: Jinrui.Zhang <[email protected]>
AuthorDate: Sun Mar 20 20:57:42 2022 +0800
complete exchange node add
---
.../org/apache/iotdb/db/mpp/common/DataRegion.java | 8 ++
.../mpp/sql/planner/plan/DistributionPlanner.java | 139 ++++++++++++++++++++-
.../db/mpp/sql/planner/plan/node/PlanNodeId.java | 11 ++
.../planner/plan/node/process/ExchangeNode.java | 80 ++++++++++++
.../plan/node/source/SeriesAggregateScanNode.java | 8 ++
.../planner/plan/node/source/SeriesScanNode.java | 8 +-
.../db/mpp/sql/plan/DistributionPlannerTest.java | 39 +++++-
7 files changed, 283 insertions(+), 10 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/common/DataRegion.java
b/server/src/main/java/org/apache/iotdb/db/mpp/common/DataRegion.java
index 04100e1..81c9665 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/DataRegion.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/DataRegion.java
@@ -47,4 +47,12 @@ public class DataRegion {
public String toString() {
return String.format("%s/%d", this.endpoint, this.dataRegionId);
}
+
+ public Integer getDataRegionId() {
+ return dataRegionId;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/DistributionPlanner.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/DistributionPlanner.java
index 9b146d5..08c8045 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/DistributionPlanner.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/DistributionPlanner.java
@@ -20,14 +20,16 @@ package org.apache.iotdb.db.mpp.sql.planner.plan;
import org.apache.iotdb.db.mpp.common.Analysis;
import org.apache.iotdb.db.mpp.common.DataRegion;
-import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.mpp.sql.planner.plan.node.SimplePlanNodeRewriter;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.*;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.TimeJoinNode;
import
org.apache.iotdb.db.mpp.sql.planner.plan.node.source.SeriesAggregateScanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.source.SeriesScanNode;
import java.util.*;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
public class DistributionPlanner {
private Analysis analysis;
private LogicalQueryPlan logicalPlan;
@@ -42,11 +44,22 @@ public class DistributionPlanner {
return rewriter.visit(logicalPlan.getRootNode(), new
DistributionPlanContext());
}
+ public PlanNode addExchangeNode(PlanNode root) {
+ ExchangeNodeAdder adder = new ExchangeNodeAdder();
+ return adder.visit(root, new NodeGroupContext());
+ }
+
public DistributedQueryPlan planFragments() {
return null;
}
private class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanContext> {
+
+ //TODO: (xingtanzjr) implement the method visitDeviceMergeNode()
+ public PlanNode visitDeviceMerge(TimeJoinNode node,
DistributionPlanContext context) {
+ return null;
+ }
+
public PlanNode visitTimeJoin(TimeJoinNode node, DistributionPlanContext
context) {
TimeJoinNode root = (TimeJoinNode) node.clone();
@@ -115,5 +128,125 @@ public class DistributionPlanner {
}
}
- private class DistributionPlanContext {}
+ private class DistributionPlanContext {
+
+ }
+
+ private class ExchangeNodeAdder extends PlanVisitor<PlanNode,
NodeGroupContext>{
+ @Override
+ public PlanNode visitPlan(PlanNode node, NodeGroupContext context) {
+ // Visit all the children of current node
+ List<PlanNode> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(toImmutableList());
+
+ // Calculate the node distribution info according to its children
+
+ // Put the node distribution info into context
+ // NOTICE: we will only process the PlanNode which has only 1 child
here. For the other PlanNode, we need to process
+ // them with special method
+ context.putNodeDistribution(node.getId(), new
NodeDistribution(NodeDistributionType.SAME_WITH_ALL_CHILDREN, null));
+
+ return node.cloneWithChildren(children);
+ }
+
+ public PlanNode visitSeriesScan(SeriesScanNode node, NodeGroupContext
context) {
+ context.putNodeDistribution(node.getId(), new
NodeDistribution(NodeDistributionType.NO_CHILD, node.getDataRegion()));
+ return node.clone();
+ }
+
+ public PlanNode visitSeriesAggregate(SeriesAggregateScanNode node,
NodeGroupContext context) {
+ context.putNodeDistribution(node.getId(), new
NodeDistribution(NodeDistributionType.NO_CHILD, node.getDataRegion()));
+ return node.clone();
+ }
+
+ public PlanNode visitTimeJoin(TimeJoinNode node, NodeGroupContext context)
{
+ TimeJoinNode newNode = (TimeJoinNode) node.clone();
+ List<PlanNode> visitedChildren = new ArrayList<>();
+ node.getChildren().forEach(child -> {
+ visitedChildren.add(visit(child, context));
+ });
+
+
+ DataRegion dataRegion = calculateDataRegionByChildren(visitedChildren,
context);
+ NodeDistributionType distributionType =
nodeDistributionIsSame(visitedChildren, context) ?
+ NodeDistributionType.SAME_WITH_ALL_CHILDREN :
NodeDistributionType.SAME_WITH_SOME_CHILD;
+ context.putNodeDistribution(newNode.getId(), new
NodeDistribution(distributionType, dataRegion));
+
+ // If the distributionType of all the children are same, no ExchangeNode
need to be added.
+ if (distributionType == NodeDistributionType.SAME_WITH_ALL_CHILDREN) {
+ newNode.setChildren(visitedChildren);
+ return newNode;
+ }
+
+ // Otherwise, we need to add ExchangeNode for the child whose DataRegion
is different from the parent.
+ visitedChildren.forEach(child -> {
+ if
(!dataRegion.equals(context.getNodeDistribution(child.getId()).dataRegion)) {
+ ExchangeNode exchangeNode = new
ExchangeNode(PlanNodeAllocator.generateId());
+ exchangeNode.setSourceNode(child);
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(child);
+ }
+ });
+ return newNode;
+ }
+
+ private DataRegion calculateDataRegionByChildren(List<PlanNode> children,
NodeGroupContext context) {
+ // We always make the dataRegion of TimeJoinNode to be the same as its
first child.
+ // TODO: (xingtanzjr) We need to implement more suitable policies here
+ DataRegion childDataRegion =
context.getNodeDistribution(children.get(0).getId()).dataRegion;
+ return new DataRegion(childDataRegion.getDataRegionId(),
childDataRegion.getEndpoint());
+ }
+
+ private boolean nodeDistributionIsSame(List<PlanNode> children,
NodeGroupContext context) {
+ // The size of children here should always be larger than 0, or our code
has Bug.
+ NodeDistribution first =
context.getNodeDistribution(children.get(0).getId());
+ for (int i = 1 ; i < children.size() ; i ++) {
+ NodeDistribution next =
context.getNodeDistribution(children.get(i).getId());
+ if (first.dataRegion == null ||
!first.dataRegion.equals(next.dataRegion)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public PlanNode visit(PlanNode node, NodeGroupContext context) {
+ return node.accept(this, context);
+ }
+
+ }
+
+ private class NodeGroupContext {
+ Map<PlanNodeId, NodeDistribution> nodeDistribution;
+ public NodeGroupContext() {
+ nodeDistribution = new HashMap<>();
+ }
+
+ public void putNodeDistribution(PlanNodeId nodeId, NodeDistribution
distribution) {
+ this.nodeDistribution.put(nodeId, distribution);
+ }
+
+ public NodeDistribution getNodeDistribution(PlanNodeId nodeId) {
+ return this.nodeDistribution.get(nodeId);
+ }
+ }
+
+ private enum NodeDistributionType {
+ SAME_WITH_ALL_CHILDREN,
+ SAME_WITH_SOME_CHILD,
+ DIFFERENT_FROM_ALL_CHILDREN,
+ NO_CHILD,
+ }
+
+ private class NodeDistribution {
+ private NodeDistributionType type;
+ private DataRegion dataRegion;
+
+ private NodeDistribution(NodeDistributionType type, DataRegion dataRegion)
{
+ this.type = type;
+ this.dataRegion = dataRegion;
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeId.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeId.java
index f829dfb..4029ddb 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeId.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/PlanNodeId.java
@@ -33,4 +33,15 @@ public class PlanNodeId {
public String toString() {
return this.id;
}
+
+ public int hashCode() {
+ return this.id.hashCode();
+ }
+
+ public boolean equals(Object obj) {
+ if (obj instanceof PlanNodeId) {
+ return this.id.equals(((PlanNodeId)obj).getId());
+ }
+ return false;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/ExchangeNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/ExchangeNode.java
new file mode 100644
index 0000000..60f172b
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/ExchangeNode.java
@@ -0,0 +1,80 @@
+/*
+ * 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.iotdb.db.mpp.sql.planner.plan.node.process;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.iotdb.db.mpp.common.FragmentId;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ExchangeNode extends PlanNode {
+ private PlanNode sourceNode;
+ private FragmentId sourceFragmentId;
+
+ public ExchangeNode(PlanNodeId id) {
+ super(id);
+ }
+
+ @Override
+ public List<PlanNode> getChildren() {
+ return ImmutableList.of(sourceNode);
+ }
+
+ @Override
+ public PlanNode clone() {
+ return new ExchangeNode(getId());
+ }
+
+ @Override
+ public PlanNode cloneWithChildren(List<PlanNode> children) {
+ ExchangeNode node = new ExchangeNode(getId());
+ node.setSourceNode(children.get(0));
+ return node;
+ }
+
+ @Override
+ public List<String> getOutputColumnNames() {
+ return null;
+ }
+
+ public void setSourceFragmentId(FragmentId sourceFragmentId) {
+ this.sourceFragmentId = sourceFragmentId;
+ }
+
+ public FragmentId getSourceFragmentId() {
+ return sourceFragmentId;
+ }
+
+ public PlanNode getSourceNode() {
+ return sourceNode;
+ }
+
+ public void setSourceNode(PlanNode sourceNode) {
+ this.sourceNode = sourceNode;
+ }
+
+ public String toString() {
+ return String.format("ExchangeNode-%s", getId());
+ }
+
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesAggregateScanNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesAggregateScanNode.java
index d071fd9..674b6b4 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesAggregateScanNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesAggregateScanNode.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.mpp.sql.planner.plan.node.source;
+import org.apache.iotdb.db.mpp.common.DataRegion;
import org.apache.iotdb.db.mpp.common.GroupByTimeParameter;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId;
@@ -59,6 +60,9 @@ public class SeriesAggregateScanNode extends SourceNode {
private String columnName;
+ // The id of DataRegion where the node will run
+ private DataRegion dataRegion;
+
public SeriesAggregateScanNode(PlanNodeId id) {
super(id);
}
@@ -100,6 +104,10 @@ public class SeriesAggregateScanNode extends SourceNode {
@Override
public void close() throws Exception {}
+ public DataRegion getDataRegion() {
+ return dataRegion;
+ }
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitSeriesAggregate(this, context);
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesScanNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesScanNode.java
index 628d4da..571034b 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesScanNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/source/SeriesScanNode.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import com.google.common.collect.ImmutableList;
+import javax.xml.crypto.Data;
import java.util.List;
/**
@@ -70,6 +71,11 @@ public class SeriesScanNode extends SourceNode {
this.seriesPath = seriesPath;
}
+ public SeriesScanNode(PlanNodeId id, PartialPath seriesPath, DataRegion
dataRegion) {
+ this(id, seriesPath);
+ this.dataRegion = dataRegion;
+ }
+
public void setTimeFilter(Filter timeFilter) {
this.timeFilter = timeFilter;
}
@@ -103,7 +109,7 @@ public class SeriesScanNode extends SourceNode {
@Override
public PlanNode clone() {
- return new SeriesScanNode(PlanNodeAllocator.generateId(),
this.getSeriesPath());
+ return new SeriesScanNode(getId(), getSeriesPath(), this.dataRegion);
}
@Override
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
index 0d9c798..4c6f27a 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.db.mpp.sql.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeAllocator;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeUtil;
+import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.process.TimeJoinNode;
import org.apache.iotdb.db.mpp.sql.planner.plan.node.source.SeriesScanNode;
@@ -40,6 +41,7 @@ import java.util.List;
import java.util.Map;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
public class DistributionPlannerTest {
@@ -64,17 +66,42 @@ public class DistributionPlannerTest {
new DistributionPlanner(analysis, new LogicalQueryPlan(new
MPPQueryContext(), root));
PlanNode newRoot = planner.rewriteSource();
- System.out.println("\nLogical-Plan:");
- System.out.println("------------------");
- PlanNodeUtil.printPlanNode(root);
- System.out.println("\nDistributed-Plan:");
- System.out.println("------------------");
- PlanNodeUtil.printPlanNode(newRoot);
+// PlanNodeUtil.printPlanNode(newRoot);
assertEquals(newRoot.getChildren().get(0).getChildren().size(), 3);
assertEquals(newRoot.getChildren().get(0).getChildren().get(0).getChildren().size(),
2);
assertEquals(newRoot.getChildren().get(0).getChildren().get(1).getChildren().size(),
2);
}
+ @Test
+ public void TestAddExchangeNode() throws IllegalPathException {
+ TimeJoinNode timeJoinNode =
+ new TimeJoinNode(
+ PlanNodeAllocator.generateId(), OrderBy.TIMESTAMP_ASC,
FilterNullPolicy.NO_FILTER);
+
+ timeJoinNode.addChild(
+ new SeriesScanNode(PlanNodeAllocator.generateId(), new
PartialPath("root.sg.d1.s1")));
+ timeJoinNode.addChild(
+ new SeriesScanNode(PlanNodeAllocator.generateId(), new
PartialPath("root.sg.d1.s2")));
+ timeJoinNode.addChild(
+ new SeriesScanNode(PlanNodeAllocator.generateId(), new
PartialPath("root.sg.d2.s1")));
+
+ LimitNode root = new LimitNode(PlanNodeAllocator.generateId(), 10,
timeJoinNode);
+
+ Analysis analysis = constructAnalysis();
+
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(new
MPPQueryContext(), root));
+ PlanNode rootAfterRewrite = planner.rewriteSource();
+ PlanNode rootWithExchange = planner.addExchangeNode(rootAfterRewrite);
+// PlanNodeUtil.printPlanNode(rootWithExchange);
+ assertEquals(rootWithExchange.getChildren().get(0).getChildren().size(),
3);
+
assertEquals(rootWithExchange.getChildren().get(0).getChildren().get(0).getChildren().size(),
2);
+ assertTrue(rootWithExchange.getChildren().get(0).getChildren().get(1)
instanceof ExchangeNode);
+
assertEquals(rootWithExchange.getChildren().get(0).getChildren().get(1).getChildren().size(),
1);
+ assertTrue(rootWithExchange.getChildren().get(0).getChildren().get(2)
instanceof ExchangeNode);
+
assertEquals(rootWithExchange.getChildren().get(0).getChildren().get(2).getChildren().size(),
1);
+ }
+
private Analysis constructAnalysis() {
Analysis analysis = new Analysis();
Map<String, Map<DataRegionTimeSlice, List<DataRegion>>> dataPartitionInfo
= new HashMap<>();
