You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2022/05/25 13:37:03 UTC
[iotdb] branch master updated: Distribution plan for delete timeseries (#6006)
This is an automated email from the ASF dual-hosted git repository.
jackietien 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 756e876256 Distribution plan for delete timeseries (#6006)
756e876256 is described below
commit 756e8762566a79868bcfaec819899b50e4c1d078
Author: Zhang.Jinrui <xi...@gmail.com>
AuthorDate: Wed May 25 21:36:58 2022 +0800
Distribution plan for delete timeseries (#6006)
---
.../iotdb/commons/partition/DataPartition.java | 19 +
.../commons/partition/RegionReplicaSetInfo.java | 27 +-
.../iotdb/commons/partition/SchemaPartition.java | 13 +
.../db/mpp/common/schematree/PathPatternTree.java | 6 +-
.../db/mpp/plan/execution/QueryExecution.java | 2 +-
.../db/mpp/plan/planner/DistributionPlanner.java | 909 --------------------
.../db/mpp/plan/planner/LogicalPlanBuilder.java | 28 +
.../iotdb/db/mpp/plan/planner/LogicalPlanner.java | 12 +
.../DistributionPlanContext.java} | 18 +-
.../planner/distribution/DistributionPlanner.java | 179 ++++
.../planner/distribution/ExchangeNodeAdder.java | 274 ++++++
.../NodeDistribution.java} | 20 +-
.../NodeDistributionType.java} | 21 +-
.../NodeGroupContext.java} | 28 +-
.../SimpleFragmentParallelPlanner.java | 3 +-
.../plan/planner/distribution/SourceRewriter.java | 590 +++++++++++++
.../WriteFragmentParallelPlanner.java | 3 +-
.../db/mpp/plan/planner/plan/PlanFragment.java | 6 +-
.../iotdb/db/mpp/plan/planner/plan/SubPlan.java | 2 +-
...itePlanNode.java => IPartitionRelatedNode.java} | 14 +-
.../db/mpp/plan/planner/plan/node/PlanVisitor.java | 10 +
.../mpp/plan/planner/plan/node/WritePlanNode.java | 5 +-
.../node/metedata/write/DeleteTimeSeriesNode.java | 38 +-
.../metedata/write/InvalidateSchemaCacheNode.java | 15 +-
.../plan/planner/plan/node/source/SourceNode.java | 5 +-
.../planner/plan/node/write/DeleteDataNode.java | 72 +-
.../metadata/DeleteTimeSeriesStatement.java | 2 +-
.../db/mpp/plan/plan/DistributionPlannerTest.java | 934 ---------------------
.../distribution/AggregationDistributionTest.java | 412 +++++++++
.../db/mpp/plan/plan/distribution/BasicTest.java | 378 +++++++++
.../plan/distribution/DeleteTimeseriesTest.java | 71 ++
.../iotdb/db/mpp/plan/plan/distribution/Util.java | 191 +++++
.../plan/node/write/DeleteDataNodeSerdeTest.java | 19 +-
33 files changed, 2359 insertions(+), 1967 deletions(-)
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
index e622a284de..b1e23cb9d6 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
@@ -313,4 +313,23 @@ public class DataPartition extends Partition {
storageGroupNum--;
}
}
+
+ public List<RegionReplicaSetInfo> getDataDistributionInfo() {
+ Map<TRegionReplicaSet, RegionReplicaSetInfo> distributionMap = new HashMap<>();
+
+ dataPartitionMap.forEach(
+ (storageGroup, partition) -> {
+ List<TRegionReplicaSet> ret =
+ partition.entrySet().stream()
+ .flatMap(
+ s -> s.getValue().entrySet().stream().flatMap(e -> e.getValue().stream()))
+ .collect(Collectors.toList());
+ for (TRegionReplicaSet regionReplicaSet : ret) {
+ distributionMap
+ .computeIfAbsent(regionReplicaSet, RegionReplicaSetInfo::new)
+ .addStorageGroup(storageGroup);
+ }
+ });
+ return new ArrayList<>(distributionMap.values());
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/RegionReplicaSetInfo.java
similarity index 56%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
copy to node-commons/src/main/java/org/apache/iotdb/commons/partition/RegionReplicaSetInfo.java
index 0d6ee0c793..f2017fdb24 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/RegionReplicaSetInfo.java
@@ -17,20 +17,33 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner.plan.node;
+package org.apache.iotdb.commons.partition;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
-public abstract class WritePlanNode extends PlanNode {
+public class RegionReplicaSetInfo {
+ private TRegionReplicaSet regionReplicaSet;
+ private Set<String> ownedStorageGroups;
- protected WritePlanNode(PlanNodeId id) {
- super(id);
+ public RegionReplicaSetInfo(TRegionReplicaSet regionReplicaSet) {
+ this.regionReplicaSet = regionReplicaSet;
+ this.ownedStorageGroups = new HashSet<>();
}
- public abstract TRegionReplicaSet getRegionReplicaSet();
+ public void addStorageGroup(String storageGroup) {
+ ownedStorageGroups.add(storageGroup);
+ }
+
+ public TRegionReplicaSet getRegionReplicaSet() {
+ return regionReplicaSet;
+ }
- public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
+ public List<String> getOwnedStorageGroups() {
+ return new ArrayList<>(ownedStorageGroups);
+ }
}
diff --git a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
index 2311f7a743..d3236136d4 100644
--- a/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
+++ b/node-commons/src/main/java/org/apache/iotdb/commons/partition/SchemaPartition.java
@@ -231,6 +231,19 @@ public class SchemaPartition extends Partition {
return result;
}
+ public List<RegionReplicaSetInfo> getSchemaDistributionInfo() {
+ Map<TRegionReplicaSet, RegionReplicaSetInfo> distributionMap = new HashMap<>();
+ schemaPartitionMap.forEach(
+ (storageGroup, partition) -> {
+ for (TRegionReplicaSet regionReplicaSet : partition.values()) {
+ distributionMap
+ .computeIfAbsent(regionReplicaSet, RegionReplicaSetInfo::new)
+ .addStorageGroup(storageGroup);
+ }
+ });
+ return new ArrayList<>(distributionMap.values());
+ }
+
private void writeMap(
Map<TSeriesPartitionSlot, TRegionReplicaSet> valueMap,
OutputStream outputStream,
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
index 2a34c2ad1c..1700a4faa7 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/common/schematree/PathPatternTree.java
@@ -251,6 +251,10 @@ public class PathPatternTree {
}
public PathPatternTree findOverlappedPattern(PartialPath pattern) {
+ return new PathPatternTree(findOverlappedPaths(pattern));
+ }
+
+ public List<PartialPath> findOverlappedPaths(PartialPath pattern) {
if (pathList.isEmpty()) {
pathList = splitToPathList();
}
@@ -261,7 +265,7 @@ public class PathPatternTree {
results.add(path);
}
}
- return new PathPatternTree(results);
+ return results;
}
@TestOnly
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
index eb289306ef..868e506dba 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/QueryExecution.java
@@ -39,8 +39,8 @@ import org.apache.iotdb.db.mpp.plan.execution.memory.StatementMemorySource;
import org.apache.iotdb.db.mpp.plan.execution.memory.StatementMemorySourceContext;
import org.apache.iotdb.db.mpp.plan.execution.memory.StatementMemorySourceVisitor;
import org.apache.iotdb.db.mpp.plan.optimization.PlanOptimizer;
-import org.apache.iotdb.db.mpp.plan.planner.DistributionPlanner;
import org.apache.iotdb.db.mpp.plan.planner.LogicalPlanner;
+import org.apache.iotdb.db.mpp.plan.planner.distribution.DistributionPlanner;
import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java
deleted file mode 100644
index 9756e4c6d6..0000000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/DistributionPlanner.java
+++ /dev/null
@@ -1,909 +0,0 @@
-/*
- * 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.planner;
-
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.mpp.common.MPPQueryContext;
-import org.apache.iotdb.db.mpp.common.PlanFragmentId;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
-import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
-import org.apache.iotdb.db.mpp.plan.expression.Expression;
-import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
-import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.PlanFragment;
-import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.SimplePlanNodeRewriter;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.AbstractSchemaMergeNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.CountSchemaMergeNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchMergeNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
-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.MultiChildNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesAggregationScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationSourceNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesSourceNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByLevelDescriptor;
-import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.stream.Collectors;
-
-import static com.google.common.collect.ImmutableList.toImmutableList;
-
-public class DistributionPlanner {
- private Analysis analysis;
- private MPPQueryContext context;
- private LogicalQueryPlan logicalPlan;
-
- private int planFragmentIndex = 0;
-
- public DistributionPlanner(Analysis analysis, LogicalQueryPlan logicalPlan) {
- this.analysis = analysis;
- this.logicalPlan = logicalPlan;
- this.context = logicalPlan.getContext();
- }
-
- public PlanNode rewriteSource() {
- SourceRewriter rewriter = new SourceRewriter();
- return rewriter.visit(logicalPlan.getRootNode(), new DistributionPlanContext(context));
- }
-
- public PlanNode addExchangeNode(PlanNode root) {
- ExchangeNodeAdder adder = new ExchangeNodeAdder();
- return adder.visit(root, new NodeGroupContext(context));
- }
-
- public SubPlan splitFragment(PlanNode root) {
- FragmentBuilder fragmentBuilder = new FragmentBuilder(context);
- return fragmentBuilder.splitToSubPlan(root);
- }
-
- public DistributedQueryPlan planFragments() {
- PlanNode rootAfterRewrite = rewriteSource();
- PlanNode rootWithExchange = addExchangeNode(rootAfterRewrite);
- if (analysis.getStatement() instanceof QueryStatement) {
- analysis
- .getRespDatasetHeader()
- .setColumnToTsBlockIndexMap(rootWithExchange.getOutputColumnNames());
- }
- SubPlan subPlan = splitFragment(rootWithExchange);
- List<FragmentInstance> fragmentInstances = planFragmentInstances(subPlan);
- // Only execute this step for READ operation
- if (context.getQueryType() == QueryType.READ) {
- SetSinkForRootInstance(subPlan, fragmentInstances);
- }
- return new DistributedQueryPlan(
- logicalPlan.getContext(), subPlan, subPlan.getPlanFragmentList(), fragmentInstances);
- }
-
- // Convert fragment to detailed instance
- // And for parallel-able fragment, clone it into several instances with different params.
- public List<FragmentInstance> planFragmentInstances(SubPlan subPlan) {
- IFragmentParallelPlaner parallelPlaner =
- context.getQueryType() == QueryType.READ
- ? new SimpleFragmentParallelPlanner(subPlan, analysis, context)
- : new WriteFragmentParallelPlanner(subPlan, analysis, context);
- return parallelPlaner.parallelPlan();
- }
-
- // TODO: (xingtanzjr) Maybe we should handle ResultNode in LogicalPlanner ?
- public void SetSinkForRootInstance(SubPlan subPlan, List<FragmentInstance> instances) {
- FragmentInstance rootInstance = null;
- for (FragmentInstance instance : instances) {
- if (instance.getFragment().getId().equals(subPlan.getPlanFragment().getId())) {
- rootInstance = instance;
- break;
- }
- }
- // root should not be null during normal process
- if (rootInstance == null) {
- return;
- }
-
- FragmentSinkNode sinkNode = new FragmentSinkNode(context.getQueryId().genPlanNodeId());
- sinkNode.setDownStream(
- context.getLocalDataBlockEndpoint(),
- context.getResultNodeContext().getVirtualFragmentInstanceId(),
- context.getResultNodeContext().getVirtualResultNodeId());
- sinkNode.setChild(rootInstance.getFragment().getRoot());
- context
- .getResultNodeContext()
- .setUpStream(
- rootInstance.getHostDataNode().dataBlockManagerEndPoint,
- rootInstance.getId(),
- sinkNode.getPlanNodeId());
- rootInstance.getFragment().setRoot(sinkNode);
- }
-
- private PlanFragmentId getNextFragmentId() {
- return new PlanFragmentId(this.logicalPlan.getContext().getQueryId(), this.planFragmentIndex++);
- }
-
- private class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanContext> {
-
- // TODO: (xingtanzjr) implement the method visitDeviceMergeNode()
- public PlanNode visitDeviceMerge(TimeJoinNode node, DistributionPlanContext context) {
- return null;
- }
-
- @Override
- public PlanNode visitSchemaQueryMerge(
- SchemaQueryMergeNode node, DistributionPlanContext context) {
- SchemaQueryMergeNode root = (SchemaQueryMergeNode) node.clone();
- SchemaQueryScanNode seed = (SchemaQueryScanNode) node.getChildren().get(0);
- TreeSet<TRegionReplicaSet> schemaRegions =
- new TreeSet<>(Comparator.comparingInt(region -> region.getRegionId().getId()));
- analysis
- .getSchemaPartitionInfo()
- .getSchemaPartitionMap()
- .forEach(
- (storageGroup, deviceGroup) -> {
- deviceGroup.forEach(
- (deviceGroupId, schemaRegionReplicaSet) ->
- schemaRegions.add(schemaRegionReplicaSet));
- });
- int count = schemaRegions.size();
- schemaRegions.forEach(
- region -> {
- SchemaQueryScanNode schemaQueryScanNode = (SchemaQueryScanNode) seed.clone();
- schemaQueryScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- schemaQueryScanNode.setRegionReplicaSet(region);
- if (count > 1) {
- schemaQueryScanNode.setLimit(
- schemaQueryScanNode.getOffset() + schemaQueryScanNode.getLimit());
- schemaQueryScanNode.setOffset(0);
- }
- root.addChild(schemaQueryScanNode);
- });
- return root;
- }
-
- @Override
- public PlanNode visitCountMerge(CountSchemaMergeNode node, DistributionPlanContext context) {
- CountSchemaMergeNode root = (CountSchemaMergeNode) node.clone();
- SchemaQueryScanNode seed = (SchemaQueryScanNode) node.getChildren().get(0);
- Set<TRegionReplicaSet> schemaRegions = new HashSet<>();
- analysis
- .getSchemaPartitionInfo()
- .getSchemaPartitionMap()
- .forEach(
- (storageGroup, deviceGroup) -> {
- deviceGroup.forEach(
- (deviceGroupId, schemaRegionReplicaSet) ->
- schemaRegions.add(schemaRegionReplicaSet));
- });
- schemaRegions.forEach(
- region -> {
- SchemaQueryScanNode schemaQueryScanNode = (SchemaQueryScanNode) seed.clone();
- schemaQueryScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- schemaQueryScanNode.setRegionReplicaSet(region);
- root.addChild(schemaQueryScanNode);
- });
- return root;
- }
-
- // TODO: (xingtanzjr) a temporary way to resolve the distribution of single SeriesScanNode issue
- @Override
- public PlanNode visitSeriesScan(SeriesScanNode node, DistributionPlanContext context) {
- List<TRegionReplicaSet> dataDistribution =
- analysis.getPartitionInfo(node.getSeriesPath(), node.getTimeFilter());
- if (dataDistribution.size() == 1) {
- node.setRegionReplicaSet(dataDistribution.get(0));
- return node;
- }
- TimeJoinNode timeJoinNode =
- new TimeJoinNode(context.queryContext.getQueryId().genPlanNodeId(), node.getScanOrder());
- for (TRegionReplicaSet dataRegion : dataDistribution) {
- SeriesScanNode split = (SeriesScanNode) node.clone();
- split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- split.setRegionReplicaSet(dataRegion);
- timeJoinNode.addChild(split);
- }
- return timeJoinNode;
- }
-
- @Override
- public PlanNode visitSeriesAggregationScan(
- SeriesAggregationScanNode node, DistributionPlanContext context) {
- return processSeriesAggregationSource(node, context);
- }
-
- @Override
- public PlanNode visitAlignedSeriesAggregationScan(
- AlignedSeriesAggregationScanNode node, DistributionPlanContext context) {
- return processSeriesAggregationSource(node, context);
- }
-
- private PlanNode processSeriesAggregationSource(
- SeriesAggregationSourceNode node, DistributionPlanContext context) {
- List<TRegionReplicaSet> dataDistribution =
- analysis.getPartitionInfo(node.getPartitionPath(), node.getPartitionTimeFilter());
- if (dataDistribution.size() == 1) {
- node.setRegionReplicaSet(dataDistribution.get(0));
- return node;
- }
- List<AggregationDescriptor> leafAggDescriptorList = new ArrayList<>();
- node.getAggregationDescriptorList()
- .forEach(
- descriptor -> {
- leafAggDescriptorList.add(
- new AggregationDescriptor(
- descriptor.getAggregationType(),
- AggregationStep.PARTIAL,
- descriptor.getInputExpressions()));
- });
-
- List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
- node.getAggregationDescriptorList()
- .forEach(
- descriptor -> {
- rootAggDescriptorList.add(
- new AggregationDescriptor(
- descriptor.getAggregationType(),
- AggregationStep.FINAL,
- descriptor.getInputExpressions()));
- });
-
- AggregationNode aggregationNode =
- new AggregationNode(
- context.queryContext.getQueryId().genPlanNodeId(), rootAggDescriptorList);
- for (TRegionReplicaSet dataRegion : dataDistribution) {
- SeriesAggregationScanNode split = (SeriesAggregationScanNode) node.clone();
- split.setAggregationDescriptorList(leafAggDescriptorList);
- split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- split.setRegionReplicaSet(dataRegion);
- aggregationNode.addChild(split);
- }
- return aggregationNode;
- }
-
- @Override
- public PlanNode visitAlignedSeriesScan(
- AlignedSeriesScanNode node, DistributionPlanContext context) {
- List<TRegionReplicaSet> dataDistribution =
- analysis.getPartitionInfo(node.getAlignedPath(), node.getTimeFilter());
- if (dataDistribution.size() == 1) {
- node.setRegionReplicaSet(dataDistribution.get(0));
- return node;
- }
- TimeJoinNode timeJoinNode =
- new TimeJoinNode(context.queryContext.getQueryId().genPlanNodeId(), node.getScanOrder());
- for (TRegionReplicaSet dataRegion : dataDistribution) {
- AlignedSeriesScanNode split = (AlignedSeriesScanNode) node.clone();
- split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- split.setRegionReplicaSet(dataRegion);
- timeJoinNode.addChild(split);
- }
- return timeJoinNode;
- }
-
- @Override
- public PlanNode visitSchemaFetchMerge(
- SchemaFetchMergeNode node, DistributionPlanContext context) {
- SchemaFetchMergeNode root = (SchemaFetchMergeNode) node.clone();
- Map<String, Set<TRegionReplicaSet>> storageGroupSchemaRegionMap = new HashMap<>();
- analysis
- .getSchemaPartitionInfo()
- .getSchemaPartitionMap()
- .forEach(
- (storageGroup, deviceGroup) -> {
- storageGroupSchemaRegionMap.put(storageGroup, new HashSet<>());
- deviceGroup.forEach(
- (deviceGroupId, schemaRegionReplicaSet) ->
- storageGroupSchemaRegionMap.get(storageGroup).add(schemaRegionReplicaSet));
- });
-
- for (PlanNode child : node.getChildren()) {
- for (TRegionReplicaSet schemaRegion :
- storageGroupSchemaRegionMap.get(
- ((SchemaFetchScanNode) child).getStorageGroup().getFullPath())) {
- SchemaFetchScanNode schemaFetchScanNode = (SchemaFetchScanNode) child.clone();
- schemaFetchScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- schemaFetchScanNode.setRegionReplicaSet(schemaRegion);
- root.addChild(schemaFetchScanNode);
- }
- }
- return root;
- }
-
- @Override
- public PlanNode visitTimeJoin(TimeJoinNode node, DistributionPlanContext context) {
- // Although some logic is similar between Aggregation and RawDataQuery,
- // we still use separate method to process the distribution planning now
- // to make the planning procedure more clear
- if (isAggregationQuery(node)) {
- return planAggregationWithTimeJoin(node, context);
- }
-
- TimeJoinNode root = (TimeJoinNode) node.clone();
- // Step 1: Get all source nodes. For the node which is not source, add it as the child of
- // current TimeJoinNode
- List<SourceNode> sources = new ArrayList<>();
- for (PlanNode child : node.getChildren()) {
- if (child instanceof SeriesSourceNode) {
- // If the child is SeriesScanNode, we need to check whether this node should be seperated
- // into several splits.
- SeriesSourceNode handle = (SeriesSourceNode) child;
- List<TRegionReplicaSet> dataDistribution =
- analysis.getPartitionInfo(handle.getPartitionPath(), handle.getPartitionTimeFilter());
- // If the size of dataDistribution is m, this SeriesScanNode should be seperated into m
- // SeriesScanNode.
- for (TRegionReplicaSet dataRegion : dataDistribution) {
- SeriesSourceNode split = (SeriesSourceNode) handle.clone();
- split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- split.setRegionReplicaSet(dataRegion);
- sources.add(split);
- }
- }
- }
- // Step 2: For the source nodes, group them by the DataRegion.
- Map<TRegionReplicaSet, List<SourceNode>> sourceGroup =
- sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
-
- // Step 3: For the source nodes which belong to same data region, add a TimeJoinNode for them
- // and make the
- // new TimeJoinNode as the child of current TimeJoinNode
- // TODO: (xingtanzjr) optimize the procedure here to remove duplicated TimeJoinNode
- final boolean[] addParent = {false};
- sourceGroup.forEach(
- (dataRegion, seriesScanNodes) -> {
- if (seriesScanNodes.size() == 1) {
- root.addChild(seriesScanNodes.get(0));
- } else {
- if (!addParent[0]) {
- seriesScanNodes.forEach(root::addChild);
- addParent[0] = true;
- } else {
- // We clone a TimeJoinNode from root to make the params to be consistent.
- // But we need to assign a new ID to it
- TimeJoinNode parentOfGroup = (TimeJoinNode) root.clone();
- parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- seriesScanNodes.forEach(parentOfGroup::addChild);
- root.addChild(parentOfGroup);
- }
- }
- });
-
- // Process the other children which are not SeriesSourceNode
- for (PlanNode child : node.getChildren()) {
- if (!(child instanceof SeriesSourceNode)) {
- // In a general logical query plan, the children of TimeJoinNode should only be
- // SeriesScanNode or SeriesAggregateScanNode
- // So this branch should not be touched.
- root.addChild(visit(child, context));
- }
- }
-
- return root;
- }
-
- private boolean isAggregationQuery(TimeJoinNode node) {
- for (PlanNode child : node.getChildren()) {
- if (child instanceof SeriesAggregationScanNode
- || child instanceof AlignedSeriesAggregationScanNode) {
- return true;
- }
- }
- return false;
- }
-
- private PlanNode planAggregationWithTimeJoin(
- TimeJoinNode root, DistributionPlanContext context) {
-
- List<SeriesAggregationSourceNode> sources = splitAggregationSourceByPartition(root, context);
- Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
- sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
-
- // construct AggregationDescriptor for AggregationNode
- List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
- for (PlanNode child : root.getChildren()) {
- SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
- handle
- .getAggregationDescriptorList()
- .forEach(
- descriptor -> {
- rootAggDescriptorList.add(
- new AggregationDescriptor(
- descriptor.getAggregationType(),
- AggregationStep.FINAL,
- descriptor.getInputExpressions()));
- });
- }
- AggregationNode aggregationNode =
- new AggregationNode(
- context.queryContext.getQueryId().genPlanNodeId(), rootAggDescriptorList);
-
- final boolean[] addParent = {false};
- sourceGroup.forEach(
- (dataRegion, sourceNodes) -> {
- if (sourceNodes.size() == 1) {
- aggregationNode.addChild(sourceNodes.get(0));
- } else {
- if (!addParent[0]) {
- sourceNodes.forEach(aggregationNode::addChild);
- addParent[0] = true;
- } else {
- // We clone a TimeJoinNode from root to make the params to be consistent.
- // But we need to assign a new ID to it
- TimeJoinNode parentOfGroup = (TimeJoinNode) root.clone();
- parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- sourceNodes.forEach(parentOfGroup::addChild);
- aggregationNode.addChild(parentOfGroup);
- }
- }
- });
-
- return aggregationNode;
- }
-
- public PlanNode visitGroupByLevel(GroupByLevelNode root, DistributionPlanContext context) {
- // Firstly, we build the tree structure for GroupByLevelNode
- List<SeriesAggregationSourceNode> sources = splitAggregationSourceByPartition(root, context);
- Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
- sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
-
- GroupByLevelNode newRoot = (GroupByLevelNode) root.clone();
- final boolean[] addParent = {false};
- sourceGroup.forEach(
- (dataRegion, sourceNodes) -> {
- if (sourceNodes.size() == 1) {
- newRoot.addChild(sourceNodes.get(0));
- } else {
- if (!addParent[0]) {
- sourceNodes.forEach(newRoot::addChild);
- addParent[0] = true;
- } else {
- // We clone a TimeJoinNode from root to make the params to be consistent.
- // But we need to assign a new ID to it
- GroupByLevelNode parentOfGroup = (GroupByLevelNode) root.clone();
- parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- sourceNodes.forEach(parentOfGroup::addChild);
- newRoot.addChild(parentOfGroup);
- }
- }
- });
-
- // Then, we calculate the attributes for GroupByLevelNode in each level
- calculateGroupByLevelNodeAttributes(newRoot, 0);
- return newRoot;
- }
-
- private void calculateGroupByLevelNodeAttributes(PlanNode node, int level) {
- if (node == null) {
- return;
- }
- node.getChildren().forEach(child -> calculateGroupByLevelNodeAttributes(child, level + 1));
- if (!(node instanceof GroupByLevelNode)) {
- return;
- }
- GroupByLevelNode handle = (GroupByLevelNode) node;
-
- // Construct all outputColumns from children. Using Set here to avoid duplication
- Set<String> childrenOutputColumns = new HashSet<>();
- handle
- .getChildren()
- .forEach(child -> childrenOutputColumns.addAll(child.getOutputColumnNames()));
-
- // Check every OutputColumn of GroupByLevelNode and set the Expression of corresponding
- // AggregationDescriptor
- List<GroupByLevelDescriptor> descriptorList = new ArrayList<>();
- for (GroupByLevelDescriptor originalDescriptor : handle.getGroupByLevelDescriptors()) {
- List<Expression> descriptorExpression = new ArrayList<>();
- for (String childColumn : childrenOutputColumns) {
- // If this condition matched, the childColumn should come from GroupByLevelNode
- if (isAggColumnMatchExpression(childColumn, originalDescriptor.getOutputExpression())) {
- descriptorExpression.add(originalDescriptor.getOutputExpression());
- continue;
- }
- for (Expression exp : originalDescriptor.getInputExpressions()) {
- if (isAggColumnMatchExpression(childColumn, exp)) {
- descriptorExpression.add(exp);
- }
- }
- }
- if (descriptorExpression.size() == 0) {
- continue;
- }
- GroupByLevelDescriptor descriptor = originalDescriptor.deepClone();
- descriptor.setStep(level == 0 ? AggregationStep.FINAL : AggregationStep.PARTIAL);
- descriptor.setInputExpressions(descriptorExpression);
-
- descriptorList.add(descriptor);
- }
- handle.setGroupByLevelDescriptors(descriptorList);
- }
-
- // TODO: (xingtanzjr) need to confirm the logic when processing UDF
- private boolean isAggColumnMatchExpression(String columnName, Expression expression) {
- if (columnName == null) {
- return false;
- }
- return columnName.contains(expression.getExpressionString());
- }
-
- private List<SeriesAggregationSourceNode> splitAggregationSourceByPartition(
- MultiChildNode root, DistributionPlanContext context) {
- // Step 1: split SeriesAggregationSourceNode according to data partition
- List<SeriesAggregationSourceNode> sources = new ArrayList<>();
- Map<PartialPath, Integer> regionCountPerSeries = new HashMap<>();
- for (PlanNode child : root.getChildren()) {
- SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
- List<TRegionReplicaSet> dataDistribution =
- analysis.getPartitionInfo(handle.getPartitionPath(), handle.getPartitionTimeFilter());
- for (TRegionReplicaSet dataRegion : dataDistribution) {
- SeriesAggregationSourceNode split = (SeriesAggregationSourceNode) handle.clone();
- split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- split.setRegionReplicaSet(dataRegion);
- // Let each split reference different object of AggregationDescriptorList
- split.setAggregationDescriptorList(
- handle.getAggregationDescriptorList().stream()
- .map(AggregationDescriptor::deepClone)
- .collect(Collectors.toList()));
- sources.add(split);
- }
- regionCountPerSeries.put(handle.getPartitionPath(), dataDistribution.size());
- }
-
- // Step 2: change the step for each SeriesAggregationSourceNode according to its split count
- for (SeriesAggregationSourceNode source : sources) {
- // boolean isFinal = regionCountPerSeries.get(source.getPartitionPath()) == 1;
- // TODO: (xingtanzjr) need to optimize this step later. We make it as Partial now.
- boolean isFinal = false;
- source
- .getAggregationDescriptorList()
- .forEach(d -> d.setStep(isFinal ? AggregationStep.FINAL : AggregationStep.PARTIAL));
- }
- return sources;
- }
-
- public PlanNode visit(PlanNode node, DistributionPlanContext context) {
- return node.accept(this, context);
- }
- }
-
- private class DistributionPlanContext {
- private MPPQueryContext queryContext;
-
- public DistributionPlanContext(MPPQueryContext queryContext) {
- this.queryContext = queryContext;
- }
- }
-
- private class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
- @Override
- public PlanNode visitPlan(PlanNode node, NodeGroupContext context) {
- // TODO: (xingtanzjr) we apply no action for IWritePlanNode currently
- if (node instanceof WritePlanNode) {
- return node;
- }
- // 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.getPlanNodeId(),
- new NodeDistribution(NodeDistributionType.SAME_WITH_ALL_CHILDREN, null));
-
- return node.cloneWithChildren(children);
- }
-
- @Override
- public PlanNode visitSchemaQueryMerge(SchemaQueryMergeNode node, NodeGroupContext context) {
- return internalVisitSchemaMerge(node, context);
- }
-
- private PlanNode internalVisitSchemaMerge(
- AbstractSchemaMergeNode node, NodeGroupContext context) {
- node.getChildren()
- .forEach(
- child -> {
- visit(child, context);
- });
- NodeDistribution nodeDistribution =
- new NodeDistribution(NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN);
- PlanNode newNode = node.clone();
- nodeDistribution.region = calculateSchemaRegionByChildren(node.getChildren(), context);
- context.putNodeDistribution(newNode.getPlanNodeId(), nodeDistribution);
- node.getChildren()
- .forEach(
- child -> {
- if (!nodeDistribution.region.equals(
- context.getNodeDistribution(child.getPlanNodeId()).region)) {
- ExchangeNode exchangeNode =
- new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
- exchangeNode.setChild(child);
- exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
- newNode.addChild(exchangeNode);
- } else {
- newNode.addChild(child);
- }
- });
- return newNode;
- }
-
- @Override
- public PlanNode visitCountMerge(CountSchemaMergeNode node, NodeGroupContext context) {
- return internalVisitSchemaMerge(node, context);
- }
-
- @Override
- public PlanNode visitSchemaQueryScan(SchemaQueryScanNode node, NodeGroupContext context) {
- NodeDistribution nodeDistribution = new NodeDistribution(NodeDistributionType.NO_CHILD);
- nodeDistribution.region = node.getRegionReplicaSet();
- context.putNodeDistribution(node.getPlanNodeId(), nodeDistribution);
- return node;
- }
-
- @Override
- public PlanNode visitSchemaFetchMerge(SchemaFetchMergeNode node, NodeGroupContext context) {
- return internalVisitSchemaMerge(node, context);
- }
-
- @Override
- public PlanNode visitSchemaFetchScan(SchemaFetchScanNode node, NodeGroupContext context) {
- NodeDistribution nodeDistribution = new NodeDistribution(NodeDistributionType.NO_CHILD);
- nodeDistribution.region = node.getRegionReplicaSet();
- context.putNodeDistribution(node.getPlanNodeId(), nodeDistribution);
- return node;
- }
-
- @Override
- public PlanNode visitSeriesScan(SeriesScanNode node, NodeGroupContext context) {
- context.putNodeDistribution(
- node.getPlanNodeId(),
- new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
- return node.clone();
- }
-
- @Override
- public PlanNode visitAlignedSeriesScan(AlignedSeriesScanNode node, NodeGroupContext context) {
- context.putNodeDistribution(
- node.getPlanNodeId(),
- new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
- return node.clone();
- }
-
- public PlanNode visitSeriesAggregationScan(
- SeriesAggregationScanNode node, NodeGroupContext context) {
- context.putNodeDistribution(
- node.getPlanNodeId(),
- new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
- return node.clone();
- }
-
- public PlanNode visitAlignedSeriesAggregationScan(
- AlignedSeriesAggregationScanNode node, NodeGroupContext context) {
- context.putNodeDistribution(
- node.getPlanNodeId(),
- new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
- return node.clone();
- }
-
- @Override
- public PlanNode visitTimeJoin(TimeJoinNode node, NodeGroupContext context) {
- return processMultiChildNode(node, context);
- }
-
- public PlanNode visitRowBasedSeriesAggregate(AggregationNode node, NodeGroupContext context) {
- return processMultiChildNode(node, context);
- }
-
- @Override
- public PlanNode visitGroupByLevel(GroupByLevelNode node, NodeGroupContext context) {
- return processMultiChildNode(node, context);
- }
-
- private PlanNode processMultiChildNode(MultiChildNode node, NodeGroupContext context) {
- MultiChildNode newNode = (MultiChildNode) node.clone();
- List<PlanNode> visitedChildren = new ArrayList<>();
- node.getChildren()
- .forEach(
- child -> {
- visitedChildren.add(visit(child, context));
- });
-
- TRegionReplicaSet dataRegion = calculateDataRegionByChildren(visitedChildren, context);
- NodeDistributionType distributionType =
- nodeDistributionIsSame(visitedChildren, context)
- ? NodeDistributionType.SAME_WITH_ALL_CHILDREN
- : NodeDistributionType.SAME_WITH_SOME_CHILD;
- context.putNodeDistribution(
- newNode.getPlanNodeId(), 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.getPlanNodeId()).region)) {
- ExchangeNode exchangeNode =
- new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
- exchangeNode.setChild(child);
- exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
- newNode.addChild(exchangeNode);
- } else {
- newNode.addChild(child);
- }
- });
- return newNode;
- }
-
- private TRegionReplicaSet calculateDataRegionByChildren(
- List<PlanNode> children, NodeGroupContext context) {
- // Step 1: calculate the count of children group by DataRegion.
- Map<TRegionReplicaSet, Long> groupByRegion =
- children.stream()
- .collect(
- Collectors.groupingBy(
- child -> context.getNodeDistribution(child.getPlanNodeId()).region,
- Collectors.counting()));
- // Step 2: return the RegionReplicaSet with max count
- return Collections.max(groupByRegion.entrySet(), Map.Entry.comparingByValue()).getKey();
- }
-
- private TRegionReplicaSet calculateSchemaRegionByChildren(
- List<PlanNode> children, NodeGroupContext context) {
- // We always make the schemaRegion of MetaMergeNode to be the same as its first child.
- return context.getNodeDistribution(children.get(0).getPlanNodeId()).region;
- }
-
- 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).getPlanNodeId());
- for (int i = 1; i < children.size(); i++) {
- NodeDistribution next = context.getNodeDistribution(children.get(i).getPlanNodeId());
- if (first.region == null || !first.region.equals(next.region)) {
- return false;
- }
- }
- return true;
- }
-
- public PlanNode visit(PlanNode node, NodeGroupContext context) {
- return node.accept(this, context);
- }
- }
-
- private class NodeGroupContext {
- private MPPQueryContext queryContext;
- private Map<PlanNodeId, NodeDistribution> nodeDistributionMap;
-
- public NodeGroupContext(MPPQueryContext queryContext) {
- this.queryContext = queryContext;
- this.nodeDistributionMap = new HashMap<>();
- }
-
- public void putNodeDistribution(PlanNodeId nodeId, NodeDistribution distribution) {
- this.nodeDistributionMap.put(nodeId, distribution);
- }
-
- public NodeDistribution getNodeDistribution(PlanNodeId nodeId) {
- return this.nodeDistributionMap.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 TRegionReplicaSet region;
-
- private NodeDistribution(NodeDistributionType type, TRegionReplicaSet region) {
- this.type = type;
- this.region = region;
- }
-
- private NodeDistribution(NodeDistributionType type) {
- this.type = type;
- }
- }
-
- private class FragmentBuilder {
- private MPPQueryContext context;
-
- public FragmentBuilder(MPPQueryContext context) {
- this.context = context;
- }
-
- public SubPlan splitToSubPlan(PlanNode root) {
- SubPlan rootSubPlan = createSubPlan(root);
- splitToSubPlan(root, rootSubPlan);
- return rootSubPlan;
- }
-
- private void splitToSubPlan(PlanNode root, SubPlan subPlan) {
- // TODO: (xingtanzjr) we apply no action for IWritePlanNode currently
- if (root instanceof WritePlanNode) {
- return;
- }
- if (root instanceof ExchangeNode) {
- // We add a FragmentSinkNode for newly created PlanFragment
- ExchangeNode exchangeNode = (ExchangeNode) root;
- FragmentSinkNode sinkNode = new FragmentSinkNode(context.getQueryId().genPlanNodeId());
- sinkNode.setChild(exchangeNode.getChild());
- sinkNode.setDownStreamPlanNodeId(exchangeNode.getPlanNodeId());
-
- // Record the source node info in the ExchangeNode so that we can keep the connection of
- // these nodes/fragments
- exchangeNode.setRemoteSourceNode(sinkNode);
- // We cut off the subtree to make the ExchangeNode as the leaf node of current PlanFragment
- exchangeNode.cleanChildren();
-
- // Build the child SubPlan Tree
- SubPlan childSubPlan = createSubPlan(sinkNode);
- splitToSubPlan(sinkNode, childSubPlan);
-
- subPlan.addChild(childSubPlan);
- return;
- }
- for (PlanNode child : root.getChildren()) {
- splitToSubPlan(child, subPlan);
- }
- }
-
- private SubPlan createSubPlan(PlanNode root) {
- PlanFragment fragment = new PlanFragment(getNextFragmentId(), root);
- return new SubPlan(fragment);
- }
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
index 23d7ee3286..68b0a39237 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanBuilder.java
@@ -45,6 +45,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchS
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.TimeSeriesCountNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.DeleteTimeSeriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.InvalidateSchemaCacheNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.AggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
@@ -60,6 +62,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesAggreg
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.FillDescriptor;
@@ -599,4 +602,29 @@ public class LogicalPlanBuilder {
this.root = memorySourceNode;
return this;
}
+
+ public LogicalPlanBuilder planInvalidateSchemaCache(
+ List<PartialPath> paths, List<String> storageGroups) {
+ this.root =
+ new InvalidateSchemaCacheNode(
+ context.getQueryId().genPlanNodeId(), context.getQueryId(), paths, storageGroups);
+ return this;
+ }
+
+ public LogicalPlanBuilder planDeleteData(List<PartialPath> paths, List<String> storageGroups) {
+ DeleteDataNode node =
+ new DeleteDataNode(
+ context.getQueryId().genPlanNodeId(), context.getQueryId(), paths, storageGroups);
+ node.addChild(this.root);
+ this.root = node;
+ return this;
+ }
+
+ public LogicalPlanBuilder planDeleteTimeseries(List<PartialPath> paths) {
+ DeleteTimeSeriesNode node =
+ new DeleteTimeSeriesNode(context.getQueryId().genPlanNodeId(), paths);
+ node.addChild(this.root);
+ this.root = node;
+ return this;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
index 3a757fb292..e3f332b877 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.mpp.plan.planner;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.confignode.rpc.thrift.NodeManagementType;
import org.apache.iotdb.db.metadata.utils.TimeseriesVersionUtil;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
@@ -53,6 +54,7 @@ import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateAlignedTimeSeriesSt
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateMultiTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesByDeviceStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.SchemaFetchStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildNodesStatement;
import org.apache.iotdb.db.mpp.plan.statement.metadata.ShowChildPathsStatement;
@@ -577,5 +579,15 @@ public class LogicalPlanner {
.planNodeManagementMemoryMerge(analysis.getMatchedNodes(), NodeManagementType.CHILD_NODES)
.getRoot();
}
+
+ @Override
+ public PlanNode visitDeleteTimeseries(
+ DeleteTimeSeriesStatement deleteTimeSeriesStatement, MPPQueryContext context) {
+ LogicalPlanBuilder planBuilder = new LogicalPlanBuilder(context);
+ List<PartialPath> paths = deleteTimeSeriesStatement.getPaths();
+ List<String> storageGroups =
+ new ArrayList<>(analysis.getSchemaPartitionInfo().getSchemaPartitionMap().keySet());
+ return planBuilder.planDeleteData(paths, storageGroups).planDeleteTimeseries(paths).getRoot();
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanContext.java
similarity index 64%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanContext.java
index 0d6ee0c793..6f9e16e6ee 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanContext.java
@@ -17,20 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner.plan.node;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
-import java.util.List;
+public class DistributionPlanContext {
+ protected MPPQueryContext queryContext;
-public abstract class WritePlanNode extends PlanNode {
-
- protected WritePlanNode(PlanNodeId id) {
- super(id);
+ protected DistributionPlanContext(MPPQueryContext queryContext) {
+ this.queryContext = queryContext;
}
-
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
- public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
new file mode 100644
index 0000000000..3be4a6daf6
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
@@ -0,0 +1,179 @@
+/*
+ * 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.planner.distribution;
+
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.common.PlanFragmentId;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.planner.IFragmentParallelPlaner;
+import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
+import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.PlanFragment;
+import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
+import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+
+import java.util.List;
+
+public class DistributionPlanner {
+ private Analysis analysis;
+ private MPPQueryContext context;
+ private LogicalQueryPlan logicalPlan;
+
+ private int planFragmentIndex = 0;
+
+ public DistributionPlanner(Analysis analysis, LogicalQueryPlan logicalPlan) {
+ this.analysis = analysis;
+ this.logicalPlan = logicalPlan;
+ this.context = logicalPlan.getContext();
+ }
+
+ public PlanNode rewriteSource() {
+ SourceRewriter rewriter = new SourceRewriter(this.analysis);
+ return rewriter.visit(logicalPlan.getRootNode(), new DistributionPlanContext(context));
+ }
+
+ public PlanNode addExchangeNode(PlanNode root) {
+ ExchangeNodeAdder adder = new ExchangeNodeAdder();
+ return adder.visit(root, new NodeGroupContext(context));
+ }
+
+ public SubPlan splitFragment(PlanNode root) {
+ FragmentBuilder fragmentBuilder = new FragmentBuilder(context);
+ return fragmentBuilder.splitToSubPlan(root);
+ }
+
+ public DistributedQueryPlan planFragments() {
+ PlanNode rootAfterRewrite = rewriteSource();
+ System.out.println(PlanNodeUtil.nodeToString(rootAfterRewrite));
+ PlanNode rootWithExchange = addExchangeNode(rootAfterRewrite);
+ System.out.println(PlanNodeUtil.nodeToString(rootWithExchange));
+ if (analysis.getStatement() instanceof QueryStatement) {
+ analysis
+ .getRespDatasetHeader()
+ .setColumnToTsBlockIndexMap(rootWithExchange.getOutputColumnNames());
+ }
+ SubPlan subPlan = splitFragment(rootWithExchange);
+ List<FragmentInstance> fragmentInstances = planFragmentInstances(subPlan);
+ // Only execute this step for READ operation
+ if (context.getQueryType() == QueryType.READ) {
+ SetSinkForRootInstance(subPlan, fragmentInstances);
+ }
+ return new DistributedQueryPlan(
+ logicalPlan.getContext(), subPlan, subPlan.getPlanFragmentList(), fragmentInstances);
+ }
+
+ // Convert fragment to detailed instance
+ // And for parallel-able fragment, clone it into several instances with different params.
+ public List<FragmentInstance> planFragmentInstances(SubPlan subPlan) {
+ IFragmentParallelPlaner parallelPlaner =
+ context.getQueryType() == QueryType.READ
+ ? new SimpleFragmentParallelPlanner(subPlan, analysis, context)
+ : new WriteFragmentParallelPlanner(subPlan, analysis, context);
+ return parallelPlaner.parallelPlan();
+ }
+
+ // TODO: (xingtanzjr) Maybe we should handle ResultNode in LogicalPlanner ?
+ public void SetSinkForRootInstance(SubPlan subPlan, List<FragmentInstance> instances) {
+ FragmentInstance rootInstance = null;
+ for (FragmentInstance instance : instances) {
+ if (instance.getFragment().getId().equals(subPlan.getPlanFragment().getId())) {
+ rootInstance = instance;
+ break;
+ }
+ }
+ // root should not be null during normal process
+ if (rootInstance == null) {
+ return;
+ }
+
+ FragmentSinkNode sinkNode = new FragmentSinkNode(context.getQueryId().genPlanNodeId());
+ sinkNode.setDownStream(
+ context.getLocalDataBlockEndpoint(),
+ context.getResultNodeContext().getVirtualFragmentInstanceId(),
+ context.getResultNodeContext().getVirtualResultNodeId());
+ sinkNode.setChild(rootInstance.getFragment().getRoot());
+ context
+ .getResultNodeContext()
+ .setUpStream(
+ rootInstance.getHostDataNode().dataBlockManagerEndPoint,
+ rootInstance.getId(),
+ sinkNode.getPlanNodeId());
+ rootInstance.getFragment().setRoot(sinkNode);
+ }
+
+ private PlanFragmentId getNextFragmentId() {
+ return new PlanFragmentId(this.logicalPlan.getContext().getQueryId(), this.planFragmentIndex++);
+ }
+
+ private class FragmentBuilder {
+ private MPPQueryContext context;
+
+ public FragmentBuilder(MPPQueryContext context) {
+ this.context = context;
+ }
+
+ public SubPlan splitToSubPlan(PlanNode root) {
+ SubPlan rootSubPlan = createSubPlan(root);
+ splitToSubPlan(root, rootSubPlan);
+ return rootSubPlan;
+ }
+
+ private void splitToSubPlan(PlanNode root, SubPlan subPlan) {
+ // TODO: (xingtanzjr) we apply no action for IWritePlanNode currently
+ if (root instanceof WritePlanNode) {
+ return;
+ }
+ if (root instanceof ExchangeNode) {
+ // We add a FragmentSinkNode for newly created PlanFragment
+ ExchangeNode exchangeNode = (ExchangeNode) root;
+ FragmentSinkNode sinkNode = new FragmentSinkNode(context.getQueryId().genPlanNodeId());
+ sinkNode.setChild(exchangeNode.getChild());
+ sinkNode.setDownStreamPlanNodeId(exchangeNode.getPlanNodeId());
+
+ // Record the source node info in the ExchangeNode so that we can keep the connection of
+ // these nodes/fragments
+ exchangeNode.setRemoteSourceNode(sinkNode);
+ // We cut off the subtree to make the ExchangeNode as the leaf node of current PlanFragment
+ exchangeNode.cleanChildren();
+
+ // Build the child SubPlan Tree
+ SubPlan childSubPlan = createSubPlan(sinkNode);
+ splitToSubPlan(sinkNode, childSubPlan);
+
+ subPlan.addChild(childSubPlan);
+ return;
+ }
+ for (PlanNode child : root.getChildren()) {
+ splitToSubPlan(child, subPlan);
+ }
+ }
+
+ private SubPlan createSubPlan(PlanNode root) {
+ PlanFragment fragment = new PlanFragment(getNextFragmentId(), root);
+ return new SubPlan(fragment);
+ }
+ }
+}
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
new file mode 100644
index 0000000000..daef502c7b
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/ExchangeNodeAdder.java
@@ -0,0 +1,274 @@
+/*
+ * 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.planner.distribution;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.AbstractSchemaMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.CountSchemaMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
+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.ExchangeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
+ @Override
+ public PlanNode visitPlan(PlanNode node, NodeGroupContext context) {
+ // TODO: (xingtanzjr) we apply no action for IWritePlanNode currently
+ if (node instanceof WritePlanNode) {
+ return node;
+ }
+ // 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.getPlanNodeId(),
+ new NodeDistribution(NodeDistributionType.SAME_WITH_ALL_CHILDREN, null));
+
+ return node.cloneWithChildren(children);
+ }
+
+ @Override
+ public PlanNode visitSchemaQueryMerge(SchemaQueryMergeNode node, NodeGroupContext context) {
+ return internalVisitSchemaMerge(node, context);
+ }
+
+ private PlanNode internalVisitSchemaMerge(
+ AbstractSchemaMergeNode node, NodeGroupContext context) {
+ node.getChildren()
+ .forEach(
+ child -> {
+ visit(child, context);
+ });
+ NodeDistribution nodeDistribution =
+ new NodeDistribution(NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN);
+ PlanNode newNode = node.clone();
+ nodeDistribution.region = calculateSchemaRegionByChildren(node.getChildren(), context);
+ context.putNodeDistribution(newNode.getPlanNodeId(), nodeDistribution);
+ node.getChildren()
+ .forEach(
+ child -> {
+ if (!nodeDistribution.region.equals(
+ context.getNodeDistribution(child.getPlanNodeId()).region)) {
+ ExchangeNode exchangeNode =
+ new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.setChild(child);
+ exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(child);
+ }
+ });
+ return newNode;
+ }
+
+ @Override
+ public PlanNode visitCountMerge(CountSchemaMergeNode node, NodeGroupContext context) {
+ return internalVisitSchemaMerge(node, context);
+ }
+
+ @Override
+ public PlanNode visitSchemaFetchMerge(SchemaFetchMergeNode node, NodeGroupContext context) {
+ return internalVisitSchemaMerge(node, context);
+ }
+
+ @Override
+ public PlanNode visitSchemaQueryScan(SchemaQueryScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitSchemaFetchScan(SchemaFetchScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitSeriesScan(SeriesScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitAlignedSeriesScan(AlignedSeriesScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ public PlanNode visitSeriesAggregationScan(
+ SeriesAggregationScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ public PlanNode visitAlignedSeriesAggregationScan(
+ AlignedSeriesAggregationScanNode node, NodeGroupContext context) {
+ return processNoChildSourceNode(node, context);
+ }
+
+ private PlanNode processNoChildSourceNode(SourceNode node, NodeGroupContext context) {
+ context.putNodeDistribution(
+ node.getPlanNodeId(),
+ new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
+ return node.clone();
+ }
+
+ public PlanNode visitDeleteData(DeleteDataNode node, NodeGroupContext context) {
+ context.putNodeDistribution(
+ node.getPlanNodeId(),
+ new NodeDistribution(NodeDistributionType.NO_CHILD, node.getRegionReplicaSet()));
+ return node;
+ }
+
+ public PlanNode visitDeleteTimeseries(DeleteTimeSeriesNode node, NodeGroupContext context) {
+ List<PlanNode> visitedChildren = new ArrayList<>();
+ node.getChildren()
+ .forEach(
+ child -> {
+ visitedChildren.add(visit(child, context));
+ });
+ node.getChildren().clear();
+ visitedChildren.forEach(
+ child -> {
+ ExchangeNode exchangeNode =
+ new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.setChild(child);
+ exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
+ node.addChild(exchangeNode);
+ });
+ return node;
+ }
+
+ @Override
+ public PlanNode visitTimeJoin(TimeJoinNode node, NodeGroupContext context) {
+ return processMultiChildNode(node, context);
+ }
+
+ public PlanNode visitRowBasedSeriesAggregate(AggregationNode node, NodeGroupContext context) {
+ return processMultiChildNode(node, context);
+ }
+
+ @Override
+ public PlanNode visitGroupByLevel(GroupByLevelNode node, NodeGroupContext context) {
+ return processMultiChildNode(node, context);
+ }
+
+ private PlanNode processMultiChildNode(MultiChildNode node, NodeGroupContext context) {
+ MultiChildNode newNode = (MultiChildNode) node.clone();
+ List<PlanNode> visitedChildren = new ArrayList<>();
+ node.getChildren()
+ .forEach(
+ child -> {
+ visitedChildren.add(visit(child, context));
+ });
+
+ TRegionReplicaSet dataRegion = calculateDataRegionByChildren(visitedChildren, context);
+ NodeDistributionType distributionType =
+ nodeDistributionIsSame(visitedChildren, context)
+ ? NodeDistributionType.SAME_WITH_ALL_CHILDREN
+ : NodeDistributionType.SAME_WITH_SOME_CHILD;
+ context.putNodeDistribution(
+ newNode.getPlanNodeId(), 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.getPlanNodeId()).region)) {
+ ExchangeNode exchangeNode =
+ new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.setChild(child);
+ exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(child);
+ }
+ });
+ return newNode;
+ }
+
+ private TRegionReplicaSet calculateDataRegionByChildren(
+ List<PlanNode> children, NodeGroupContext context) {
+ // Step 1: calculate the count of children group by DataRegion.
+ Map<TRegionReplicaSet, Long> groupByRegion =
+ children.stream()
+ .collect(
+ Collectors.groupingBy(
+ child -> context.getNodeDistribution(child.getPlanNodeId()).region,
+ Collectors.counting()));
+ // Step 2: return the RegionReplicaSet with max count
+ return Collections.max(groupByRegion.entrySet(), Map.Entry.comparingByValue()).getKey();
+ }
+
+ private TRegionReplicaSet calculateSchemaRegionByChildren(
+ List<PlanNode> children, NodeGroupContext context) {
+ // We always make the schemaRegion of MetaMergeNode to be the same as its first child.
+ return context.getNodeDistribution(children.get(0).getPlanNodeId()).region;
+ }
+
+ 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).getPlanNodeId());
+ for (int i = 1; i < children.size(); i++) {
+ NodeDistribution next = context.getNodeDistribution(children.get(i).getPlanNodeId());
+ if (first.region == null || !first.region.equals(next.region)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public PlanNode visit(PlanNode node, NodeGroupContext context) {
+ return node.accept(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistribution.java
similarity index 69%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistribution.java
index 0d6ee0c793..fb4561460d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistribution.java
@@ -17,20 +17,20 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner.plan.node;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
-import java.util.List;
+public class NodeDistribution {
+ protected NodeDistributionType type;
+ protected TRegionReplicaSet region;
-public abstract class WritePlanNode extends PlanNode {
-
- protected WritePlanNode(PlanNodeId id) {
- super(id);
+ protected NodeDistribution(NodeDistributionType type, TRegionReplicaSet region) {
+ this.type = type;
+ this.region = region;
}
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
- public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
+ protected NodeDistribution(NodeDistributionType type) {
+ this.type = type;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistributionType.java
similarity index 64%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistributionType.java
index 0d6ee0c793..be34998958 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeDistributionType.java
@@ -17,20 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner.plan.node;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
-
-import java.util.List;
-
-public abstract class WritePlanNode extends PlanNode {
-
- protected WritePlanNode(PlanNodeId id) {
- super(id);
- }
-
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
- public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
+public enum NodeDistributionType {
+ SAME_WITH_ALL_CHILDREN,
+ SAME_WITH_SOME_CHILD,
+ DIFFERENT_FROM_ALL_CHILDREN,
+ NO_CHILD,
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeGroupContext.java
similarity index 55%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeGroupContext.java
index e3209a03e4..ba3ddb400c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/NodeGroupContext.java
@@ -16,21 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner.plan.node.source;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
+
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
-public abstract class SourceNode extends PlanNode implements AutoCloseable {
+import java.util.HashMap;
+import java.util.Map;
- public SourceNode(PlanNodeId id) {
- super(id);
- }
+public class NodeGroupContext {
+ protected MPPQueryContext queryContext;
+ protected Map<PlanNodeId, NodeDistribution> nodeDistributionMap;
- public abstract void open() throws Exception;
+ public NodeGroupContext(MPPQueryContext queryContext) {
+ this.queryContext = queryContext;
+ this.nodeDistributionMap = new HashMap<>();
+ }
- public abstract TRegionReplicaSet getRegionReplicaSet();
+ public void putNodeDistribution(PlanNodeId nodeId, NodeDistribution distribution) {
+ this.nodeDistributionMap.put(nodeId, distribution);
+ }
- public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
+ public NodeDistribution getNodeDistribution(PlanNodeId nodeId) {
+ return this.nodeDistributionMap.get(nodeId);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/SimpleFragmentParallelPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/SimpleFragmentParallelPlanner.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
index 5b06be36fe..fd991196df 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/SimpleFragmentParallelPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
@@ -16,12 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.IFragmentParallelPlaner;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.mpp.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
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
new file mode 100644
index 0000000000..d72a1d1b60
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SourceRewriter.java
@@ -0,0 +1,590 @@
+/*
+ * 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.planner.distribution;
+
+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.schematree.PathPatternTree;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.SimplePlanNodeRewriter;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.CountSchemaMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaFetchScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
+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.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationSourceNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesSourceNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByLevelDescriptor;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
+
+public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanContext> {
+
+ private Analysis analysis;
+
+ public SourceRewriter(Analysis analysis) {
+ this.analysis = analysis;
+ }
+
+ // TODO: (xingtanzjr) implement the method visitDeviceMergeNode()
+ public PlanNode visitDeviceMerge(TimeJoinNode node, DistributionPlanContext context) {
+ return null;
+ }
+
+ public PlanNode visitDeleteTimeseries(
+ DeleteTimeSeriesNode node, DistributionPlanContext context) {
+ // Step 1: split DeleteDataNode by partition
+ checkArgument(node.getChildren().size() == 1, "DeleteTimeSeriesNode should have 1 child");
+ checkArgument(
+ node.getChildren().get(0) instanceof DeleteDataNode,
+ "Child of DeleteTimeSeriesNode should be DeleteDataNode");
+
+ DeleteDataNode deleteDataNode = (DeleteDataNode) node.getChildren().get(0);
+ List<DeleteDataNode> deleteDataNodes = splitDeleteDataNode(deleteDataNode, context);
+
+ // Step 2: split DeleteTimeseriesNode by partition
+ List<DeleteTimeSeriesNode> deleteTimeSeriesNodes = splitDeleteTimeseries(node, context);
+
+ // Step 3: construct them as a Tree
+ checkArgument(
+ deleteTimeSeriesNodes.size() > 0,
+ "Size of DeleteTimeseriesNode splits should be larger than 0");
+ deleteDataNodes.forEach(split -> deleteTimeSeriesNodes.get(0).addChild(split));
+ for (int i = 1; i < deleteTimeSeriesNodes.size(); i++) {
+ deleteTimeSeriesNodes.get(i).addChild(deleteTimeSeriesNodes.get(i - 1));
+ }
+ return deleteTimeSeriesNodes.get(deleteTimeSeriesNodes.size() - 1);
+ }
+
+ private List<DeleteTimeSeriesNode> splitDeleteTimeseries(
+ DeleteTimeSeriesNode node, DistributionPlanContext context) {
+ List<DeleteTimeSeriesNode> ret = new ArrayList<>();
+ List<PartialPath> rawPaths = node.getPathList();
+ List<RegionReplicaSetInfo> relatedRegions =
+ analysis.getSchemaPartitionInfo().getSchemaDistributionInfo();
+ for (RegionReplicaSetInfo regionReplicaSetInfo : relatedRegions) {
+ List<PartialPath> newPaths =
+ getRelatedPaths(rawPaths, regionReplicaSetInfo.getOwnedStorageGroups());
+ DeleteTimeSeriesNode split =
+ new DeleteTimeSeriesNode(context.queryContext.getQueryId().genPlanNodeId(), newPaths);
+ split.setRegionReplicaSet(regionReplicaSetInfo.getRegionReplicaSet());
+ ret.add(split);
+ }
+ return ret;
+ }
+
+ private List<DeleteDataNode> splitDeleteDataNode(
+ DeleteDataNode node, DistributionPlanContext context) {
+ List<DeleteDataNode> ret = new ArrayList<>();
+ List<PartialPath> rawPaths = node.getPathList();
+ List<RegionReplicaSetInfo> relatedRegions =
+ analysis.getDataPartitionInfo().getDataDistributionInfo();
+ for (RegionReplicaSetInfo regionReplicaSetInfo : relatedRegions) {
+ List<PartialPath> newPaths =
+ getRelatedPaths(rawPaths, regionReplicaSetInfo.getOwnedStorageGroups());
+ DeleteDataNode split =
+ new DeleteDataNode(
+ context.queryContext.getQueryId().genPlanNodeId(),
+ context.queryContext.getQueryId(),
+ newPaths,
+ regionReplicaSetInfo.getOwnedStorageGroups());
+ split.setRegionReplicaSet(regionReplicaSetInfo.getRegionReplicaSet());
+ ret.add(split);
+ }
+ return ret;
+ }
+
+ private List<PartialPath> getRelatedPaths(List<PartialPath> paths, List<String> storageGroups) {
+ List<PartialPath> ret = new ArrayList<>();
+ PathPatternTree patternTree = new PathPatternTree(paths);
+ for (String storageGroup : storageGroups) {
+ try {
+ ret.addAll(
+ patternTree.findOverlappedPaths(
+ new PartialPath(storageGroup).concatNode(MULTI_LEVEL_PATH_WILDCARD)));
+ } catch (IllegalPathException e) {
+ // The IllegalPathException is definitely not threw here
+ throw new RuntimeException(e);
+ }
+ }
+ return ret;
+ }
+
+ @Override
+ public PlanNode visitSchemaQueryMerge(
+ SchemaQueryMergeNode node, DistributionPlanContext context) {
+ SchemaQueryMergeNode root = (SchemaQueryMergeNode) node.clone();
+ SchemaQueryScanNode seed = (SchemaQueryScanNode) node.getChildren().get(0);
+ TreeSet<TRegionReplicaSet> schemaRegions =
+ new TreeSet<>(Comparator.comparingInt(region -> region.getRegionId().getId()));
+ analysis
+ .getSchemaPartitionInfo()
+ .getSchemaPartitionMap()
+ .forEach(
+ (storageGroup, deviceGroup) -> {
+ deviceGroup.forEach(
+ (deviceGroupId, schemaRegionReplicaSet) ->
+ schemaRegions.add(schemaRegionReplicaSet));
+ });
+ int count = schemaRegions.size();
+ schemaRegions.forEach(
+ region -> {
+ SchemaQueryScanNode schemaQueryScanNode = (SchemaQueryScanNode) seed.clone();
+ schemaQueryScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ schemaQueryScanNode.setRegionReplicaSet(region);
+ if (count > 1) {
+ schemaQueryScanNode.setLimit(
+ schemaQueryScanNode.getOffset() + schemaQueryScanNode.getLimit());
+ schemaQueryScanNode.setOffset(0);
+ }
+ root.addChild(schemaQueryScanNode);
+ });
+ return root;
+ }
+
+ @Override
+ public PlanNode visitCountMerge(CountSchemaMergeNode node, DistributionPlanContext context) {
+ CountSchemaMergeNode root = (CountSchemaMergeNode) node.clone();
+ SchemaQueryScanNode seed = (SchemaQueryScanNode) node.getChildren().get(0);
+ Set<TRegionReplicaSet> schemaRegions = new HashSet<>();
+ analysis
+ .getSchemaPartitionInfo()
+ .getSchemaPartitionMap()
+ .forEach(
+ (storageGroup, deviceGroup) -> {
+ deviceGroup.forEach(
+ (deviceGroupId, schemaRegionReplicaSet) ->
+ schemaRegions.add(schemaRegionReplicaSet));
+ });
+ schemaRegions.forEach(
+ region -> {
+ SchemaQueryScanNode schemaQueryScanNode = (SchemaQueryScanNode) seed.clone();
+ schemaQueryScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ schemaQueryScanNode.setRegionReplicaSet(region);
+ root.addChild(schemaQueryScanNode);
+ });
+ return root;
+ }
+
+ // TODO: (xingtanzjr) a temporary way to resolve the distribution of single SeriesScanNode issue
+ @Override
+ public PlanNode visitSeriesScan(SeriesScanNode node, DistributionPlanContext context) {
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(node.getSeriesPath(), node.getTimeFilter());
+ if (dataDistribution.size() == 1) {
+ node.setRegionReplicaSet(dataDistribution.get(0));
+ return node;
+ }
+ TimeJoinNode timeJoinNode =
+ new TimeJoinNode(context.queryContext.getQueryId().genPlanNodeId(), node.getScanOrder());
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ SeriesScanNode split = (SeriesScanNode) node.clone();
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ timeJoinNode.addChild(split);
+ }
+ return timeJoinNode;
+ }
+
+ @Override
+ public PlanNode visitSeriesAggregationScan(
+ SeriesAggregationScanNode node, DistributionPlanContext context) {
+ return processSeriesAggregationSource(node, context);
+ }
+
+ @Override
+ public PlanNode visitAlignedSeriesAggregationScan(
+ AlignedSeriesAggregationScanNode node, DistributionPlanContext context) {
+ return processSeriesAggregationSource(node, context);
+ }
+
+ private PlanNode processSeriesAggregationSource(
+ SeriesAggregationSourceNode node, DistributionPlanContext context) {
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(node.getPartitionPath(), node.getPartitionTimeFilter());
+ if (dataDistribution.size() == 1) {
+ node.setRegionReplicaSet(dataDistribution.get(0));
+ return node;
+ }
+ List<AggregationDescriptor> leafAggDescriptorList = new ArrayList<>();
+ node.getAggregationDescriptorList()
+ .forEach(
+ descriptor -> {
+ leafAggDescriptorList.add(
+ new AggregationDescriptor(
+ descriptor.getAggregationType(),
+ AggregationStep.PARTIAL,
+ descriptor.getInputExpressions()));
+ });
+
+ List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
+ node.getAggregationDescriptorList()
+ .forEach(
+ descriptor -> {
+ rootAggDescriptorList.add(
+ new AggregationDescriptor(
+ descriptor.getAggregationType(),
+ AggregationStep.FINAL,
+ descriptor.getInputExpressions()));
+ });
+
+ AggregationNode aggregationNode =
+ new AggregationNode(
+ context.queryContext.getQueryId().genPlanNodeId(), rootAggDescriptorList);
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ SeriesAggregationScanNode split = (SeriesAggregationScanNode) node.clone();
+ split.setAggregationDescriptorList(leafAggDescriptorList);
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ aggregationNode.addChild(split);
+ }
+ return aggregationNode;
+ }
+
+ @Override
+ public PlanNode visitAlignedSeriesScan(
+ AlignedSeriesScanNode node, DistributionPlanContext context) {
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(node.getAlignedPath(), node.getTimeFilter());
+ if (dataDistribution.size() == 1) {
+ node.setRegionReplicaSet(dataDistribution.get(0));
+ return node;
+ }
+ TimeJoinNode timeJoinNode =
+ new TimeJoinNode(context.queryContext.getQueryId().genPlanNodeId(), node.getScanOrder());
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ AlignedSeriesScanNode split = (AlignedSeriesScanNode) node.clone();
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ timeJoinNode.addChild(split);
+ }
+ return timeJoinNode;
+ }
+
+ @Override
+ public PlanNode visitSchemaFetchMerge(
+ SchemaFetchMergeNode node, DistributionPlanContext context) {
+ SchemaFetchMergeNode root = (SchemaFetchMergeNode) node.clone();
+ Map<String, Set<TRegionReplicaSet>> storageGroupSchemaRegionMap = new HashMap<>();
+ analysis
+ .getSchemaPartitionInfo()
+ .getSchemaPartitionMap()
+ .forEach(
+ (storageGroup, deviceGroup) -> {
+ storageGroupSchemaRegionMap.put(storageGroup, new HashSet<>());
+ deviceGroup.forEach(
+ (deviceGroupId, schemaRegionReplicaSet) ->
+ storageGroupSchemaRegionMap.get(storageGroup).add(schemaRegionReplicaSet));
+ });
+
+ for (PlanNode child : node.getChildren()) {
+ for (TRegionReplicaSet schemaRegion :
+ storageGroupSchemaRegionMap.get(
+ ((SchemaFetchScanNode) child).getStorageGroup().getFullPath())) {
+ SchemaFetchScanNode schemaFetchScanNode = (SchemaFetchScanNode) child.clone();
+ schemaFetchScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ schemaFetchScanNode.setRegionReplicaSet(schemaRegion);
+ root.addChild(schemaFetchScanNode);
+ }
+ }
+ return root;
+ }
+
+ @Override
+ public PlanNode visitTimeJoin(TimeJoinNode node, DistributionPlanContext context) {
+ // Although some logic is similar between Aggregation and RawDataQuery,
+ // we still use separate method to process the distribution planning now
+ // to make the planning procedure more clear
+ if (isAggregationQuery(node)) {
+ return planAggregationWithTimeJoin(node, context);
+ }
+
+ TimeJoinNode root = (TimeJoinNode) node.clone();
+ // Step 1: Get all source nodes. For the node which is not source, add it as the child of
+ // current TimeJoinNode
+ List<SourceNode> sources = new ArrayList<>();
+ for (PlanNode child : node.getChildren()) {
+ if (child instanceof SeriesSourceNode) {
+ // If the child is SeriesScanNode, we need to check whether this node should be seperated
+ // into several splits.
+ SeriesSourceNode handle = (SeriesSourceNode) child;
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(handle.getPartitionPath(), handle.getPartitionTimeFilter());
+ // If the size of dataDistribution is m, this SeriesScanNode should be seperated into m
+ // SeriesScanNode.
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ SeriesSourceNode split = (SeriesSourceNode) handle.clone();
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ sources.add(split);
+ }
+ }
+ }
+ // Step 2: For the source nodes, group them by the DataRegion.
+ Map<TRegionReplicaSet, List<SourceNode>> sourceGroup =
+ sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
+
+ // Step 3: For the source nodes which belong to same data region, add a TimeJoinNode for them
+ // and make the
+ // new TimeJoinNode as the child of current TimeJoinNode
+ // TODO: (xingtanzjr) optimize the procedure here to remove duplicated TimeJoinNode
+ final boolean[] addParent = {false};
+ sourceGroup.forEach(
+ (dataRegion, seriesScanNodes) -> {
+ if (seriesScanNodes.size() == 1) {
+ root.addChild(seriesScanNodes.get(0));
+ } else {
+ if (!addParent[0]) {
+ seriesScanNodes.forEach(root::addChild);
+ addParent[0] = true;
+ } else {
+ // We clone a TimeJoinNode from root to make the params to be consistent.
+ // But we need to assign a new ID to it
+ TimeJoinNode parentOfGroup = (TimeJoinNode) root.clone();
+ parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ seriesScanNodes.forEach(parentOfGroup::addChild);
+ root.addChild(parentOfGroup);
+ }
+ }
+ });
+
+ // Process the other children which are not SeriesSourceNode
+ for (PlanNode child : node.getChildren()) {
+ if (!(child instanceof SeriesSourceNode)) {
+ // In a general logical query plan, the children of TimeJoinNode should only be
+ // SeriesScanNode or SeriesAggregateScanNode
+ // So this branch should not be touched.
+ root.addChild(visit(child, context));
+ }
+ }
+
+ return root;
+ }
+
+ private boolean isAggregationQuery(TimeJoinNode node) {
+ for (PlanNode child : node.getChildren()) {
+ if (child instanceof SeriesAggregationScanNode
+ || child instanceof AlignedSeriesAggregationScanNode) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private PlanNode planAggregationWithTimeJoin(TimeJoinNode root, DistributionPlanContext context) {
+
+ List<SeriesAggregationSourceNode> sources = splitAggregationSourceByPartition(root, context);
+ Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
+ sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
+
+ // construct AggregationDescriptor for AggregationNode
+ List<AggregationDescriptor> rootAggDescriptorList = new ArrayList<>();
+ for (PlanNode child : root.getChildren()) {
+ SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
+ handle
+ .getAggregationDescriptorList()
+ .forEach(
+ descriptor -> {
+ rootAggDescriptorList.add(
+ new AggregationDescriptor(
+ descriptor.getAggregationType(),
+ AggregationStep.FINAL,
+ descriptor.getInputExpressions()));
+ });
+ }
+ AggregationNode aggregationNode =
+ new AggregationNode(
+ context.queryContext.getQueryId().genPlanNodeId(), rootAggDescriptorList);
+
+ final boolean[] addParent = {false};
+ sourceGroup.forEach(
+ (dataRegion, sourceNodes) -> {
+ if (sourceNodes.size() == 1) {
+ aggregationNode.addChild(sourceNodes.get(0));
+ } else {
+ if (!addParent[0]) {
+ sourceNodes.forEach(aggregationNode::addChild);
+ addParent[0] = true;
+ } else {
+ // We clone a TimeJoinNode from root to make the params to be consistent.
+ // But we need to assign a new ID to it
+ TimeJoinNode parentOfGroup = (TimeJoinNode) root.clone();
+ parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ sourceNodes.forEach(parentOfGroup::addChild);
+ aggregationNode.addChild(parentOfGroup);
+ }
+ }
+ });
+
+ return aggregationNode;
+ }
+
+ public PlanNode visitGroupByLevel(GroupByLevelNode root, DistributionPlanContext context) {
+ // Firstly, we build the tree structure for GroupByLevelNode
+ List<SeriesAggregationSourceNode> sources = splitAggregationSourceByPartition(root, context);
+ Map<TRegionReplicaSet, List<SeriesAggregationSourceNode>> sourceGroup =
+ sources.stream().collect(Collectors.groupingBy(SourceNode::getRegionReplicaSet));
+
+ GroupByLevelNode newRoot = (GroupByLevelNode) root.clone();
+ final boolean[] addParent = {false};
+ sourceGroup.forEach(
+ (dataRegion, sourceNodes) -> {
+ if (sourceNodes.size() == 1) {
+ newRoot.addChild(sourceNodes.get(0));
+ } else {
+ if (!addParent[0]) {
+ sourceNodes.forEach(newRoot::addChild);
+ addParent[0] = true;
+ } else {
+ // We clone a TimeJoinNode from root to make the params to be consistent.
+ // But we need to assign a new ID to it
+ GroupByLevelNode parentOfGroup = (GroupByLevelNode) root.clone();
+ parentOfGroup.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ sourceNodes.forEach(parentOfGroup::addChild);
+ newRoot.addChild(parentOfGroup);
+ }
+ }
+ });
+
+ // Then, we calculate the attributes for GroupByLevelNode in each level
+ calculateGroupByLevelNodeAttributes(newRoot, 0);
+ return newRoot;
+ }
+
+ private void calculateGroupByLevelNodeAttributes(PlanNode node, int level) {
+ if (node == null) {
+ return;
+ }
+ node.getChildren().forEach(child -> calculateGroupByLevelNodeAttributes(child, level + 1));
+ if (!(node instanceof GroupByLevelNode)) {
+ return;
+ }
+ GroupByLevelNode handle = (GroupByLevelNode) node;
+
+ // Construct all outputColumns from children. Using Set here to avoid duplication
+ Set<String> childrenOutputColumns = new HashSet<>();
+ handle
+ .getChildren()
+ .forEach(child -> childrenOutputColumns.addAll(child.getOutputColumnNames()));
+
+ // Check every OutputColumn of GroupByLevelNode and set the Expression of corresponding
+ // AggregationDescriptor
+ List<GroupByLevelDescriptor> descriptorList = new ArrayList<>();
+ for (GroupByLevelDescriptor originalDescriptor : handle.getGroupByLevelDescriptors()) {
+ List<Expression> descriptorExpression = new ArrayList<>();
+ for (String childColumn : childrenOutputColumns) {
+ // If this condition matched, the childColumn should come from GroupByLevelNode
+ if (isAggColumnMatchExpression(childColumn, originalDescriptor.getOutputExpression())) {
+ descriptorExpression.add(originalDescriptor.getOutputExpression());
+ continue;
+ }
+ for (Expression exp : originalDescriptor.getInputExpressions()) {
+ if (isAggColumnMatchExpression(childColumn, exp)) {
+ descriptorExpression.add(exp);
+ }
+ }
+ }
+ if (descriptorExpression.size() == 0) {
+ continue;
+ }
+ GroupByLevelDescriptor descriptor = originalDescriptor.deepClone();
+ descriptor.setStep(level == 0 ? AggregationStep.FINAL : AggregationStep.PARTIAL);
+ descriptor.setInputExpressions(descriptorExpression);
+
+ descriptorList.add(descriptor);
+ }
+ handle.setGroupByLevelDescriptors(descriptorList);
+ }
+
+ // TODO: (xingtanzjr) need to confirm the logic when processing UDF
+ private boolean isAggColumnMatchExpression(String columnName, Expression expression) {
+ if (columnName == null) {
+ return false;
+ }
+ return columnName.contains(expression.getExpressionString());
+ }
+
+ private List<SeriesAggregationSourceNode> splitAggregationSourceByPartition(
+ MultiChildNode root, DistributionPlanContext context) {
+ // Step 1: split SeriesAggregationSourceNode according to data partition
+ List<SeriesAggregationSourceNode> sources = new ArrayList<>();
+ Map<PartialPath, Integer> regionCountPerSeries = new HashMap<>();
+ for (PlanNode child : root.getChildren()) {
+ SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) child;
+ List<TRegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(handle.getPartitionPath(), handle.getPartitionTimeFilter());
+ for (TRegionReplicaSet dataRegion : dataDistribution) {
+ SeriesAggregationSourceNode split = (SeriesAggregationSourceNode) handle.clone();
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ // Let each split reference different object of AggregationDescriptorList
+ split.setAggregationDescriptorList(
+ handle.getAggregationDescriptorList().stream()
+ .map(AggregationDescriptor::deepClone)
+ .collect(Collectors.toList()));
+ sources.add(split);
+ }
+ regionCountPerSeries.put(handle.getPartitionPath(), dataDistribution.size());
+ }
+
+ // Step 2: change the step for each SeriesAggregationSourceNode according to its split count
+ for (SeriesAggregationSourceNode source : sources) {
+ // boolean isFinal = regionCountPerSeries.get(source.getPartitionPath()) == 1;
+ // TODO: (xingtanzjr) need to optimize this step later. We make it as Partial now.
+ boolean isFinal = false;
+ source
+ .getAggregationDescriptorList()
+ .forEach(d -> d.setStep(isFinal ? AggregationStep.FINAL : AggregationStep.PARTIAL));
+ }
+ return sources;
+ }
+
+ public PlanNode visit(PlanNode node, DistributionPlanContext context) {
+ return node.accept(this, context);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/WriteFragmentParallelPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/WriteFragmentParallelPlanner.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/WriteFragmentParallelPlanner.java
rename to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/WriteFragmentParallelPlanner.java
index 6925194e9c..fb91a0c2fe 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/WriteFragmentParallelPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/WriteFragmentParallelPlanner.java
@@ -17,10 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.mpp.plan.planner;
+package org.apache.iotdb.db.mpp.plan.planner.distribution;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.IFragmentParallelPlaner;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.mpp.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
index 626893c152..a500f65d18 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/PlanFragment.java
@@ -21,10 +21,10 @@ package org.apache.iotdb.db.mpp.plan.planner.plan;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.plan.analyze.TypeProvider;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.IPartitionRelatedNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.nio.ByteBuffer;
@@ -77,8 +77,8 @@ public class PlanFragment {
}
private TRegionReplicaSet getNodeRegion(PlanNode root) {
- if (root instanceof SourceNode) {
- return ((SourceNode) root).getRegionReplicaSet();
+ if (root instanceof IPartitionRelatedNode) {
+ return ((IPartitionRelatedNode) root).getRegionReplicaSet();
}
for (PlanNode child : root.getChildren()) {
TRegionReplicaSet result = getNodeRegion(child);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/SubPlan.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/SubPlan.java
index b922c37c8a..7ac6f92d0a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/SubPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/SubPlan.java
@@ -62,7 +62,7 @@ public class SubPlan {
result.add(this.planFragment);
this.children.forEach(
child -> {
- result.add(child.getPlanFragment());
+ result.addAll(child.getPlanFragmentList());
});
return result;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/IPartitionRelatedNode.java
similarity index 73%
copy from server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
copy to server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/IPartitionRelatedNode.java
index 0d6ee0c793..ca82777058 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/IPartitionRelatedNode.java
@@ -20,17 +20,7 @@
package org.apache.iotdb.db.mpp.plan.planner.plan.node;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
-import java.util.List;
-
-public abstract class WritePlanNode extends PlanNode {
-
- protected WritePlanNode(PlanNodeId id) {
- super(id);
- }
-
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
- public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
+public interface IPartitionRelatedNode {
+ TRegionReplicaSet getRegionReplicaSet();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
index d60df01d36..2af623bfbe 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.AlterTimeSe
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+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;
@@ -58,6 +59,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNo
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.LastQueryScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsNode;
@@ -255,4 +257,12 @@ public abstract class PlanVisitor<R, C> {
public R visitLastQueryMerge(LastQueryMergeNode node, C context) {
return visitPlan(node, context);
}
+
+ public R visitDeleteTimeseries(DeleteTimeSeriesNode node, C context) {
+ return visitPlan(node, context);
+ }
+
+ public R visitDeleteData(DeleteDataNode node, C context) {
+ return visitPlan(node, context);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
index 0d6ee0c793..ef877add59 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/WritePlanNode.java
@@ -19,18 +19,15 @@
package org.apache.iotdb.db.mpp.plan.planner.plan.node;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import java.util.List;
-public abstract class WritePlanNode extends PlanNode {
+public abstract class WritePlanNode extends PlanNode implements IPartitionRelatedNode {
protected WritePlanNode(PlanNodeId id) {
super(id);
}
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
public abstract List<WritePlanNode> splitByPartition(Analysis analysis);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
index 98df329b1a..454db8da22 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/DeleteTimeSeriesNode.java
@@ -22,24 +22,28 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.metadata.path.PathDeserializeUtil;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.IPartitionRelatedNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-public class DeleteTimeSeriesNode extends WritePlanNode {
+public class DeleteTimeSeriesNode extends PlanNode implements IPartitionRelatedNode {
private final List<PartialPath> pathList;
+ private TRegionReplicaSet regionReplicaSet;
+ private List<PlanNode> children;
+
public DeleteTimeSeriesNode(PlanNodeId id, List<PartialPath> pathList) {
super(id);
this.pathList = pathList;
+ this.children = new ArrayList<>();
}
public List<PartialPath> getPathList() {
@@ -48,11 +52,13 @@ public class DeleteTimeSeriesNode extends WritePlanNode {
@Override
public List<PlanNode> getChildren() {
- return null;
+ return children;
}
@Override
- public void addChild(PlanNode child) {}
+ public void addChild(PlanNode child) {
+ children.add(child);
+ }
@Override
public PlanNode clone() {
@@ -61,7 +67,7 @@ public class DeleteTimeSeriesNode extends WritePlanNode {
@Override
public int allowedChildCount() {
- return 0;
+ return CHILD_COUNT_NO_LIMIT;
}
@Override
@@ -69,6 +75,11 @@ public class DeleteTimeSeriesNode extends WritePlanNode {
return null;
}
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitDeleteTimeseries(this, context);
+ }
+
@Override
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.DELETE_TIMESERIES.serialize(byteBuffer);
@@ -90,11 +101,18 @@ public class DeleteTimeSeriesNode extends WritePlanNode {
@Override
public TRegionReplicaSet getRegionReplicaSet() {
- return null;
+ return regionReplicaSet;
}
- @Override
- public List<WritePlanNode> splitByPartition(Analysis analysis) {
- return null;
+ public void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet) {
+ this.regionReplicaSet = regionReplicaSet;
+ }
+
+ public String toString() {
+ return String.format(
+ "DeleteTimeseriesNode-%s: %s. Region: %s",
+ getPlanNodeId(),
+ pathList,
+ regionReplicaSet == null ? "Not Assigned" : regionReplicaSet.getRegionId());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
index 95649d8248..8af09b8a3d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/metedata/write/InvalidateSchemaCacheNode.java
@@ -19,22 +19,19 @@
package org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.write;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.metadata.path.PathDeserializeUtil;
import org.apache.iotdb.db.mpp.common.QueryId;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-public class InvalidateSchemaCacheNode extends WritePlanNode {
+public class InvalidateSchemaCacheNode extends PlanNode {
private final QueryId queryId;
@@ -114,14 +111,4 @@ public class InvalidateSchemaCacheNode extends WritePlanNode {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new InvalidateSchemaCacheNode(planNodeId, queryId, pathList, storageGroups);
}
-
- @Override
- public TRegionReplicaSet getRegionReplicaSet() {
- return null;
- }
-
- @Override
- public List<WritePlanNode> splitByPartition(Analysis analysis) {
- return null;
- }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java
index e3209a03e4..63203cf07f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/SourceNode.java
@@ -19,10 +19,11 @@
package org.apache.iotdb.db.mpp.plan.planner.plan.node.source;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.IPartitionRelatedNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
-public abstract class SourceNode extends PlanNode implements AutoCloseable {
+public abstract class SourceNode extends PlanNode implements AutoCloseable, IPartitionRelatedNode {
public SourceNode(PlanNodeId id) {
super(id);
@@ -30,7 +31,5 @@ public abstract class SourceNode extends PlanNode implements AutoCloseable {
public abstract void open() throws Exception;
- public abstract TRegionReplicaSet getRegionReplicaSet();
-
public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
index 21ebfec88d..a594bba0c9 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/DeleteDataNode.java
@@ -22,24 +22,32 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.node.write;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.metadata.path.PathDeserializeUtil;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.IPartitionRelatedNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-public class DeleteDataNode extends WritePlanNode {
+public class DeleteDataNode extends PlanNode implements IPartitionRelatedNode {
+ private final QueryId queryId;
private final List<PartialPath> pathList;
+ private final List<String> storageGroups;
- public DeleteDataNode(PlanNodeId id, List<PartialPath> pathList) {
+ private TRegionReplicaSet regionReplicaSet;
+
+ public DeleteDataNode(
+ PlanNodeId id, QueryId queryId, List<PartialPath> pathList, List<String> storageGroups) {
super(id);
this.pathList = pathList;
+ this.queryId = queryId;
+ this.storageGroups = storageGroups;
}
public List<PartialPath> getPathList() {
@@ -48,7 +56,7 @@ public class DeleteDataNode extends WritePlanNode {
@Override
public List<PlanNode> getChildren() {
- return null;
+ return new ArrayList<>();
}
@Override
@@ -56,12 +64,12 @@ public class DeleteDataNode extends WritePlanNode {
@Override
public PlanNode clone() {
- return new DeleteDataNode(getPlanNodeId(), pathList);
+ return new DeleteDataNode(getPlanNodeId(), queryId, pathList, storageGroups);
}
@Override
public int allowedChildCount() {
- return 0;
+ return NO_CHILD_ALLOWED;
}
@Override
@@ -72,29 +80,61 @@ public class DeleteDataNode extends WritePlanNode {
@Override
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.DELETE_DATA.serialize(byteBuffer);
+ queryId.serialize(byteBuffer);
ReadWriteIOUtils.write(pathList.size(), byteBuffer);
for (PartialPath path : pathList) {
path.serialize(byteBuffer);
}
+ ReadWriteIOUtils.write(storageGroups.size(), byteBuffer);
+ for (String storageGroup : storageGroups) {
+ ReadWriteIOUtils.write(storageGroup, byteBuffer);
+ }
}
- public static DeleteDataNode deserialize(ByteBuffer buffer) {
- int size = ReadWriteIOUtils.readInt(buffer);
+ public static DeleteDataNode deserialize(ByteBuffer byteBuffer) {
+ QueryId queryId = QueryId.deserialize(byteBuffer);
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
List<PartialPath> pathList = new ArrayList<>(size);
for (int i = 0; i < size; i++) {
- pathList.add((PartialPath) PathDeserializeUtil.deserialize(buffer));
+ pathList.add((PartialPath) PathDeserializeUtil.deserialize(byteBuffer));
+ }
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<String> storageGroups = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ storageGroups.add(ReadWriteIOUtils.readString(byteBuffer));
}
- PlanNodeId planNodeId = PlanNodeId.deserialize(buffer);
- return new DeleteDataNode(planNodeId, pathList);
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new DeleteDataNode(planNodeId, queryId, pathList, storageGroups);
}
@Override
- public TRegionReplicaSet getRegionReplicaSet() {
- return null;
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitDeleteData(this, context);
}
@Override
- public List<WritePlanNode> splitByPartition(Analysis analysis) {
- return null;
+ public TRegionReplicaSet getRegionReplicaSet() {
+ return regionReplicaSet;
+ }
+
+ public void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet) {
+ this.regionReplicaSet = regionReplicaSet;
+ }
+
+ public QueryId getQueryId() {
+ return queryId;
+ }
+
+ public List<String> getStorageGroups() {
+ return storageGroups;
+ }
+
+ public String toString() {
+ return String.format(
+ "DeleteDataNode-%s[ Paths: %s, StorageGroups: %s, Region: %s ]",
+ getPlanNodeId(),
+ pathList,
+ storageGroups,
+ regionReplicaSet == null ? "Not Assigned" : regionReplicaSet.getRegionId());
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DeleteTimeSeriesStatement.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DeleteTimeSeriesStatement.java
index edcccd58e0..99cb5b57ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DeleteTimeSeriesStatement.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/statement/metadata/DeleteTimeSeriesStatement.java
@@ -31,7 +31,7 @@ public class DeleteTimeSeriesStatement extends Statement {
@Override
public List<PartialPath> getPaths() {
- return null;
+ return partialPaths;
}
public void setPartialPaths(List<PartialPath> partialPaths) {
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/DistributionPlannerTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/DistributionPlannerTest.java
deleted file mode 100644
index 5882cfb66b..0000000000
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/DistributionPlannerTest.java
+++ /dev/null
@@ -1,934 +0,0 @@
-/*
- * 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;
-
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
-import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
-import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
-import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.commons.exception.IllegalPathException;
-import org.apache.iotdb.commons.partition.DataPartition;
-import org.apache.iotdb.commons.partition.SchemaPartition;
-import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
-import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.metadata.path.AlignedPath;
-import org.apache.iotdb.db.metadata.path.MeasurementPath;
-import org.apache.iotdb.db.mpp.common.MPPQueryContext;
-import org.apache.iotdb.db.mpp.common.QueryId;
-import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
-import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
-import org.apache.iotdb.db.mpp.plan.expression.Expression;
-import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
-import org.apache.iotdb.db.mpp.plan.planner.DistributionPlanner;
-import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
-import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
-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.LimitNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationSourceNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationDescriptor;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.AggregationStep;
-import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByLevelDescriptor;
-import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
-import org.apache.iotdb.db.query.aggregation.AggregationType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-public class DistributionPlannerTest {
-
- @Test
- public void testSingleSeriesScan() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- SeriesScanNode root =
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC);
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(2, plan.getInstances().size());
- }
-
- @Test
- public void testSingleSeriesScanRewriteSource() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- SeriesScanNode root =
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC);
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- PlanNode rootAfterRewrite = planner.rewriteSource();
- assertEquals(2, rootAfterRewrite.getChildren().size());
- }
-
- @Test
- public void testRewriteSourceNode() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
-
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
-
- Analysis analysis = constructAnalysis();
-
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root));
- PlanNode newRoot = planner.rewriteSource();
- assertEquals(4, newRoot.getChildren().get(0).getChildren().size());
- }
-
- @Test
- public void testRewriteMetaSourceNode() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- SchemaQueryMergeNode metaMergeNode = new SchemaQueryMergeNode(queryId.genPlanNodeId(), false);
- metaMergeNode.addChild(
- new TimeSeriesSchemaScanNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d1.s1"),
- null,
- null,
- 10,
- 0,
- false,
- false,
- false));
- metaMergeNode.addChild(
- new TimeSeriesSchemaScanNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d1.s2"),
- null,
- null,
- 10,
- 0,
- false,
- false,
- false));
- metaMergeNode.addChild(
- new TimeSeriesSchemaScanNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d22.s1"),
- null,
- null,
- 10,
- 0,
- false,
- false,
- false));
- LimitNode root2 = new LimitNode(queryId.genPlanNodeId(), metaMergeNode, 10);
- Analysis analysis = constructAnalysis();
- DistributionPlanner planner2 =
- new DistributionPlanner(
- analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root2));
- PlanNode newRoot2 = planner2.rewriteSource();
- assertEquals(newRoot2.getChildren().get(0).getChildren().size(), 2);
- }
-
- @Test
- public void testAddExchangeNode() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
-
- Analysis analysis = constructAnalysis();
-
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root));
- PlanNode rootAfterRewrite = planner.rewriteSource();
- PlanNode rootWithExchange = planner.addExchangeNode(rootAfterRewrite);
- assertEquals(4, rootWithExchange.getChildren().get(0).getChildren().size());
- int exchangeNodeCount = 0;
- for (PlanNode child : rootWithExchange.getChildren().get(0).getChildren()) {
- exchangeNodeCount += child instanceof ExchangeNode ? 1 : 0;
- }
- assertEquals(2, exchangeNodeCount);
- }
-
- @Test
- public void testSplitFragment() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- PlanNode rootAfterRewrite = planner.rewriteSource();
- PlanNode rootWithExchange = planner.addExchangeNode(rootAfterRewrite);
- SubPlan subPlan = planner.splitFragment(rootWithExchange);
- assertEquals(subPlan.getChildren().size(), 2);
- }
-
- @Test
- public void testParallelPlan() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d333.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- }
-
- @Test
- public void testTimeJoinAggregationSinglePerRegion() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query_time_join_aggregation");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
- String d1s1Path = "root.sg.d1.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT));
-
- String d2s1Path = "root.sg.d22.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT));
-
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
- }
-
- private void verifyAggregationStep(Map<String, AggregationStep> expected, PlanNode root) {
- if (root == null) {
- return;
- }
- if (root instanceof SeriesAggregationSourceNode) {
- SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) root;
- List<AggregationDescriptor> descriptorList = handle.getAggregationDescriptorList();
- descriptorList.forEach(
- d -> {
- assertEquals(expected.get(handle.getPartitionPath().getFullPath()), d.getStep());
- });
- }
- root.getChildren().forEach(child -> verifyAggregationStep(expected, child));
- }
-
- @Test
- public void testTimeJoinAggregationMultiPerRegion() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query_time_join_aggregation");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
- String d1s1Path = "root.sg.d1.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT));
-
- String d3s1Path = "root.sg.d333.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT));
-
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
- }
-
- @Test
- public void testTimeJoinAggregationMultiPerRegion2() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query_time_join_aggregation");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
- String d3s1Path = "root.sg.d333.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT));
-
- String d4s1Path = "root.sg.d4444.s1";
- timeJoinNode.addChild(genAggregationSourceNode(queryId, d4s1Path, AggregationType.COUNT));
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(2, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d4s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
- }
-
- @Test
- public void testGroupByLevelWithTwoChildren() throws IllegalPathException {
- QueryId queryId = new QueryId("test_group_by_level_two_children");
- String d1s1Path = "root.sg.d1.s1";
- String d2s1Path = "root.sg.d22.s1";
- String groupedPath = "root.sg.*.s1";
-
- GroupByLevelNode groupByLevelNode =
- new GroupByLevelNode(
- new PlanNodeId("TestGroupByLevelNode"),
- Arrays.asList(
- genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
- genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT)),
- Collections.singletonList(
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Arrays.asList(
- new TimeSeriesOperand(new PartialPath(d1s1Path)),
- new TimeSeriesOperand(new PartialPath(d2s1Path))),
- new TimeSeriesOperand(new PartialPath(groupedPath)))));
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
- }
-
- @Test
- public void testAggregationWithMultiGroupByLevelNode() throws IllegalPathException {
- QueryId queryId = new QueryId("test_group_by_level_two_children");
- String d3s1Path = "root.sg.d333.s1";
- String d4s1Path = "root.sg.d4444.s1";
- String groupedPath = "root.sg.*.s1";
-
- GroupByLevelNode groupByLevelNode =
- new GroupByLevelNode(
- new PlanNodeId("TestGroupByLevelNode"),
- Arrays.asList(
- genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT),
- genAggregationSourceNode(queryId, d4s1Path, AggregationType.COUNT)),
- Collections.singletonList(
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Arrays.asList(
- new TimeSeriesOperand(new PartialPath(d3s1Path)),
- new TimeSeriesOperand(new PartialPath(d4s1Path))),
- new TimeSeriesOperand(new PartialPath(groupedPath)))));
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(2, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d4s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
-
- Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
- expectedDescriptorValue.put(groupedPath, Arrays.asList(groupedPath, d3s1Path, d4s1Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue,
- (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
-
- Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
- expectedDescriptorValue2.put(groupedPath, Arrays.asList(d3s1Path, d4s1Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue2,
- (GroupByLevelNode) fragmentInstances.get(1).getFragment().getRoot().getChildren().get(0));
- }
-
- @Test
- public void testGroupByLevelTwoSeries() throws IllegalPathException {
- QueryId queryId = new QueryId("test_group_by_level_two_series");
- String d1s1Path = "root.sg.d1.s1";
- String d1s2Path = "root.sg.d1.s2";
- String groupedPathS1 = "root.sg.*.s1";
- String groupedPathS2 = "root.sg.*.s2";
-
- GroupByLevelNode groupByLevelNode =
- new GroupByLevelNode(
- new PlanNodeId("TestGroupByLevelNode"),
- Arrays.asList(
- genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
- genAggregationSourceNode(queryId, d1s2Path, AggregationType.COUNT)),
- Arrays.asList(
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s1Path))),
- new TimeSeriesOperand(new PartialPath(groupedPathS1))),
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s2Path))),
- new TimeSeriesOperand(new PartialPath(groupedPathS2)))));
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(2, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d1s2Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
-
- Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
- expectedDescriptorValue.put(groupedPathS1, Arrays.asList(groupedPathS1, d1s1Path));
- expectedDescriptorValue.put(groupedPathS2, Arrays.asList(groupedPathS2, d1s2Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue,
- (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
-
- Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
- expectedDescriptorValue2.put(groupedPathS1, Collections.singletonList(d1s1Path));
- expectedDescriptorValue2.put(groupedPathS2, Collections.singletonList(d1s2Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue2,
- (GroupByLevelNode) fragmentInstances.get(1).getFragment().getRoot().getChildren().get(0));
- }
-
- @Test
- public void testGroupByLevel2Series2Devices3Regions() throws IllegalPathException {
- QueryId queryId = new QueryId("test_group_by_level_two_series");
- String d1s1Path = "root.sg.d1.s1";
- String d1s2Path = "root.sg.d1.s2";
- String d2s1Path = "root.sg.d22.s1";
- String groupedPathS1 = "root.sg.*.s1";
- String groupedPathS2 = "root.sg.*.s2";
-
- GroupByLevelNode groupByLevelNode =
- new GroupByLevelNode(
- new PlanNodeId("TestGroupByLevelNode"),
- Arrays.asList(
- genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
- genAggregationSourceNode(queryId, d1s2Path, AggregationType.COUNT),
- genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT)),
- Arrays.asList(
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Arrays.asList(
- new TimeSeriesOperand(new PartialPath(d1s1Path)),
- new TimeSeriesOperand(new PartialPath(d2s1Path))),
- new TimeSeriesOperand(new PartialPath(groupedPathS1))),
- new GroupByLevelDescriptor(
- AggregationType.COUNT,
- AggregationStep.FINAL,
- Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s2Path))),
- new TimeSeriesOperand(new PartialPath(groupedPathS2)))));
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- Map<String, AggregationStep> expectedStep = new HashMap<>();
- expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
- expectedStep.put(d1s2Path, AggregationStep.PARTIAL);
- expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
- List<FragmentInstance> fragmentInstances = plan.getInstances();
- fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
-
- Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
- expectedDescriptorValue.put(groupedPathS1, Arrays.asList(groupedPathS1, d1s1Path, d2s1Path));
- expectedDescriptorValue.put(groupedPathS2, Arrays.asList(groupedPathS2, d1s2Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue,
- (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
-
- Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
- expectedDescriptorValue2.put(groupedPathS1, Collections.singletonList(d1s1Path));
- expectedDescriptorValue2.put(groupedPathS2, Collections.singletonList(d1s2Path));
- verifyGroupByLevelDescriptor(
- expectedDescriptorValue2,
- (GroupByLevelNode) fragmentInstances.get(2).getFragment().getRoot().getChildren().get(0));
- }
-
- @Test
- public void testAggregation1Series1Region() throws IllegalPathException {
- QueryId queryId = new QueryId("test_aggregation_1_series_1_region");
- String d2s1Path = "root.sg.d22.s1";
-
- PlanNode root = genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT);
- Analysis analysis = constructAnalysis();
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(1, plan.getInstances().size());
- assertEquals(root, plan.getInstances().get(0).getFragment().getRoot().getChildren().get(0));
- }
-
- private void verifyGroupByLevelDescriptor(
- Map<String, List<String>> expected, GroupByLevelNode node) {
- List<GroupByLevelDescriptor> descriptors = node.getGroupByLevelDescriptors();
- assertEquals(expected.size(), descriptors.size());
- for (GroupByLevelDescriptor descriptor : descriptors) {
- String outputExpression = descriptor.getOutputExpression().getExpressionString();
- assertEquals(expected.get(outputExpression).size(), descriptor.getInputExpressions().size());
- for (Expression inputExpression : descriptor.getInputExpressions()) {
- assertTrue(expected.get(outputExpression).contains(inputExpression.getExpressionString()));
- }
- }
- }
-
- private SeriesAggregationSourceNode genAggregationSourceNode(
- QueryId queryId, String path, AggregationType type) throws IllegalPathException {
- List<AggregationDescriptor> descriptors = new ArrayList<>();
- descriptors.add(
- new AggregationDescriptor(
- type,
- AggregationStep.FINAL,
- Collections.singletonList(new TimeSeriesOperand(new PartialPath(path)))));
-
- return new SeriesAggregationScanNode(
- queryId.genPlanNodeId(), new MeasurementPath(path, TSDataType.INT32), descriptors);
- }
-
- @Test
- public void testParallelPlanWithAlignedSeries() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query_aligned");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new AlignedSeriesScanNode(
- queryId.genPlanNodeId(),
- new AlignedPath("root.sg.d1", Arrays.asList("s1", "s2")),
- OrderBy.TIMESTAMP_ASC));
- timeJoinNode.addChild(
- new SeriesScanNode(
- queryId.genPlanNodeId(),
- new MeasurementPath("root.sg.d333.s1", TSDataType.INT32),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(3, plan.getInstances().size());
- }
-
- @Test
- public void testSingleAlignedSeries() throws IllegalPathException {
- QueryId queryId = new QueryId("test_query_aligned");
- TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
-
- timeJoinNode.addChild(
- new AlignedSeriesScanNode(
- queryId.genPlanNodeId(),
- new AlignedPath("root.sg.d22", Arrays.asList("s1", "s2")),
- OrderBy.TIMESTAMP_ASC));
-
- LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(1, plan.getInstances().size());
- }
-
- @Test
- public void testInsertRowNodeParallelPlan() throws IllegalPathException {
- QueryId queryId = new QueryId("test_write");
- InsertRowNode insertRowNode =
- new InsertRowNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d1"),
- false,
- new String[] {
- "s1",
- },
- new TSDataType[] {TSDataType.INT32},
- 1L,
- new Object[] {10},
- false);
- insertRowNode.setMeasurementSchemas(
- new MeasurementSchema[] {
- new MeasurementSchema("s1", TSDataType.INT32),
- });
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- context.setQueryType(QueryType.WRITE);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, insertRowNode));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(1, plan.getInstances().size());
- }
-
- @Test
- public void testInsertRowsNodeParallelPlan() throws IllegalPathException {
- QueryId queryId = new QueryId("test_write");
- InsertRowNode insertRowNode1 =
- new InsertRowNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d1"),
- false,
- new String[] {"s1"},
- new TSDataType[] {TSDataType.INT32},
- 1L,
- new Object[] {10},
- false);
- insertRowNode1.setMeasurementSchemas(
- new MeasurementSchema[] {
- new MeasurementSchema("s1", TSDataType.INT32),
- });
-
- InsertRowNode insertRowNode2 =
- new InsertRowNode(
- queryId.genPlanNodeId(),
- new PartialPath("root.sg.d1"),
- false,
- new String[] {"s1"},
- new TSDataType[] {TSDataType.INT32},
- 100000L,
- new Object[] {10},
- false);
- insertRowNode2.setMeasurementSchemas(
- new MeasurementSchema[] {
- new MeasurementSchema("s1", TSDataType.INT32),
- });
-
- InsertRowsNode node = new InsertRowsNode(queryId.genPlanNodeId());
- node.setInsertRowNodeList(Arrays.asList(insertRowNode1, insertRowNode2));
- node.setInsertRowNodeIndexList(Arrays.asList(0, 1));
-
- Analysis analysis = constructAnalysis();
-
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- context.setQueryType(QueryType.WRITE);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, node));
- DistributedQueryPlan plan = planner.planFragments();
- assertEquals(1, plan.getInstances().size());
- }
-
- private Analysis constructAnalysis() throws IllegalPathException {
-
- SeriesPartitionExecutor executor =
- SeriesPartitionExecutor.getSeriesPartitionExecutor(
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
- Analysis analysis = new Analysis();
-
- String device1 = "root.sg.d1";
- String device2 = "root.sg.d22";
- String device3 = "root.sg.d333";
- String device4 = "root.sg.d4444";
-
- DataPartition dataPartition =
- new DataPartition(
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
-
- Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
- dataPartitionMap = new HashMap<>();
- Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> sgPartitionMap =
- new HashMap<>();
-
- List<TRegionReplicaSet> d1DataRegions = new ArrayList<>();
- d1DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(11)
- .setExternalEndPoint(new TEndPoint("192.0.1.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(12)
- .setExternalEndPoint(new TEndPoint("192.0.1.2", 9000)))));
- d1DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 2),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(21)
- .setExternalEndPoint(new TEndPoint("192.0.2.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(22)
- .setExternalEndPoint(new TEndPoint("192.0.2.2", 9000)))));
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d1DataRegionMap = new HashMap<>();
- d1DataRegionMap.put(new TTimePartitionSlot(), d1DataRegions);
-
- List<TRegionReplicaSet> d2DataRegions = new ArrayList<>();
- d2DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 3),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(31)
- .setExternalEndPoint(new TEndPoint("192.0.3.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(32)
- .setExternalEndPoint(new TEndPoint("192.0.3.2", 9000)))));
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d2DataRegionMap = new HashMap<>();
- d2DataRegionMap.put(new TTimePartitionSlot(), d2DataRegions);
-
- List<TRegionReplicaSet> d3DataRegions = new ArrayList<>();
- d3DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(11)
- .setExternalEndPoint(new TEndPoint("192.0.1.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(12)
- .setExternalEndPoint(new TEndPoint("192.0.1.2", 9000)))));
- d3DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 4),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(41)
- .setExternalEndPoint(new TEndPoint("192.0.4.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(42)
- .setExternalEndPoint(new TEndPoint("192.0.4.2", 9000)))));
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d3DataRegionMap = new HashMap<>();
- d3DataRegionMap.put(new TTimePartitionSlot(), d3DataRegions);
-
- List<TRegionReplicaSet> d4DataRegions = new ArrayList<>();
- d4DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(11)
- .setExternalEndPoint(new TEndPoint("192.0.1.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(12)
- .setExternalEndPoint(new TEndPoint("192.0.1.2", 9000)))));
- d4DataRegions.add(
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 4),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(41)
- .setExternalEndPoint(new TEndPoint("192.0.4.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(42)
- .setExternalEndPoint(new TEndPoint("192.0.4.2", 9000)))));
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d4DataRegionMap = new HashMap<>();
- d4DataRegionMap.put(new TTimePartitionSlot(), d4DataRegions);
-
- sgPartitionMap.put(executor.getSeriesPartitionSlot(device1), d1DataRegionMap);
- sgPartitionMap.put(executor.getSeriesPartitionSlot(device2), d2DataRegionMap);
- sgPartitionMap.put(executor.getSeriesPartitionSlot(device3), d3DataRegionMap);
- sgPartitionMap.put(executor.getSeriesPartitionSlot(device4), d4DataRegionMap);
-
- dataPartitionMap.put("root.sg", sgPartitionMap);
-
- dataPartition.setDataPartitionMap(dataPartitionMap);
-
- analysis.setDataPartitionInfo(dataPartition);
-
- // construct AggregationExpression for GroupByLevel
- Map<String, Set<Expression>> aggregationExpression = new HashMap<>();
- Set<Expression> s1Expression = new HashSet<>();
- s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1")));
- s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d22.s1")));
- s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d333.s1")));
- s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d4444.s1")));
-
- Set<Expression> s2Expression = new HashSet<>();
- s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d1.s2")));
- s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d22.s2")));
- s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d333.s2")));
- s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d4444.s2")));
-
- aggregationExpression.put("root.sg.*.s1", s1Expression);
- aggregationExpression.put("root.sg.*.s2", s2Expression);
- // analysis.setAggregationExpressions(aggregationExpression);
-
- // construct schema partition
- SchemaPartition schemaPartition =
- new SchemaPartition(
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
- Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> schemaPartitionMap = new HashMap<>();
-
- TRegionReplicaSet schemaRegion1 =
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 11),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(11)
- .setExternalEndPoint(new TEndPoint("192.0.1.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(12)
- .setExternalEndPoint(new TEndPoint("192.0.1.2", 9000))));
- Map<TSeriesPartitionSlot, TRegionReplicaSet> schemaRegionMap = new HashMap<>();
-
- TRegionReplicaSet schemaRegion2 =
- new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 21),
- Arrays.asList(
- new TDataNodeLocation()
- .setDataNodeId(21)
- .setExternalEndPoint(new TEndPoint("192.0.1.1", 9000)),
- new TDataNodeLocation()
- .setDataNodeId(22)
- .setExternalEndPoint(new TEndPoint("192.0.1.2", 9000))));
-
- schemaRegionMap.put(executor.getSeriesPartitionSlot(device1), schemaRegion1);
- schemaRegionMap.put(executor.getSeriesPartitionSlot(device2), schemaRegion2);
- schemaRegionMap.put(executor.getSeriesPartitionSlot(device3), schemaRegion2);
- schemaPartitionMap.put("root.sg", schemaRegionMap);
- schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
-
- analysis.setDataPartitionInfo(dataPartition);
- analysis.setSchemaPartitionInfo(schemaPartition);
- return analysis;
- }
-}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AggregationDistributionTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AggregationDistributionTest.java
new file mode 100644
index 0000000000..befbeb9660
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/AggregationDistributionTest.java
@@ -0,0 +1,412 @@
+/*
+ * 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.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.metadata.path.AlignedPath;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
+import org.apache.iotdb.db.mpp.plan.planner.distribution.DistributionPlanner;
+import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
+import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.GroupByLevelNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationSourceNode;
+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.AggregationStep;
+import org.apache.iotdb.db.mpp.plan.planner.plan.parameter.GroupByLevelDescriptor;
+import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
+import org.apache.iotdb.db.query.aggregation.AggregationType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AggregationDistributionTest {
+ @Test
+ public void testTimeJoinAggregationSinglePerRegion() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query_time_join_aggregation");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+ String d1s1Path = "root.sg.d1.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT));
+
+ String d2s1Path = "root.sg.d22.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT));
+
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+ }
+
+ private void verifyAggregationStep(Map<String, AggregationStep> expected, PlanNode root) {
+ if (root == null) {
+ return;
+ }
+ if (root instanceof SeriesAggregationSourceNode) {
+ SeriesAggregationSourceNode handle = (SeriesAggregationSourceNode) root;
+ List<AggregationDescriptor> descriptorList = handle.getAggregationDescriptorList();
+ descriptorList.forEach(
+ d -> {
+ assertEquals(expected.get(handle.getPartitionPath().getFullPath()), d.getStep());
+ });
+ }
+ root.getChildren().forEach(child -> verifyAggregationStep(expected, child));
+ }
+
+ @Test
+ public void testTimeJoinAggregationMultiPerRegion() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query_time_join_aggregation");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+ String d1s1Path = "root.sg.d1.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT));
+
+ String d3s1Path = "root.sg.d333.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT));
+
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+ }
+
+ @Test
+ public void testTimeJoinAggregationMultiPerRegion2() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query_time_join_aggregation");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+ String d3s1Path = "root.sg.d333.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT));
+
+ String d4s1Path = "root.sg.d4444.s1";
+ timeJoinNode.addChild(genAggregationSourceNode(queryId, d4s1Path, AggregationType.COUNT));
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, timeJoinNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(2, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d4s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+ }
+
+ @Test
+ public void testGroupByLevelWithTwoChildren() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_group_by_level_two_children");
+ String d1s1Path = "root.sg.d1.s1";
+ String d2s1Path = "root.sg.d22.s1";
+ String groupedPath = "root.sg.*.s1";
+
+ GroupByLevelNode groupByLevelNode =
+ new GroupByLevelNode(
+ new PlanNodeId("TestGroupByLevelNode"),
+ Arrays.asList(
+ genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
+ genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT)),
+ Collections.singletonList(
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Arrays.asList(
+ new TimeSeriesOperand(new PartialPath(d1s1Path)),
+ new TimeSeriesOperand(new PartialPath(d2s1Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPath)))));
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+ }
+
+ @Test
+ public void testAggregationWithMultiGroupByLevelNode() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_group_by_level_two_children");
+ String d3s1Path = "root.sg.d333.s1";
+ String d4s1Path = "root.sg.d4444.s1";
+ String groupedPath = "root.sg.*.s1";
+
+ GroupByLevelNode groupByLevelNode =
+ new GroupByLevelNode(
+ new PlanNodeId("TestGroupByLevelNode"),
+ Arrays.asList(
+ genAggregationSourceNode(queryId, d3s1Path, AggregationType.COUNT),
+ genAggregationSourceNode(queryId, d4s1Path, AggregationType.COUNT)),
+ Collections.singletonList(
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Arrays.asList(
+ new TimeSeriesOperand(new PartialPath(d3s1Path)),
+ new TimeSeriesOperand(new PartialPath(d4s1Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPath)))));
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(2, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d3s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d4s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+
+ Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
+ expectedDescriptorValue.put(groupedPath, Arrays.asList(groupedPath, d3s1Path, d4s1Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue,
+ (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
+
+ Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
+ expectedDescriptorValue2.put(groupedPath, Arrays.asList(d3s1Path, d4s1Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue2,
+ (GroupByLevelNode) fragmentInstances.get(1).getFragment().getRoot().getChildren().get(0));
+ }
+
+ @Test
+ public void testGroupByLevelTwoSeries() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_group_by_level_two_series");
+ String d1s1Path = "root.sg.d1.s1";
+ String d1s2Path = "root.sg.d1.s2";
+ String groupedPathS1 = "root.sg.*.s1";
+ String groupedPathS2 = "root.sg.*.s2";
+
+ GroupByLevelNode groupByLevelNode =
+ new GroupByLevelNode(
+ new PlanNodeId("TestGroupByLevelNode"),
+ Arrays.asList(
+ genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
+ genAggregationSourceNode(queryId, d1s2Path, AggregationType.COUNT)),
+ Arrays.asList(
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s1Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPathS1))),
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s2Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPathS2)))));
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(2, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d1s2Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+
+ Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
+ expectedDescriptorValue.put(groupedPathS1, Arrays.asList(groupedPathS1, d1s1Path));
+ expectedDescriptorValue.put(groupedPathS2, Arrays.asList(groupedPathS2, d1s2Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue,
+ (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
+
+ Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
+ expectedDescriptorValue2.put(groupedPathS1, Collections.singletonList(d1s1Path));
+ expectedDescriptorValue2.put(groupedPathS2, Collections.singletonList(d1s2Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue2,
+ (GroupByLevelNode) fragmentInstances.get(1).getFragment().getRoot().getChildren().get(0));
+ }
+
+ @Test
+ public void testGroupByLevel2Series2Devices3Regions() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_group_by_level_two_series");
+ String d1s1Path = "root.sg.d1.s1";
+ String d1s2Path = "root.sg.d1.s2";
+ String d2s1Path = "root.sg.d22.s1";
+ String groupedPathS1 = "root.sg.*.s1";
+ String groupedPathS2 = "root.sg.*.s2";
+
+ GroupByLevelNode groupByLevelNode =
+ new GroupByLevelNode(
+ new PlanNodeId("TestGroupByLevelNode"),
+ Arrays.asList(
+ genAggregationSourceNode(queryId, d1s1Path, AggregationType.COUNT),
+ genAggregationSourceNode(queryId, d1s2Path, AggregationType.COUNT),
+ genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT)),
+ Arrays.asList(
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Arrays.asList(
+ new TimeSeriesOperand(new PartialPath(d1s1Path)),
+ new TimeSeriesOperand(new PartialPath(d2s1Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPathS1))),
+ new GroupByLevelDescriptor(
+ AggregationType.COUNT,
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath(d1s2Path))),
+ new TimeSeriesOperand(new PartialPath(groupedPathS2)))));
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, groupByLevelNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ Map<String, AggregationStep> expectedStep = new HashMap<>();
+ expectedStep.put(d1s1Path, AggregationStep.PARTIAL);
+ expectedStep.put(d1s2Path, AggregationStep.PARTIAL);
+ expectedStep.put(d2s1Path, AggregationStep.PARTIAL);
+ List<FragmentInstance> fragmentInstances = plan.getInstances();
+ fragmentInstances.forEach(f -> verifyAggregationStep(expectedStep, f.getFragment().getRoot()));
+
+ Map<String, List<String>> expectedDescriptorValue = new HashMap<>();
+ expectedDescriptorValue.put(groupedPathS1, Arrays.asList(groupedPathS1, d1s1Path, d2s1Path));
+ expectedDescriptorValue.put(groupedPathS2, Arrays.asList(groupedPathS2, d1s2Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue,
+ (GroupByLevelNode) fragmentInstances.get(0).getFragment().getRoot().getChildren().get(0));
+
+ Map<String, List<String>> expectedDescriptorValue2 = new HashMap<>();
+ expectedDescriptorValue2.put(groupedPathS1, Collections.singletonList(d1s1Path));
+ expectedDescriptorValue2.put(groupedPathS2, Collections.singletonList(d1s2Path));
+ verifyGroupByLevelDescriptor(
+ expectedDescriptorValue2,
+ (GroupByLevelNode) fragmentInstances.get(2).getFragment().getRoot().getChildren().get(0));
+ }
+
+ @Test
+ public void testAggregation1Series1Region() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_aggregation_1_series_1_region");
+ String d2s1Path = "root.sg.d22.s1";
+
+ PlanNode root = genAggregationSourceNode(queryId, d2s1Path, AggregationType.COUNT);
+ Analysis analysis = Util.constructAnalysis();
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(1, plan.getInstances().size());
+ assertEquals(root, plan.getInstances().get(0).getFragment().getRoot().getChildren().get(0));
+ }
+
+ private void verifyGroupByLevelDescriptor(
+ Map<String, List<String>> expected, GroupByLevelNode node) {
+ List<GroupByLevelDescriptor> descriptors = node.getGroupByLevelDescriptors();
+ assertEquals(expected.size(), descriptors.size());
+ for (GroupByLevelDescriptor descriptor : descriptors) {
+ String outputExpression = descriptor.getOutputExpression().getExpressionString();
+ assertEquals(expected.get(outputExpression).size(), descriptor.getInputExpressions().size());
+ for (Expression inputExpression : descriptor.getInputExpressions()) {
+ assertTrue(expected.get(outputExpression).contains(inputExpression.getExpressionString()));
+ }
+ }
+ }
+
+ private SeriesAggregationSourceNode genAggregationSourceNode(
+ QueryId queryId, String path, AggregationType type) throws IllegalPathException {
+ List<AggregationDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(
+ new AggregationDescriptor(
+ type,
+ AggregationStep.FINAL,
+ Collections.singletonList(new TimeSeriesOperand(new PartialPath(path)))));
+
+ return new SeriesAggregationScanNode(
+ queryId.genPlanNodeId(), new MeasurementPath(path, TSDataType.INT32), descriptors);
+ }
+
+ @Test
+ public void testParallelPlanWithAlignedSeries() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query_aligned");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new AlignedSeriesScanNode(
+ queryId.genPlanNodeId(),
+ new AlignedPath("root.sg.d1", Arrays.asList("s1", "s2")),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d333.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/BasicTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/BasicTest.java
new file mode 100644
index 0000000000..f6133ac41c
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/BasicTest.java
@@ -0,0 +1,378 @@
+/*
+ * 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.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.metadata.path.AlignedPath;
+import org.apache.iotdb.db.metadata.path.MeasurementPath;
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
+import org.apache.iotdb.db.mpp.plan.planner.distribution.DistributionPlanner;
+import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.SubPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.metedata.read.TimeSeriesSchemaScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TimeJoinNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowsNode;
+import org.apache.iotdb.db.mpp.plan.statement.component.OrderBy;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+
+public class BasicTest {
+
+ @Test
+ public void testSingleSeriesScan() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ SeriesScanNode root =
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(2, plan.getInstances().size());
+ }
+
+ @Test
+ public void testSingleSeriesScanRewriteSource() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ SeriesScanNode root =
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ PlanNode rootAfterRewrite = planner.rewriteSource();
+ assertEquals(2, rootAfterRewrite.getChildren().size());
+ }
+
+ @Test
+ public void testRewriteSourceNode() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root));
+ PlanNode newRoot = planner.rewriteSource();
+ assertEquals(4, newRoot.getChildren().get(0).getChildren().size());
+ }
+
+ @Test
+ public void testRewriteMetaSourceNode() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ SchemaQueryMergeNode metaMergeNode = new SchemaQueryMergeNode(queryId.genPlanNodeId(), false);
+ metaMergeNode.addChild(
+ new TimeSeriesSchemaScanNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d1.s1"),
+ null,
+ null,
+ 10,
+ 0,
+ false,
+ false,
+ false));
+ metaMergeNode.addChild(
+ new TimeSeriesSchemaScanNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d1.s2"),
+ null,
+ null,
+ 10,
+ 0,
+ false,
+ false,
+ false));
+ metaMergeNode.addChild(
+ new TimeSeriesSchemaScanNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d22.s1"),
+ null,
+ null,
+ 10,
+ 0,
+ false,
+ false,
+ false));
+ LimitNode root2 = new LimitNode(queryId.genPlanNodeId(), metaMergeNode, 10);
+ Analysis analysis = Util.constructAnalysis();
+ DistributionPlanner planner2 =
+ new DistributionPlanner(
+ analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root2));
+ PlanNode newRoot2 = planner2.rewriteSource();
+ assertEquals(newRoot2.getChildren().get(0).getChildren().size(), 2);
+ }
+
+ @Test
+ public void testAddExchangeNode() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(new MPPQueryContext(queryId), root));
+ PlanNode rootAfterRewrite = planner.rewriteSource();
+ PlanNode rootWithExchange = planner.addExchangeNode(rootAfterRewrite);
+ assertEquals(4, rootWithExchange.getChildren().get(0).getChildren().size());
+ int exchangeNodeCount = 0;
+ for (PlanNode child : rootWithExchange.getChildren().get(0).getChildren()) {
+ exchangeNodeCount += child instanceof ExchangeNode ? 1 : 0;
+ }
+ assertEquals(2, exchangeNodeCount);
+ }
+
+ @Test
+ public void testSplitFragment() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d22.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ PlanNode rootAfterRewrite = planner.rewriteSource();
+ PlanNode rootWithExchange = planner.addExchangeNode(rootAfterRewrite);
+ SubPlan subPlan = planner.splitFragment(rootWithExchange);
+ assertEquals(subPlan.getChildren().size(), 2);
+ }
+
+ @Test
+ public void testParallelPlan() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s2", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+ timeJoinNode.addChild(
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d333.s1", TSDataType.INT32),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ }
+
+ @Test
+ public void testSingleAlignedSeries() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query_aligned");
+ TimeJoinNode timeJoinNode = new TimeJoinNode(queryId.genPlanNodeId(), OrderBy.TIMESTAMP_ASC);
+
+ timeJoinNode.addChild(
+ new AlignedSeriesScanNode(
+ queryId.genPlanNodeId(),
+ new AlignedPath("root.sg.d22", Arrays.asList("s1", "s2")),
+ OrderBy.TIMESTAMP_ASC));
+
+ LimitNode root = new LimitNode(queryId.genPlanNodeId(), timeJoinNode, 10);
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(1, plan.getInstances().size());
+ }
+
+ @Test
+ public void testInsertRowNodeParallelPlan() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_write");
+ InsertRowNode insertRowNode =
+ new InsertRowNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d1"),
+ false,
+ new String[] {
+ "s1",
+ },
+ new TSDataType[] {TSDataType.INT32},
+ 1L,
+ new Object[] {10},
+ false);
+ insertRowNode.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.INT32),
+ });
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ context.setQueryType(QueryType.WRITE);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, insertRowNode));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(1, plan.getInstances().size());
+ }
+
+ @Test
+ public void testInsertRowsNodeParallelPlan() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_write");
+ InsertRowNode insertRowNode1 =
+ new InsertRowNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d1"),
+ false,
+ new String[] {"s1"},
+ new TSDataType[] {TSDataType.INT32},
+ 1L,
+ new Object[] {10},
+ false);
+ insertRowNode1.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.INT32),
+ });
+
+ InsertRowNode insertRowNode2 =
+ new InsertRowNode(
+ queryId.genPlanNodeId(),
+ new PartialPath("root.sg.d1"),
+ false,
+ new String[] {"s1"},
+ new TSDataType[] {TSDataType.INT32},
+ 100000L,
+ new Object[] {10},
+ false);
+ insertRowNode2.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.INT32),
+ });
+
+ InsertRowsNode node = new InsertRowsNode(queryId.genPlanNodeId());
+ node.setInsertRowNodeList(Arrays.asList(insertRowNode1, insertRowNode2));
+ node.setInsertRowNodeIndexList(Arrays.asList(0, 1));
+
+ Analysis analysis = Util.constructAnalysis();
+
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ context.setQueryType(QueryType.WRITE);
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, node));
+ DistributedQueryPlan plan = planner.planFragments();
+ assertEquals(1, plan.getInstances().size());
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/DeleteTimeseriesTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/DeleteTimeseriesTest.java
new file mode 100644
index 0000000000..02a094aca9
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/DeleteTimeseriesTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.common.QueryId;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.planner.LogicalPlanner;
+import org.apache.iotdb.db.mpp.plan.planner.distribution.DistributionPlanner;
+import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeUtil;
+import org.apache.iotdb.db.mpp.plan.statement.metadata.DeleteTimeSeriesStatement;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class DeleteTimeseriesTest {
+
+ @Test
+ public void test1DataRegion1SchemaRegion() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_group_by_level_two_series");
+ String d2s1Path = "root.sg.d22.s1";
+ List<PartialPath> paths = Collections.singletonList(new PartialPath(d2s1Path));
+ Analysis analysis = Util.constructAnalysis();
+ LogicalQueryPlan logicalQueryPlan = constructLogicalPlan(queryId, paths, analysis);
+
+ System.out.println(PlanNodeUtil.nodeToString(logicalQueryPlan.getRootNode()));
+
+ DistributionPlanner planner = new DistributionPlanner(analysis, logicalQueryPlan);
+ DistributedQueryPlan distributedQueryPlan = planner.planFragments();
+ distributedQueryPlan.getInstances().forEach(System.out::println);
+ Assert.assertEquals(6, distributedQueryPlan.getInstances().size());
+ }
+
+ private LogicalQueryPlan constructLogicalPlan(
+ QueryId queryId, List<PartialPath> paths, Analysis analysis) throws IllegalPathException {
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ DeleteTimeSeriesStatement statement = new DeleteTimeSeriesStatement();
+
+ analysis.setStatement(statement);
+ statement.setPartialPaths(paths);
+ LogicalPlanner planner = new LogicalPlanner(context, new ArrayList<>());
+ return planner.plan(analysis);
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/Util.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/Util.java
new file mode 100644
index 0000000000..16694a6ae3
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/distribution/Util.java
@@ -0,0 +1,191 @@
+/*
+ * 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.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.partition.DataPartition;
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.expression.leaf.TimeSeriesOperand;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class Util {
+ public static Analysis constructAnalysis() throws IllegalPathException {
+
+ SeriesPartitionExecutor executor =
+ SeriesPartitionExecutor.getSeriesPartitionExecutor(
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+ Analysis analysis = new Analysis();
+
+ String device1 = "root.sg.d1";
+ String device2 = "root.sg.d22";
+ String device3 = "root.sg.d333";
+ String device4 = "root.sg.d4444";
+
+ TRegionReplicaSet dataRegion1 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 1),
+ Arrays.asList(
+ genDataNodeLocation(11, "192.0.1.1"), genDataNodeLocation(12, "192.0.1.2")));
+
+ TRegionReplicaSet dataRegion2 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 2),
+ Arrays.asList(
+ genDataNodeLocation(21, "192.0.2.1"), genDataNodeLocation(22, "192.0.2.2")));
+
+ TRegionReplicaSet dataRegion3 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 3),
+ Arrays.asList(
+ genDataNodeLocation(31, "192.0.3.1"), genDataNodeLocation(32, "192.0.3.2")));
+
+ TRegionReplicaSet dataRegion4 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 4),
+ Arrays.asList(
+ genDataNodeLocation(41, "192.0.4.1"), genDataNodeLocation(42, "192.0.4.2")));
+
+ TRegionReplicaSet dataRegion5 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 5),
+ Arrays.asList(
+ genDataNodeLocation(51, "192.0.5.1"), genDataNodeLocation(52, "192.0.5.2")));
+
+ DataPartition dataPartition =
+ new DataPartition(
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+
+ Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
+ dataPartitionMap = new HashMap<>();
+ Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> sgPartitionMap =
+ new HashMap<>();
+
+ List<TRegionReplicaSet> d1DataRegions = new ArrayList<>();
+ d1DataRegions.add(dataRegion1);
+ d1DataRegions.add(dataRegion2);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d1DataRegionMap = new HashMap<>();
+ d1DataRegionMap.put(new TTimePartitionSlot(), d1DataRegions);
+
+ List<TRegionReplicaSet> d2DataRegions = new ArrayList<>();
+ d2DataRegions.add(dataRegion3);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d2DataRegionMap = new HashMap<>();
+ d2DataRegionMap.put(new TTimePartitionSlot(), d2DataRegions);
+
+ List<TRegionReplicaSet> d3DataRegions = new ArrayList<>();
+ d3DataRegions.add(dataRegion1);
+ d3DataRegions.add(dataRegion4);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d3DataRegionMap = new HashMap<>();
+ d3DataRegionMap.put(new TTimePartitionSlot(), d3DataRegions);
+
+ List<TRegionReplicaSet> d4DataRegions = new ArrayList<>();
+ d4DataRegions.add(dataRegion1);
+ d4DataRegions.add(dataRegion4);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> d4DataRegionMap = new HashMap<>();
+ d4DataRegionMap.put(new TTimePartitionSlot(), d4DataRegions);
+
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device1), d1DataRegionMap);
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device2), d2DataRegionMap);
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device3), d3DataRegionMap);
+ sgPartitionMap.put(executor.getSeriesPartitionSlot(device4), d4DataRegionMap);
+
+ dataPartitionMap.put("root.sg", sgPartitionMap);
+
+ dataPartition.setDataPartitionMap(dataPartitionMap);
+
+ analysis.setDataPartitionInfo(dataPartition);
+
+ // construct AggregationExpression for GroupByLevel
+ Map<String, Set<Expression>> aggregationExpression = new HashMap<>();
+ Set<Expression> s1Expression = new HashSet<>();
+ s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d1.s1")));
+ s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d22.s1")));
+ s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d333.s1")));
+ s1Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d4444.s1")));
+
+ Set<Expression> s2Expression = new HashSet<>();
+ s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d1.s2")));
+ s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d22.s2")));
+ s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d333.s2")));
+ s2Expression.add(new TimeSeriesOperand(new PartialPath("root.sg.d4444.s2")));
+
+ aggregationExpression.put("root.sg.*.s1", s1Expression);
+ aggregationExpression.put("root.sg.*.s2", s2Expression);
+ // analysis.setAggregationExpressions(aggregationExpression);
+
+ // construct schema partition
+ SchemaPartition schemaPartition =
+ new SchemaPartition(
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+ Map<String, Map<TSeriesPartitionSlot, TRegionReplicaSet>> schemaPartitionMap = new HashMap<>();
+
+ TRegionReplicaSet schemaRegion1 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 11),
+ Arrays.asList(
+ genDataNodeLocation(11, "192.0.1.1"), genDataNodeLocation(12, "192.0.1.2")));
+
+ TRegionReplicaSet schemaRegion2 =
+ new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 21),
+ Arrays.asList(
+ genDataNodeLocation(21, "192.0.2.1"), genDataNodeLocation(22, "192.0.2.2")));
+
+ Map<TSeriesPartitionSlot, TRegionReplicaSet> schemaRegionMap = new HashMap<>();
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device1), schemaRegion1);
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device2), schemaRegion2);
+ schemaRegionMap.put(executor.getSeriesPartitionSlot(device3), schemaRegion2);
+ schemaPartitionMap.put("root.sg", schemaRegionMap);
+ schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
+
+ analysis.setDataPartitionInfo(dataPartition);
+ analysis.setSchemaPartitionInfo(schemaPartition);
+ return analysis;
+ }
+
+ private static TDataNodeLocation genDataNodeLocation(int dataNodeId, String ip) {
+ return new TDataNodeLocation()
+ .setDataNodeId(dataNodeId)
+ .setExternalEndPoint(new TEndPoint(ip, 9000))
+ .setDataBlockManagerEndPoint(new TEndPoint(ip, 9001))
+ .setInternalEndPoint(new TEndPoint(ip, 9002));
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/DeleteDataNodeSerdeTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/DeleteDataNodeSerdeTest.java
index 5dbde28d3d..86f8ce1dff 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/DeleteDataNodeSerdeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/DeleteDataNodeSerdeTest.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.plan.plan.node.write;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType;
@@ -37,11 +38,16 @@ public class DeleteDataNodeSerdeTest {
@Test
public void testSerializeAndDeserialize() throws IllegalPathException {
- PlanNodeId planNodeId = new PlanNodeId("DeleteDataNode");
+ PlanNodeId planNodeId = new PlanNodeId("InvalidateSchemaCacheNode");
+ QueryId queryId = new QueryId("query");
List<PartialPath> pathList = new ArrayList<>();
pathList.add(new PartialPath("root.sg.d1.s1"));
pathList.add(new PartialPath("root.sg.d2.*"));
- DeleteDataNode deleteDataNode = new DeleteDataNode(planNodeId, pathList);
+ List<String> storageGroups = new ArrayList<>();
+ storageGroups.add("root.sg1");
+ storageGroups.add("root.sg2");
+ DeleteDataNode deleteDataNode =
+ new DeleteDataNode(planNodeId, queryId, pathList, storageGroups);
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
deleteDataNode.serialize(byteBuffer);
@@ -52,10 +58,19 @@ public class DeleteDataNodeSerdeTest {
Assert.assertEquals(planNodeId, deserializedNode.getPlanNodeId());
deleteDataNode = (DeleteDataNode) deserializedNode;
+
+ Assert.assertEquals(queryId, deleteDataNode.getQueryId());
+
List<PartialPath> deserializedPathList = deleteDataNode.getPathList();
Assert.assertEquals(pathList.size(), deserializedPathList.size());
for (int i = 0; i < pathList.size(); i++) {
Assert.assertEquals(pathList.get(i), deserializedPathList.get(i));
}
+
+ List<String> deserializedStorageGroups = deleteDataNode.getStorageGroups();
+ Assert.assertEquals(storageGroups.size(), deserializedStorageGroups.size());
+ for (int i = 0; i < storageGroups.size(); i++) {
+ Assert.assertEquals(storageGroups.get(i), deserializedStorageGroups.get(i));
+ }
}
}