This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 982140c9e7 Distribution plan for Align by device and GroupByLevel
(#6053)
982140c9e7 is described below
commit 982140c9e759756ddec974b393b180122aa795fa
Author: Zhang.Jinrui <[email protected]>
AuthorDate: Sat May 28 12:37:30 2022 +0800
Distribution plan for Align by device and GroupByLevel (#6053)
---
.../apache/iotdb/db/mpp/plan/analyze/Analysis.java | 4 +
.../db/mpp/plan/planner/LocalExecutionPlanner.java | 38 ++++++++-
.../planner/distribution/ExchangeNodeAdder.java | 23 +++++-
.../plan/planner/distribution/SourceRewriter.java | 91 +++++++++++++++++++++-
.../planner/plan/node/process/DeviceMergeNode.java | 9 ++-
.../planner/plan/node/process/DeviceViewNode.java | 14 +++-
.../plan/node/process/GroupByLevelNode.java | 9 +++
.../query/reader/chunk/MemAlignedPageReader.java | 1 -
.../plan/distribution/AlignedByDeviceTest.java | 36 +++++++++
9 files changed, 212 insertions(+), 13 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
index 2e4eab4950..2a7137dd22 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
@@ -152,6 +152,10 @@ public class Analysis {
return dataPartition.getDataRegionReplicaSet(seriesPath.getDevice(), null);
}
+ public List<TRegionReplicaSet> getPartitionInfo(String deviceName, Filter
globalTimeFilter) {
+ return dataPartition.getDataRegionReplicaSet(deviceName, null);
+ }
+
public Statement getStatement() {
return statement;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
index 6587e05306..e2a3176515 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LocalExecutionPlanner.java
@@ -143,6 +143,7 @@ import
org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationSc
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
import
org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.FillDescriptor;
+import
org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByLevelDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.OutputColumn;
import org.apache.iotdb.db.mpp.plan.statement.component.FillPolicy;
@@ -785,7 +786,42 @@ public class LocalExecutionPlanner {
@Override
public Operator visitGroupByLevel(GroupByLevelNode node,
LocalExecutionPlanContext context) {
- return super.visitGroupByLevel(node, context);
+ checkArgument(
+ node.getGroupByLevelDescriptors().size() >= 1,
+ "GroupByLevel descriptorList cannot be empty");
+ List<Operator> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(Collectors.toList());
+ boolean ascending = node.getScanOrder() == OrderBy.TIMESTAMP_ASC;
+ List<Aggregator> aggregators = new ArrayList<>();
+ Map<String, List<InputLocation>> layout = makeLayout(node);
+ for (GroupByLevelDescriptor descriptor :
node.getGroupByLevelDescriptors()) {
+ List<String> inputColumnNames = descriptor.getInputColumnNames();
+ List<InputLocation[]> inputLocationList = new
ArrayList<>(inputColumnNames.size());
+ inputColumnNames.forEach(
+ inputColumnName ->
+ inputLocationList.add(layout.get(inputColumnName).toArray(new
InputLocation[0])));
+
+ aggregators.add(
+ new Aggregator(
+ AccumulatorFactory.createAccumulator(
+ descriptor.getAggregationType(),
+ context
+ .getTypeProvider()
+ // get the type of first inputExpression
+
.getType(descriptor.getInputExpressions().get(0).toString()),
+ ascending),
+ descriptor.getStep(),
+ inputLocationList));
+ }
+ OperatorContext operatorContext =
+ context.instanceContext.addOperatorContext(
+ context.getNextOperatorId(),
+ node.getPlanNodeId(),
+ AggregationOperator.class.getSimpleName());
+ return new AggregationOperator(
+ operatorContext, aggregators, children, ascending,
node.getGroupByTimeParameter());
}
@Override
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
index 6c502c464f..f85931231b 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
@@ -31,6 +31,8 @@ import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryM
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryScanNode;
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LastQueryMergeNode;
@@ -197,6 +199,16 @@ public class ExchangeNodeAdder extends
PlanVisitor<PlanNode, NodeGroupContext> {
return node;
}
+ @Override
+ public PlanNode visitDeviceView(DeviceViewNode node, NodeGroupContext
context) {
+ return processMultiChildNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitDeviceMerge(DeviceMergeNode node, NodeGroupContext
context) {
+ return processMultiChildNode(node, context);
+ }
+
@Override
public PlanNode visitLastQueryMerge(LastQueryMergeNode node,
NodeGroupContext context) {
return processMultiChildNode(node, context);
@@ -263,7 +275,16 @@ public class ExchangeNodeAdder extends
PlanVisitor<PlanNode, NodeGroupContext> {
children.stream()
.collect(
Collectors.groupingBy(
- child ->
context.getNodeDistribution(child.getPlanNodeId()).region,
+ child -> {
+ TRegionReplicaSet region =
+
context.getNodeDistribution(child.getPlanNodeId()).region;
+ if (region == null
+ &&
context.getNodeDistribution(child.getPlanNodeId()).type
+ == NodeDistributionType.SAME_WITH_ALL_CHILDREN) {
+ return
calculateSchemaRegionByChildren(child.getChildren(), context);
+ }
+ return region;
+ },
Collectors.counting()));
// Step 2: return the RegionReplicaSet with max count
return Collections.max(groupByRegion.entrySet(),
Map.Entry.comparingByValue()).getKey();
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
index 72c1819c80..903e60d481 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.partition.RegionReplicaSetInfo;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.schematree.PathPatternTree;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.expression.Expression;
@@ -35,6 +36,8 @@ import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryM
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryScanNode;
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
import
org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LastQueryMergeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
@@ -74,9 +77,91 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
this.analysis = analysis;
}
- // TODO: (xingtanzjr) implement the method visitDeviceMergeNode()
- public PlanNode visitDeviceMerge(TimeJoinNode node, DistributionPlanContext
context) {
- return null;
+ @Override
+ public PlanNode visitDeviceView(DeviceViewNode node, DistributionPlanContext
context) {
+ checkArgument(
+ node.getDevices().size() == node.getChildren().size(),
+ "size of devices and its children in DeviceViewNode should be same");
+
+ Set<TRegionReplicaSet> relatedDataRegions = new HashSet<>();
+
+ List<DeviceViewSplit> deviceViewSplits = new ArrayList<>();
+ // Step 1: constructs DeviceViewSplit
+ for (int i = 0; i < node.getDevices().size(); i++) {
+ String device = node.getDevices().get(i);
+ PlanNode child = node.getChildren().get(i);
+ List<TRegionReplicaSet> regionReplicaSets =
+ analysis.getPartitionInfo(device, analysis.getGlobalTimeFilter());
+ deviceViewSplits.add(new DeviceViewSplit(device, child,
regionReplicaSets));
+ relatedDataRegions.addAll(regionReplicaSets);
+ }
+
+ DeviceMergeNode deviceMergeNode =
+ new DeviceMergeNode(
+ context.queryContext.getQueryId().genPlanNodeId(),
+ node.getMergeOrders(),
+ node.getDevices());
+
+ // Step 2: Iterate all partition and create DeviceViewNode for each region
+ for (TRegionReplicaSet regionReplicaSet : relatedDataRegions) {
+ List<String> devices = new ArrayList<>();
+ List<PlanNode> children = new ArrayList<>();
+ for (DeviceViewSplit split : deviceViewSplits) {
+ if (split.needDistributeTo(regionReplicaSet)) {
+ devices.add(split.device);
+ children.add(split.buildPlanNodeInRegion(regionReplicaSet,
context.queryContext));
+ }
+ }
+ DeviceViewNode regionDeviceViewNode =
+ new DeviceViewNode(
+ context.queryContext.getQueryId().genPlanNodeId(),
+ node.getMergeOrders(),
+ node.getOutputColumnNames(),
+ node.getDeviceToMeasurementIndexesMap());
+ for (int i = 0; i < devices.size(); i++) {
+ regionDeviceViewNode.addChildDeviceNode(devices.get(i),
children.get(i));
+ }
+ deviceMergeNode.addChild(regionDeviceViewNode);
+ }
+
+ return deviceMergeNode;
+ }
+
+ private static class DeviceViewSplit {
+ protected String device;
+ protected PlanNode root;
+ protected Set<TRegionReplicaSet> dataPartitions;
+
+ protected DeviceViewSplit(
+ String device, PlanNode root, List<TRegionReplicaSet> dataPartitions) {
+ this.device = device;
+ this.root = root;
+ this.dataPartitions = new HashSet<>();
+ this.dataPartitions.addAll(dataPartitions);
+ }
+
+ protected PlanNode buildPlanNodeInRegion(
+ TRegionReplicaSet regionReplicaSet, MPPQueryContext context) {
+ return buildPlanNodeInRegion(this.root, regionReplicaSet, context);
+ }
+
+ protected boolean needDistributeTo(TRegionReplicaSet regionReplicaSet) {
+ return this.dataPartitions.contains(regionReplicaSet);
+ }
+
+ private PlanNode buildPlanNodeInRegion(
+ PlanNode root, TRegionReplicaSet regionReplicaSet, MPPQueryContext
context) {
+ List<PlanNode> children =
+ root.getChildren().stream()
+ .map(child -> buildPlanNodeInRegion(child, regionReplicaSet,
context))
+ .collect(Collectors.toList());
+ PlanNode newRoot = root.cloneWithChildren(children);
+ newRoot.setPlanNodeId(context.getQueryId().genPlanNodeId());
+ if (newRoot instanceof SourceNode) {
+ ((SourceNode) newRoot).setRegionReplicaSet(regionReplicaSet);
+ }
+ return newRoot;
+ }
}
public PlanNode visitDeleteTimeseries(
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
index 613277bd67..f10b69a474 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceMergeNode.java
@@ -32,7 +32,7 @@ import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
-public class DeviceMergeNode extends ProcessNode {
+public class DeviceMergeNode extends MultiChildNode {
// The result output order, which could sort by device and time.
// The size of this list is 2 and the first OrderBy in this list has higher
priority.
@@ -41,8 +41,6 @@ public class DeviceMergeNode extends ProcessNode {
// the list of selected devices
private final List<String> devices;
- private final List<PlanNode> children;
-
public DeviceMergeNode(
PlanNodeId id, List<PlanNode> children, List<OrderBy> mergeOrders,
List<String> devices) {
super(id);
@@ -146,4 +144,9 @@ public class DeviceMergeNode extends ProcessNode {
public int hashCode() {
return Objects.hash(super.hashCode(), mergeOrders, devices, children);
}
+
+ @Override
+ public String toString() {
+ return "DeviceMerge-" + this.getPlanNodeId();
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
index 589e666297..a70252c55d 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/DeviceViewNode.java
@@ -40,7 +40,7 @@ import java.util.Objects;
* same between these TsBlocks. If the input TsBlock contains n columns, the
device-based view will
* contain n+1 columns where the new column is Device column.
*/
-public class DeviceViewNode extends ProcessNode {
+public class DeviceViewNode extends MultiChildNode {
// The result output order, which could sort by device and time.
// The size of this list is 2 and the first OrderBy in this list has higher
priority.
@@ -49,9 +49,6 @@ public class DeviceViewNode extends ProcessNode {
// The size devices and children should be the same.
private final List<String> devices = new ArrayList<>();
- // each child node whose output TsBlock contains the data belonged to one
device.
- private final List<PlanNode> children = new ArrayList<>();
-
// Device column and measurement columns in result output
private final List<String> outputColumnNames;
@@ -117,6 +114,10 @@ public class DeviceViewNode extends ProcessNode {
getPlanNodeId(), mergeOrders, outputColumnNames, devices,
deviceToMeasurementIndexesMap);
}
+ public List<OrderBy> getMergeOrders() {
+ return mergeOrders;
+ }
+
@Override
public List<String> getOutputColumnNames() {
return outputColumnNames;
@@ -213,4 +214,9 @@ public class DeviceViewNode extends ProcessNode {
outputColumnNames,
deviceToMeasurementIndexesMap);
}
+
+ @Override
+ public String toString() {
+ return "DeviceView-" + this.getPlanNodeId();
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
index 7210bf449e..34c5bc7708 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/process/GroupByLevelNode.java
@@ -161,6 +161,15 @@ public class GroupByLevelNode extends MultiChildNode {
planNodeId, groupByLevelDescriptors, groupByTimeParameter, scanOrder);
}
+ @Nullable
+ public GroupByTimeParameter getGroupByTimeParameter() {
+ return groupByTimeParameter;
+ }
+
+ public OrderBy getScanOrder() {
+ return scanOrder;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
index a29d20bf12..6b561c4fed 100644
---
a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
+++
b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/MemAlignedPageReader.java
@@ -41,7 +41,6 @@ public class MemAlignedPageReader implements IPageReader,
IAlignedPageReader {
private final TsBlock tsBlock;
private final AlignedChunkMetadata chunkMetadata;
private Filter valueFilter;
-
private TsBlockBuilder builder;
public MemAlignedPageReader(TsBlock tsBlock, AlignedChunkMetadata
chunkMetadata, Filter filter) {
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AlignedByDeviceTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AlignedByDeviceTest.java
new file mode 100644
index 0000000000..af64fc39bb
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AlignedByDeviceTest.java
@@ -0,0 +1,36 @@
+/*
+ * 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.plan.plan.distribution;
+
+import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
+
+import org.junit.Test;
+
+import java.util.List;
+
+public class AlignedByDeviceTest {
+
+ @Test
+ public void test1Device1Region() {}
+
+ private LogicalQueryPlan constructLogicalPlan(List<String> series) {
+ return null;
+ }
+}
