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 2024/07/04 11:33:24 UTC
(iotdb) branch ty/TableModelGrammar updated: Add table model distributed plan generate rules
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ty/TableModelGrammar
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/ty/TableModelGrammar by this push:
new 7070aacfcdf Add table model distributed plan generate rules
7070aacfcdf is described below
commit 7070aacfcdf90a21c930c6262d1009743a2ffdab
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Thu Jul 4 19:33:18 2024 +0800
Add table model distributed plan generate rules
---
.../TableModelStatementMemorySourceVisitor.java | 30 +-
.../plan/planner/plan/node/PlanNode.java | 2 +-
.../plan/planner/plan/node/PlanNodeType.java | 3 +-
.../plan/relational/metadata/ColumnSchema.java | 1 +
.../plan/relational/metadata/MetadataUtil.java | 2 +-
.../plan/relational/planner/LogicalPlanner.java | 33 +-
.../plan/relational/planner/RelationPlanner.java | 22 +-
.../plan/relational/planner/Symbol.java | 5 +
.../planner/distribute/AddExchangeNodes.java | 94 +
.../distribute/DistributedPlanGenerator.java | 493 +++++
.../planner/distribute/ExchangeNodeGenerator.java | 219 --
.../planner/distribute/SimplePlanRewriter.java | 44 -
.../distribute/TableDistributionPlanner.java | 47 +-
.../TableModelTypeProviderExtractor.java | 7 +
.../plan/relational/planner/node/CollectNode.java | 20 +
.../plan/relational/planner/node/FilterNode.java | 21 +
.../plan/relational/planner/node/LimitNode.java | 22 +-
.../relational/planner/node/MergeSortNode.java | 21 +
.../plan/relational/planner/node/OffsetNode.java | 20 +
.../plan/relational/planner/node/OutputNode.java | 43 +-
.../plan/relational/planner/node/ProjectNode.java | 20 +
.../plan/relational/planner/node/SortNode.java | 44 +-
.../relational/planner/node/StreamSortNode.java | 43 +-
.../relational/planner/node/TableScanNode.java | 64 +-
.../plan/relational/planner/node/TopKNode.java | 59 +-
.../planner/optimizations/PruneUnUsedColumns.java | 16 +-
.../optimizations/PushPredicateIntoTableScan.java | 88 +-
.../RemoveRedundantIdentityProjections.java | 2 +-
.../planner/optimizations/SimplifyExpressions.java | 2 +-
...lPlanOptimizer.java => TablePlanOptimizer.java} | 2 +-
.../plan/relational/analyzer/AnalyzerTest.java | 197 +-
.../analyzer/MockTableModelDataPartition.java | 166 ++
.../relational/analyzer/MockTablePartition.java | 174 --
.../plan/relational/analyzer/SortTest.java | 2251 ++++++++++++++++++++
.../plan/relational/analyzer/TestMatadata.java | 30 +-
.../iotdb/commons/partition/DataPartition.java | 17 +-
36 files changed, 3688 insertions(+), 636 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/memory/TableModelStatementMemorySourceVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/memory/TableModelStatementMemorySourceVisitor.java
index 7b74850287c..ec9c5d145f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/memory/TableModelStatementMemorySourceVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/memory/TableModelStatementMemorySourceVisitor.java
@@ -27,7 +27,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanGraphPrinter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.ExchangeNodeGenerator;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.AddExchangeNodes;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.DistributedPlanGenerator;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Explain;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Node;
@@ -38,6 +39,7 @@ import org.apache.tsfile.read.common.block.TsBlock;
import java.util.Collections;
import java.util.List;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.iotdb.db.queryengine.common.header.DatasetHeader.EMPTY_HEADER;
import static org.apache.iotdb.db.queryengine.plan.execution.memory.StatementMemorySourceVisitor.getStatementMemorySource;
@@ -72,19 +74,23 @@ public class TableModelStatementMemorySourceVisitor
return new StatementMemorySource(new TsBlock(0), header);
}
- // TODO(beyyes) adapt this logic after optimize ExchangeNodeAdder
- ExchangeNodeGenerator.PlanContext exchangeContext =
- new ExchangeNodeGenerator.PlanContext(context.getQueryContext(), context.getAnalysis());
- List<PlanNode> distributedPlanNodeResult =
- new ExchangeNodeGenerator().visitPlan(logicalPlan.getRootNode(), exchangeContext);
+ // generate table model distributed plan
+ DistributedPlanGenerator.PlanContext planContext = new DistributedPlanGenerator.PlanContext();
+ List<PlanNode> distributedPlanResult =
+ new DistributedPlanGenerator(context.getQueryContext(), context.getAnalysis())
+ .genResult(logicalPlan.getRootNode(), planContext);
+ checkArgument(distributedPlanResult.size() == 1, "Root node must return only one");
+
+ // add exchange node for distributed plan
+ PlanNode outputNodeWithExchange =
+ new AddExchangeNodes(context.getQueryContext())
+ .addExchangeNodes(distributedPlanResult.get(0), planContext);
List<String> lines =
- distributedPlanNodeResult
- .get(0)
- .accept(
- new PlanGraphPrinter(),
- new PlanGraphPrinter.GraphContext(
- context.getQueryContext().getTypeProvider().getTemplatedInfo()));
+ outputNodeWithExchange.accept(
+ new PlanGraphPrinter(),
+ new PlanGraphPrinter.GraphContext(
+ context.getQueryContext().getTypeProvider().getTemplatedInfo()));
return getStatementMemorySource(header, lines);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
index ae5340e335e..3e6942dd147 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
@@ -218,7 +218,7 @@ public abstract class PlanNode implements IConsensusRequest {
return Objects.hash(id);
}
- // =========================== Used for Relational Model ============================
+ // =========================== Used for Table Model ============================
public List<Symbol> getOutputSymbols() {
throw new UnsupportedOperationException("This planNode does not support getOutputSymbols().");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
index 589d6206950..91ab840732a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
@@ -239,7 +239,8 @@ public enum PlanNodeType {
TABLE_SORT_NODE((short) 1006),
TABLE_MERGESORT_NODE((short) 1007),
TABLE_TOPK_NODE((short) 1008),
- TABLE_COLLECT_NODE((short) 1009);
+ TABLE_COLLECT_NODE((short) 1009),
+ TABLE_STREAM_SORT_NODE((short) 1010);
public static final int BYTES = Short.BYTES;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/ColumnSchema.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/ColumnSchema.java
index 36afc7d1461..715c9488127 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/ColumnSchema.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/ColumnSchema.java
@@ -53,6 +53,7 @@ public class ColumnSchema {
public ColumnSchema(
String name, Type type, boolean hidden, TsTableColumnCategory columnCategory) {
+ requireNonNull(name, "name is null");
requireNonNull(type, "type is null");
this.name = name.toLowerCase(ENGLISH);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/MetadataUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/MetadataUtil.java
index b97141817b0..34a14a70e5e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/MetadataUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/MetadataUtil.java
@@ -54,7 +54,7 @@ public class MetadataUtil {
}
public static void checkObjectName(String dbName, String objectName) {
- checkLowerCase(dbName, "schemaName");
+ checkLowerCase(dbName, "dbName");
checkLowerCase(objectName, "objectName");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/LogicalPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/LogicalPlanner.java
index 76d76fd000a..ac17c3ff500 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/LogicalPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/LogicalPlanner.java
@@ -36,9 +36,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CreateTableD
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PruneUnUsedColumns;
import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PushPredicateIntoTableScan;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.RelationalPlanOptimizer;
import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.RemoveRedundantIdentityProjections;
import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.SimplifyExpressions;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.TablePlanOptimizer;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Explain;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice;
@@ -66,7 +66,7 @@ public class LogicalPlanner {
private final MPPQueryContext context;
private final SessionInfo sessionInfo;
private final SymbolAllocator symbolAllocator = new SymbolAllocator();
- private final List<RelationalPlanOptimizer> relationalPlanOptimizers;
+ private final List<TablePlanOptimizer> tablePlanOptimizers;
private final Metadata metadata;
private final WarningCollector warningCollector;
@@ -80,7 +80,7 @@ public class LogicalPlanner {
this.sessionInfo = requireNonNull(sessionInfo, "session is null");
this.warningCollector = requireNonNull(warningCollector, "warningCollector is null");
- this.relationalPlanOptimizers =
+ this.tablePlanOptimizers =
Arrays.asList(
new SimplifyExpressions(),
new PruneUnUsedColumns(),
@@ -88,10 +88,23 @@ public class LogicalPlanner {
new RemoveRedundantIdentityProjections());
}
+ public LogicalPlanner(
+ MPPQueryContext context,
+ Metadata metadata,
+ SessionInfo sessionInfo,
+ List<TablePlanOptimizer> tablePlanOptimizers,
+ WarningCollector warningCollector) {
+ this.context = context;
+ this.metadata = metadata;
+ this.sessionInfo = requireNonNull(sessionInfo, "session is null");
+ this.warningCollector = requireNonNull(warningCollector, "warningCollector is null");
+ this.tablePlanOptimizers = tablePlanOptimizers;
+ }
+
public LogicalQueryPlan plan(Analysis analysis) {
PlanNode planNode = planStatement(analysis, analysis.getStatement());
- relationalPlanOptimizers.forEach(
+ tablePlanOptimizers.forEach(
optimizer -> optimizer.optimize(planNode, analysis, metadata, sessionInfo, context));
return new LogicalQueryPlan(context, planNode);
@@ -166,12 +179,6 @@ public class LogicalPlanner {
return new RelationPlanner(analysis, symbolAllocator, context, sessionInfo, ImmutableMap.of());
}
- private enum Stage {
- CREATED,
- OPTIMIZED,
- OPTIMIZED_AND_VALIDATED
- }
-
private PlanNode planCreateDevice(CreateDevice statement, Analysis analysis) {
context.setQueryType(QueryType.WRITE);
@@ -270,4 +277,10 @@ public class LogicalPlanner {
}
return columnHeaderList;
}
+
+ private enum Stage {
+ CREATED,
+ OPTIMIZED,
+ OPTIMIZED_AND_VALIDATED
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
index fca9a8af675..d85fcc9056c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
@@ -13,6 +13,7 @@
*/
package org.apache.iotdb.db.queryengine.plan.relational.planner;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.QueryId;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
@@ -42,6 +43,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -98,14 +100,23 @@ public class RelationPlanner extends AstVisitor<RelationPlan, Void> {
ImmutableList.Builder<Symbol> outputSymbolsBuilder = ImmutableList.builder();
ImmutableMap.Builder<Symbol, ColumnSchema> symbolToColumnSchema = ImmutableMap.builder();
Collection<Field> fields = scope.getRelationType().getAllFields();
-
+ // on the basis of that the order of fields is same with the column category order of segments
+ // in DeviceEntry
+ Map<Symbol, Integer> idAndAttributeIndexMap = new HashMap<>();
+ int idIndex = 0, attributeIndex = 0;
for (Field field : fields) {
Symbol symbol = symbolAllocator.newSymbol(field);
outputSymbolsBuilder.add(symbol);
+ TsTableColumnCategory category = field.getColumnCategory();
symbolToColumnSchema.put(
symbol,
new ColumnSchema(
- field.getName().get(), field.getType(), field.isHidden(), field.getColumnCategory()));
+ field.getName().orElse(null), field.getType(), field.isHidden(), category));
+ if (category == TsTableColumnCategory.ID) {
+ idAndAttributeIndexMap.put(symbol, idIndex++);
+ } else if (category == TsTableColumnCategory.ATTRIBUTE) {
+ idAndAttributeIndexMap.put(symbol, attributeIndex++);
+ }
}
List<Symbol> outputSymbols = outputSymbolsBuilder.build();
@@ -113,13 +124,16 @@ public class RelationPlanner extends AstVisitor<RelationPlan, Void> {
if (!qualifiedName.getPrefix().isPresent()) {
throw new IllegalStateException("Table " + table.getName() + " has no prefix!");
}
+
TableScanNode tableScanNode =
new TableScanNode(
idAllocator.genPlanNodeId(),
new QualifiedObjectName(
- qualifiedName.getPrefix().get().toString(), qualifiedName.getSuffix()),
+ qualifiedName.getPrefix().map(QualifiedName::toString).orElse(null),
+ qualifiedName.getSuffix()),
outputSymbols,
- symbolToColumnSchema.build());
+ symbolToColumnSchema.build(),
+ idAndAttributeIndexMap);
return new RelationPlan(tableScanNode, scope, outputSymbols);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/Symbol.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/Symbol.java
index 4f48df0890f..eb90d8183e5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/Symbol.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/Symbol.java
@@ -44,6 +44,11 @@ public class Symbol implements Comparable<Symbol> {
this.name = name;
}
+ public static Symbol of(String name) {
+ requireNonNull(name, "name is null");
+ return new Symbol(name);
+ }
+
public String getName() {
return name;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
new file mode 100644
index 00000000000..61f993ee600
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
@@ -0,0 +1,94 @@
+/*
+ * 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.queryengine.plan.relational.planner.distribute;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistribution;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+
+import static org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.SAME_WITH_ALL_CHILDREN;
+import static org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.SAME_WITH_SOME_CHILD;
+
+public class AddExchangeNodes extends PlanVisitor<PlanNode, DistributedPlanGenerator.PlanContext> {
+
+ private final MPPQueryContext queryContext;
+
+ public AddExchangeNodes(MPPQueryContext queryContext) {
+ this.queryContext = queryContext;
+ }
+
+ public PlanNode addExchangeNodes(PlanNode node, DistributedPlanGenerator.PlanContext context) {
+ return node.accept(this, context);
+ }
+
+ @Override
+ public PlanNode visitPlan(PlanNode node, DistributedPlanGenerator.PlanContext context) {
+ if (node instanceof WritePlanNode) {
+ return node;
+ }
+
+ PlanNode newNode = node.clone();
+ if (node.getChildren().size() == 1) {
+ newNode.addChild(node.getChildren().get(0).accept(this, context));
+ context.nodeDistributionMap.put(
+ node.getPlanNodeId(),
+ new NodeDistribution(
+ SAME_WITH_ALL_CHILDREN,
+ context
+ .nodeDistributionMap
+ .get(node.getChildren().get(0).getPlanNodeId())
+ .getRegion()));
+ return newNode;
+ }
+
+ for (PlanNode child : node.getChildren()) {
+ PlanNode rewriteNode = child.accept(this, context);
+
+ TRegionReplicaSet region =
+ context.nodeDistributionMap.get(rewriteNode.getPlanNodeId()).getRegion();
+ if (!region.equals(context.mostUsedDataRegion)) {
+ ExchangeNode exchangeNode = new ExchangeNode(queryContext.getQueryId().genPlanNodeId());
+ exchangeNode.addChild(rewriteNode);
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(rewriteNode);
+ }
+ }
+
+ context.nodeDistributionMap.put(
+ node.getPlanNodeId(),
+ new NodeDistribution(SAME_WITH_SOME_CHILD, context.mostUsedDataRegion));
+
+ return newNode;
+ }
+
+ @Override
+ public PlanNode visitTableScan(TableScanNode node, DistributedPlanGenerator.PlanContext context) {
+ context.nodeDistributionMap.put(
+ node.getPlanNodeId(),
+ new NodeDistribution(SAME_WITH_ALL_CHILDREN, node.getRegionReplicaSet()));
+ return node;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java
new file mode 100644
index 00000000000..0bf5bb32a1b
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java
@@ -0,0 +1,493 @@
+/*
+ * Licensed 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.queryengine.plan.relational.planner.distribute;
+
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistribution;
+import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.AbstractSchemaMergeNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.TableDeviceSourceNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SourceNode;
+import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+
+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.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PushPredicateIntoTableScan.containsDiffFunction;
+import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
+
+/** This class is used to generate distributed plan for table model. */
+public class DistributedPlanGenerator
+ extends PlanVisitor<List<PlanNode>, DistributedPlanGenerator.PlanContext> {
+ private final QueryId queryId;
+ private final Analysis analysis;
+ Map<PlanNodeId, OrderingScheme> nodeOrderingMap = new HashMap<>();
+
+ public DistributedPlanGenerator(MPPQueryContext queryContext, Analysis analysis) {
+ this.queryId = queryContext.getQueryId();
+ this.analysis = analysis;
+ }
+
+ public List<PlanNode> genResult(PlanNode node, PlanContext context) {
+ return node.accept(this, context);
+ }
+
+ @Override
+ public List<PlanNode> visitPlan(PlanNode node, DistributedPlanGenerator.PlanContext context) {
+ if (node instanceof WritePlanNode) {
+ return Collections.singletonList(node);
+ }
+
+ List<List<PlanNode>> children =
+ node.getChildren().stream()
+ .map(child -> child.accept(this, context))
+ .collect(toImmutableList());
+
+ PlanNode newNode = node.clone();
+ for (List<PlanNode> planNodes : children) {
+ planNodes.forEach(newNode::addChild);
+ }
+ return Collections.singletonList(newNode);
+ }
+
+ @Override
+ public List<PlanNode> visitOutput(OutputNode node, PlanContext context) {
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
+
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes));
+ return Collections.singletonList(node);
+ }
+
+ @Override
+ public List<PlanNode> visitLimit(LimitNode node, PlanContext context) {
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
+
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ // push down LimitNode in distributed plan optimize rule
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes));
+ return Collections.singletonList(node);
+ }
+
+ @Override
+ public List<PlanNode> visitOffset(OffsetNode node, PlanContext context) {
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
+
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes));
+ return Collections.singletonList(node);
+ }
+
+ @Override
+ public List<PlanNode> visitProject(ProjectNode node, PlanContext context) {
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
+
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ for (Expression expression : node.getAssignments().getMap().values()) {
+ if (containsDiffFunction(expression)) {
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes));
+ return Collections.singletonList(node);
+ }
+ }
+
+ List<PlanNode> resultNodeList = new ArrayList<>();
+ for (int i = 0; i < childrenNodes.size(); i++) {
+ PlanNode child = childrenNodes.get(i);
+ ProjectNode subProjectNode =
+ new ProjectNode(queryId.genPlanNodeId(), child, node.getAssignments());
+ resultNodeList.add(subProjectNode);
+ if (i == 0) {
+ nodeOrderingMap.put(subProjectNode.getPlanNodeId(), childOrdering);
+ }
+ }
+ return resultNodeList;
+ }
+
+ @Override
+ public List<PlanNode> visitSort(SortNode node, PlanContext context) {
+ context.expectedOrderingScheme = node.getOrderingScheme();
+ context.hasSortNode = true;
+ nodeOrderingMap.put(node.getPlanNodeId(), node.getOrderingScheme());
+
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ // may have ProjectNode above SortNode later, so use MergeSortNode but not return SortNode list
+ MergeSortNode mergeSortNode =
+ new MergeSortNode(
+ queryId.genPlanNodeId(), node.getOrderingScheme(), node.getOutputSymbols());
+ for (PlanNode child : childrenNodes) {
+ SortNode subSortNode =
+ new SortNode(queryId.genPlanNodeId(), child, node.getOrderingScheme(), false);
+ mergeSortNode.addChild(subSortNode);
+ }
+ nodeOrderingMap.put(mergeSortNode.getPlanNodeId(), mergeSortNode.getOrderingScheme());
+
+ return Collections.singletonList(mergeSortNode);
+ }
+
+ @Override
+ public List<PlanNode> visitFilter(FilterNode node, PlanContext context) {
+ List<PlanNode> childrenNodes = node.getChild().accept(this, context);
+ OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ if (childOrdering != null) {
+ nodeOrderingMap.put(node.getPlanNodeId(), childOrdering);
+ }
+
+ if (childrenNodes.size() == 1) {
+ node.setChild(childrenNodes.get(0));
+ return Collections.singletonList(node);
+ }
+
+ if (containsDiffFunction(node.getPredicate())) {
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes));
+ return Collections.singletonList(node);
+ }
+
+ List<PlanNode> resultNodeList = new ArrayList<>();
+ for (int i = 0; i < childrenNodes.size(); i++) {
+ PlanNode child = childrenNodes.get(i);
+ FilterNode subFilterNode =
+ new FilterNode(queryId.genPlanNodeId(), child, node.getPredicate());
+ resultNodeList.add(subFilterNode);
+ if (i == 0) {
+ nodeOrderingMap.put(subFilterNode.getPlanNodeId(), childOrdering);
+ }
+ }
+ return resultNodeList;
+ }
+
+ @Override
+ public List<PlanNode> visitTableScan(TableScanNode node, PlanContext context) {
+
+ Map<TRegionReplicaSet, TableScanNode> tableScanNodeMap = new HashMap<>();
+
+ for (DeviceEntry deviceEntry : node.getDeviceEntries()) {
+ List<TRegionReplicaSet> regionReplicaSets =
+ analysis
+ .getDataPartitionInfo()
+ .getDataRegionReplicaSetWithTimeFilter(
+ node.getQualifiedObjectName().getDatabaseName(),
+ deviceEntry.getDeviceID(),
+ node.getTimeFilter());
+ for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
+ TableScanNode tableScanNode =
+ tableScanNodeMap.computeIfAbsent(
+ regionReplicaSet,
+ k -> {
+ TableScanNode scanNode =
+ new TableScanNode(
+ queryId.genPlanNodeId(),
+ node.getQualifiedObjectName(),
+ node.getOutputSymbols(),
+ node.getAssignments(),
+ new ArrayList<>(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate());
+ scanNode.setRegionReplicaSet(regionReplicaSet);
+ return scanNode;
+ });
+ tableScanNode.appendDeviceEntry(deviceEntry);
+ }
+ }
+
+ context.hasExchangeNode = tableScanNodeMap.size() > 1;
+
+ List<PlanNode> resultTableScanNodeList = new ArrayList<>();
+ TRegionReplicaSet mostUsedDataRegion = null;
+ int maxDeviceEntrySizeOfTableScan = 0;
+ for (Map.Entry<TRegionReplicaSet, TableScanNode> entry : tableScanNodeMap.entrySet()) {
+ TRegionReplicaSet regionReplicaSet = entry.getKey();
+ TableScanNode subTableScanNode = entry.getValue();
+ subTableScanNode.setPlanNodeId(queryId.genPlanNodeId());
+ subTableScanNode.setRegionReplicaSet(regionReplicaSet);
+ resultTableScanNodeList.add(subTableScanNode);
+
+ if (mostUsedDataRegion == null
+ || subTableScanNode.getDeviceEntries().size() > maxDeviceEntrySizeOfTableScan) {
+ mostUsedDataRegion = regionReplicaSet;
+ maxDeviceEntrySizeOfTableScan = subTableScanNode.getDeviceEntries().size();
+ }
+ }
+ context.mostUsedDataRegion = mostUsedDataRegion;
+
+ if (!context.hasSortNode) {
+ return resultTableScanNodeList;
+ }
+
+ processSortProperty(node, resultTableScanNodeList, context);
+ return resultTableScanNodeList;
+ }
+
+ private PlanNode mergeChildrenViaCollectOrMergeSort(
+ OrderingScheme childOrdering, List<PlanNode> childrenNodes) {
+ PlanNode firstChild = childrenNodes.get(0);
+
+ // children has sort property, use MergeSort to merge children
+ if (childOrdering != null) {
+ MergeSortNode mergeSortNode =
+ new MergeSortNode(queryId.genPlanNodeId(), childOrdering, firstChild.getOutputSymbols());
+ childrenNodes.forEach(mergeSortNode::addChild);
+ nodeOrderingMap.put(mergeSortNode.getPlanNodeId(), childOrdering);
+ return mergeSortNode;
+ }
+
+ // children has no sort property, use CollectNode to merge children
+ CollectNode collectNode = new CollectNode(queryId.genPlanNodeId());
+ childrenNodes.forEach(collectNode::addChild);
+ return collectNode;
+ }
+
+ private void processSortProperty(
+ TableScanNode tableScanNode, List<PlanNode> resultTableScanNodeList, PlanContext context) {
+ List<Symbol> newOrderingSymbols = new ArrayList<>();
+ List<SortOrder> newSortOrders = new ArrayList<>();
+ OrderingScheme expectedOrderingScheme = context.expectedOrderingScheme;
+
+ for (Symbol symbol : expectedOrderingScheme.getOrderBy()) {
+ if (TIMESTAMP_STR.equalsIgnoreCase(symbol.getName())) {
+ if (!expectedOrderingScheme.getOrderings().get(symbol).isAscending()) {
+ // TODO(beyyes) move scan order judgement into logical plan optimizer
+ resultTableScanNodeList.forEach(
+ node -> ((TableScanNode) node).setScanOrder(Ordering.DESC));
+ }
+ break;
+ } else if (!tableScanNode.getIdAndAttributeIndexMap().containsKey(symbol)) {
+ break;
+ }
+
+ newOrderingSymbols.add(symbol);
+ newSortOrders.add(expectedOrderingScheme.getOrdering(symbol));
+ }
+
+ // no sort property can be pushed down into TableScanNode
+ if (newOrderingSymbols.isEmpty()) {
+ return;
+ }
+
+ List<Function<DeviceEntry, String>> orderingRules = new ArrayList<>();
+ for (Symbol symbol : newOrderingSymbols) {
+ int idx = tableScanNode.getIdAndAttributeIndexMap().get(symbol);
+ if (tableScanNode.getAssignments().get(symbol).getColumnCategory()
+ == TsTableColumnCategory.ID) {
+ // segments[0] is always tableName
+ orderingRules.add(deviceEntry -> (String) deviceEntry.getDeviceID().getSegments()[idx + 1]);
+ } else {
+ orderingRules.add(deviceEntry -> deviceEntry.getAttributeColumnValues().get(idx));
+ }
+ }
+
+ Comparator<DeviceEntry> comparator;
+ if (newSortOrders.get(0).isNullsFirst()) {
+ comparator =
+ newSortOrders.get(0).isAscending()
+ ? Comparator.nullsFirst(Comparator.comparing(orderingRules.get(0)))
+ : Comparator.nullsFirst(Comparator.comparing(orderingRules.get(0))).reversed();
+ } else {
+ comparator =
+ newSortOrders.get(0).isAscending()
+ ? Comparator.nullsLast(Comparator.comparing(orderingRules.get(0)))
+ : Comparator.nullsLast(Comparator.comparing(orderingRules.get(0))).reversed();
+ }
+ for (int i = 1; i < orderingRules.size(); i++) {
+ Comparator<DeviceEntry> thenComparator;
+ if (newSortOrders.get(i).isNullsFirst()) {
+ thenComparator =
+ newSortOrders.get(i).isAscending()
+ ? Comparator.nullsFirst(Comparator.comparing(orderingRules.get(i)))
+ : Comparator.nullsFirst(Comparator.comparing(orderingRules.get(i))).reversed();
+ } else {
+ thenComparator =
+ newSortOrders.get(i).isAscending()
+ ? Comparator.nullsLast(Comparator.comparing(orderingRules.get(i)))
+ : Comparator.nullsLast(Comparator.comparing(orderingRules.get(i))).reversed();
+ }
+ comparator = comparator.thenComparing(thenComparator);
+ }
+
+ OrderingScheme newOrderingScheme =
+ new OrderingScheme(
+ newOrderingSymbols,
+ IntStream.range(0, newOrderingSymbols.size())
+ .boxed()
+ .collect(Collectors.toMap(newOrderingSymbols::get, newSortOrders::get)));
+ for (PlanNode planNode : resultTableScanNodeList) {
+ TableScanNode scanNode = (TableScanNode) planNode;
+ nodeOrderingMap.put(scanNode.getPlanNodeId(), newOrderingScheme);
+ scanNode.getDeviceEntries().sort(comparator);
+ }
+ }
+
+ // ------------------- schema related interface ---------------------------------------------
+
+ @Override
+ public List<PlanNode> visitSchemaQueryMerge(SchemaQueryMergeNode node, PlanContext context) {
+ return Collections.singletonList(
+ addExchangeNodeForSchemaMerge(rewriteSchemaQuerySource(node, context), context));
+ }
+
+ private SchemaQueryMergeNode rewriteSchemaQuerySource(
+ SchemaQueryMergeNode node, PlanContext context) {
+ SchemaQueryMergeNode root = (SchemaQueryMergeNode) node.clone();
+
+ String database = ((TableDeviceSourceNode) node.getChildren().get(0)).getDatabase();
+ Set<TRegionReplicaSet> schemaRegionSet = new HashSet<>();
+ analysis
+ .getSchemaPartitionInfo()
+ .getSchemaPartitionMap()
+ .get(database)
+ .forEach(
+ (deviceGroupId, schemaRegionReplicaSet) -> schemaRegionSet.add(schemaRegionReplicaSet));
+
+ for (PlanNode child : node.getChildren()) {
+ for (TRegionReplicaSet schemaRegion : schemaRegionSet) {
+ SourceNode clonedChild = (SourceNode) child.clone();
+ clonedChild.setPlanNodeId(queryId.genPlanNodeId());
+ clonedChild.setRegionReplicaSet(schemaRegion);
+ root.addChild(clonedChild);
+ }
+ }
+ return root;
+ }
+
+ private PlanNode addExchangeNodeForSchemaMerge(
+ AbstractSchemaMergeNode node, PlanContext context) {
+ node.getChildren()
+ .forEach(
+ child ->
+ context.putNodeDistribution(
+ child.getPlanNodeId(),
+ new NodeDistribution(
+ NodeDistributionType.NO_CHILD,
+ ((SourceNode) child).getRegionReplicaSet())));
+ NodeDistribution nodeDistribution =
+ new NodeDistribution(NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN);
+ PlanNode newNode = node.clone();
+ nodeDistribution.setRegion(calculateSchemaRegionByChildren(node.getChildren(), context));
+ context.putNodeDistribution(newNode.getPlanNodeId(), nodeDistribution);
+ node.getChildren()
+ .forEach(
+ child -> {
+ if (!nodeDistribution
+ .getRegion()
+ .equals(context.getNodeDistribution(child.getPlanNodeId()).getRegion())) {
+ ExchangeNode exchangeNode = new ExchangeNode(queryId.genPlanNodeId());
+ exchangeNode.setChild(child);
+ exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
+ context.hasExchangeNode = true;
+ newNode.addChild(exchangeNode);
+ } else {
+ newNode.addChild(child);
+ }
+ });
+ return newNode;
+ }
+
+ private TRegionReplicaSet calculateSchemaRegionByChildren(
+ List<PlanNode> children, PlanContext context) {
+ // We always make the schemaRegion of SchemaMergeNode to be the same as its first child.
+ return context.getNodeDistribution(children.get(0).getPlanNodeId()).getRegion();
+ }
+
+ public static class PlanContext {
+ final Map<PlanNodeId, NodeDistribution> nodeDistributionMap;
+ boolean hasExchangeNode = false;
+ boolean hasSortNode = false;
+ OrderingScheme expectedOrderingScheme;
+ TRegionReplicaSet mostUsedDataRegion;
+
+ public PlanContext() {
+ this.nodeDistributionMap = new HashMap<>();
+ }
+
+ public NodeDistribution getNodeDistribution(PlanNodeId nodeId) {
+ return this.nodeDistributionMap.get(nodeId);
+ }
+
+ public void putNodeDistribution(PlanNodeId nodeId, NodeDistribution distribution) {
+ this.nodeDistributionMap.put(nodeId, distribution);
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/ExchangeNodeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/ExchangeNodeGenerator.java
deleted file mode 100644
index 809d1398aee..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/ExchangeNodeGenerator.java
+++ /dev/null
@@ -1,219 +0,0 @@
-/*
- * Licensed 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.queryengine.plan.relational.planner.distribute;
-
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
-import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistribution;
-import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.AbstractSchemaMergeNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.SchemaQueryMergeNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.read.TableDeviceSourceNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SourceNode;
-import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis;
-import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
-import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
-
-import java.util.ArrayList;
-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.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistributionType.SAME_WITH_ALL_CHILDREN;
-
-public class ExchangeNodeGenerator extends SimplePlanRewriter<ExchangeNodeGenerator.PlanContext> {
-
- @Override
- public List<PlanNode> visitTableScan(TableScanNode node, PlanContext context) {
-
- Map<TRegionReplicaSet, TableScanNode> tableScanNodeMap = new HashMap<>();
-
- for (DeviceEntry deviceEntry : node.getDeviceEntries()) {
- List<TRegionReplicaSet> regionReplicaSets =
- context
- .analysis
- .getDataPartitionInfo()
- .getDataRegionReplicaSetWithTimeFilter(
- node.getQualifiedObjectName().getDatabaseName(),
- deviceEntry.getDeviceID(),
- node.getTimeFilter());
- for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
- TableScanNode tableScanNode =
- tableScanNodeMap.computeIfAbsent(
- regionReplicaSet,
- k -> {
- TableScanNode scanNode =
- new TableScanNode(
- context.queryContext.getQueryId().genPlanNodeId(),
- node.getQualifiedObjectName(),
- node.getOutputSymbols(),
- node.getAssignments(),
- new ArrayList<>(),
- node.getIdAndAttributeIndexMap(),
- node.getScanOrder(),
- node.getTimePredicate().orElse(null),
- node.getPushDownPredicate());
- scanNode.setRegionReplicaSet(regionReplicaSet);
- return scanNode;
- });
- tableScanNode.appendDeviceEntry(deviceEntry);
- }
- }
-
- int i = 0;
- if (tableScanNodeMap.size() > 1) {
- context.hasExchangeNode = true;
- List<Symbol> orderBy = node.getOutputSymbols().subList(0, 1);
- Map<Symbol, SortOrder> orderings =
- Collections.singletonMap(node.getOutputSymbols().get(0), SortOrder.ASC_NULLS_LAST);
- OrderingScheme orderingScheme = new OrderingScheme(orderBy, orderings);
- MergeSortNode mergeSortNode =
- new MergeSortNode(
- context.queryContext.getQueryId().genPlanNodeId(),
- orderingScheme,
- node.getOutputSymbols());
-
- for (Map.Entry<TRegionReplicaSet, TableScanNode> entry : tableScanNodeMap.entrySet()) {
- TRegionReplicaSet regionReplicaSet = entry.getKey();
- TableScanNode subTableScanNode = entry.getValue();
- subTableScanNode.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- subTableScanNode.setRegionReplicaSet(regionReplicaSet);
- context.nodeDistributionMap.put(
- subTableScanNode.getPlanNodeId(),
- new NodeDistribution(SAME_WITH_ALL_CHILDREN, regionReplicaSet));
-
- // TODO not use 0 replica set as root replica?
- if (i == 0) {
- mergeSortNode.addChild(subTableScanNode);
- context.nodeDistributionMap.put(
- mergeSortNode.getPlanNodeId(),
- new NodeDistribution(SAME_WITH_ALL_CHILDREN, regionReplicaSet));
- } else {
- ExchangeNode exchangeNode =
- new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
- exchangeNode.addChild(subTableScanNode);
- mergeSortNode.addChild(exchangeNode);
- }
- i++;
- }
- return Collections.singletonList(mergeSortNode);
- } else {
- return Collections.singletonList(tableScanNodeMap.entrySet().iterator().next().getValue());
- }
- }
-
- @Override
- public List<PlanNode> visitSchemaQueryMerge(SchemaQueryMergeNode node, PlanContext context) {
- return Collections.singletonList(
- addExchangeNodeForSchemaMerge(rewriteSchemaQuerySource(node, context), context));
- }
-
- private SchemaQueryMergeNode rewriteSchemaQuerySource(
- SchemaQueryMergeNode node, PlanContext context) {
- SchemaQueryMergeNode root = (SchemaQueryMergeNode) node.clone();
-
- String database = ((TableDeviceSourceNode) node.getChildren().get(0)).getDatabase();
- Set<TRegionReplicaSet> schemaRegionSet = new HashSet<>();
- context
- .analysis
- .getSchemaPartitionInfo()
- .getSchemaPartitionMap()
- .get(database)
- .forEach(
- (deviceGroupId, schemaRegionReplicaSet) -> schemaRegionSet.add(schemaRegionReplicaSet));
-
- for (PlanNode child : node.getChildren()) {
- for (TRegionReplicaSet schemaRegion : schemaRegionSet) {
- SourceNode clonedChild = (SourceNode) child.clone();
- clonedChild.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
- clonedChild.setRegionReplicaSet(schemaRegion);
- root.addChild(clonedChild);
- }
- }
- return root;
- }
-
- private PlanNode addExchangeNodeForSchemaMerge(
- AbstractSchemaMergeNode node, PlanContext context) {
- node.getChildren()
- .forEach(
- child ->
- context.putNodeDistribution(
- child.getPlanNodeId(),
- new NodeDistribution(
- NodeDistributionType.NO_CHILD,
- ((SourceNode) child).getRegionReplicaSet())));
- NodeDistribution nodeDistribution =
- new NodeDistribution(NodeDistributionType.DIFFERENT_FROM_ALL_CHILDREN);
- PlanNode newNode = node.clone();
- nodeDistribution.setRegion(calculateSchemaRegionByChildren(node.getChildren(), context));
- context.putNodeDistribution(newNode.getPlanNodeId(), nodeDistribution);
- node.getChildren()
- .forEach(
- child -> {
- if (!nodeDistribution
- .getRegion()
- .equals(context.getNodeDistribution(child.getPlanNodeId()).getRegion())) {
- ExchangeNode exchangeNode =
- new ExchangeNode(context.queryContext.getQueryId().genPlanNodeId());
- exchangeNode.setChild(child);
- exchangeNode.setOutputColumnNames(child.getOutputColumnNames());
- context.hasExchangeNode = true;
- newNode.addChild(exchangeNode);
- } else {
- newNode.addChild(child);
- }
- });
- return newNode;
- }
-
- private TRegionReplicaSet calculateSchemaRegionByChildren(
- List<PlanNode> children, PlanContext context) {
- // We always make the schemaRegion of SchemaMergeNode to be the same as its first child.
- return context.getNodeDistribution(children.get(0).getPlanNodeId()).getRegion();
- }
-
- public static class PlanContext {
- final MPPQueryContext queryContext;
- final Map<PlanNodeId, NodeDistribution> nodeDistributionMap;
- TRegionReplicaSet mostlyUsedDataRegion;
- boolean hasExchangeNode = false;
-
- final Analysis analysis;
-
- public PlanContext(MPPQueryContext queryContext, Analysis analysis) {
- this.queryContext = queryContext;
- this.nodeDistributionMap = new HashMap<>();
- this.analysis = analysis;
- }
-
- public NodeDistribution getNodeDistribution(PlanNodeId nodeId) {
- return this.nodeDistributionMap.get(nodeId);
- }
-
- public void putNodeDistribution(PlanNodeId nodeId, NodeDistribution distribution) {
- this.nodeDistributionMap.put(nodeId, distribution);
- }
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/SimplePlanRewriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/SimplePlanRewriter.java
deleted file mode 100644
index 9c5ac404c6b..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/SimplePlanRewriter.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed 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.queryengine.plan.relational.planner.distribute;
-
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
-
-import java.util.Collections;
-import java.util.List;
-
-import static com.google.common.collect.ImmutableList.toImmutableList;
-
-public class SimplePlanRewriter<C> extends PlanVisitor<List<PlanNode>, C> {
-
- @Override
- public List<PlanNode> visitPlan(PlanNode node, C context) {
- if (node instanceof WritePlanNode) {
- return Collections.singletonList(node);
- }
-
- List<List<PlanNode>> children =
- node.getChildren().stream()
- .map(child -> child.accept(this, context))
- .collect(toImmutableList());
-
- PlanNode newNode = node.clone();
- for (List<PlanNode> planNodes : children) {
- planNodes.forEach(newNode::addChild);
- }
- return Collections.singletonList(newNode);
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
index 684bffc6acb..70a87f06dec 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
@@ -35,6 +35,7 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand.TIMESTAMP_EXPRESSION_STRING;
public class TableDistributionPlanner {
@@ -50,16 +51,18 @@ public class TableDistributionPlanner {
}
public DistributedQueryPlan plan() {
- ExchangeNodeGenerator.PlanContext exchangeContext =
- new ExchangeNodeGenerator.PlanContext(mppQueryContext, analysis);
- List<PlanNode> distributedPlanNodeResult =
- new ExchangeNodeGenerator().visitPlan(logicalQueryPlan.getRootNode(), exchangeContext);
- if (distributedPlanNodeResult.size() != 1) {
- throw new IllegalStateException("root node must return only one");
- }
-
- PlanNode outputNodeWithExchange = distributedPlanNodeResult.get(0);
+ // generate table model distributed plan
+ DistributedPlanGenerator.PlanContext planContext = new DistributedPlanGenerator.PlanContext();
+ List<PlanNode> distributedPlanResult =
+ new DistributedPlanGenerator(mppQueryContext, analysis)
+ .genResult(logicalQueryPlan.getRootNode(), planContext);
+ checkArgument(distributedPlanResult.size() == 1, "Root node must return only one");
+
+ // add exchange node for distributed plan
+ PlanNode outputNodeWithExchange =
+ new AddExchangeNodes(mppQueryContext)
+ .addExchangeNodes(distributedPlanResult.get(0), planContext);
if (analysis.getStatement() instanceof Query) {
analysis
.getRespDatasetHeader()
@@ -69,19 +72,21 @@ public class TableDistributionPlanner {
.filter(e -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(e))
.collect(Collectors.toList()));
}
- adjustUpStream(outputNodeWithExchange, exchangeContext);
+ adjustUpStream(outputNodeWithExchange, planContext);
+ // generate subPlan
SubPlan subPlan =
new SubPlanGenerator()
.splitToSubPlan(logicalQueryPlan.getContext().getQueryId(), outputNodeWithExchange);
subPlan.getPlanFragment().setRoot(true);
+ // generate fragment instances
List<FragmentInstance> fragmentInstances =
mppQueryContext.getQueryType() == QueryType.READ
? new TableModelQueryFragmentPlanner(subPlan, analysis, mppQueryContext).plan()
: new WriteFragmentParallelPlanner(subPlan, analysis, mppQueryContext).parallelPlan();
- // Only execute this step for READ operation
+ // only execute this step for READ operation
if (mppQueryContext.getQueryType() == QueryType.READ) {
setSinkForRootInstance(subPlan, fragmentInstances);
}
@@ -124,32 +129,32 @@ public class TableDistributionPlanner {
rootInstance.getFragment().setPlanNodeTree(sinkNode);
}
- private void adjustUpStream(PlanNode root, ExchangeNodeGenerator.PlanContext exchangeContext) {
- if (!exchangeContext.hasExchangeNode) {
+ private void adjustUpStream(PlanNode root, DistributedPlanGenerator.PlanContext context) {
+ if (!context.hasExchangeNode) {
return;
}
- adjustUpStreamHelper(root, exchangeContext, new HashMap<>());
+ adjustUpStreamHelper(root, context, new HashMap<>());
}
private void adjustUpStreamHelper(
PlanNode root,
- ExchangeNodeGenerator.PlanContext exchangeContext,
- Map<TRegionReplicaSet, IdentitySinkNode> regionNodemap) {
+ DistributedPlanGenerator.PlanContext context,
+ Map<TRegionReplicaSet, IdentitySinkNode> regionNodeMap) {
for (PlanNode child : root.getChildren()) {
- adjustUpStreamHelper(child, exchangeContext, regionNodemap);
+ adjustUpStreamHelper(child, context, regionNodeMap);
if (child instanceof ExchangeNode) {
ExchangeNode exchangeNode = (ExchangeNode) child;
+
IdentitySinkNode identitySinkNode =
- regionNodemap.computeIfAbsent(
- exchangeContext
- .getNodeDistribution(exchangeNode.getChild().getPlanNodeId())
- .getRegion(),
+ regionNodeMap.computeIfAbsent(
+ context.getNodeDistribution(exchangeNode.getChild().getPlanNodeId()).getRegion(),
k -> new IdentitySinkNode(mppQueryContext.getQueryId().genPlanNodeId()));
identitySinkNode.addChild(exchangeNode.getChild());
identitySinkNode.addDownStreamChannelLocation(
new DownStreamChannelLocation(exchangeNode.getPlanNodeId().toString()));
+
exchangeNode.setChild(identitySinkNode);
exchangeNode.setIndexOfUpstreamSinkHandle(identitySinkNode.getCurrentLastIndex());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java
index ff73c8c7715..9065ddaa4dc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelTypeProviderExtractor.java
@@ -20,6 +20,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.SimplePlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
@@ -106,6 +107,12 @@ public class TableModelTypeProviderExtractor {
return null;
}
+ @Override
+ public Void visitCollect(CollectNode node, Void context) {
+ node.getChildren().forEach(c -> c.accept(this, context));
+ return null;
+ }
+
@Override
public Void visitSort(SortNode node, Void context) {
node.getChild().accept(this, context);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CollectNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CollectNode.java
index 332277013ba..0dc0a4cb570 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CollectNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/CollectNode.java
@@ -26,6 +26,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -72,4 +74,22 @@ public class CollectNode extends MultiChildProcessNode {
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
return new CollectNode(planNodeId);
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode());
+ }
+
+ @Override
+ public String toString() {
+ return "CollectNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/FilterNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/FilterNode.java
index fc4f662f6e3..2258e35ffeb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/FilterNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/FilterNode.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChil
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import com.google.common.base.Objects;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -85,4 +87,23 @@ public class FilterNode extends SingleChildProcessNode {
public List<Symbol> getOutputSymbols() {
return child.getOutputSymbols();
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ FilterNode filterNode = (FilterNode) o;
+ return Objects.equal(predicate, filterNode.predicate);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), predicate);
+ }
+
+ @Override
+ public String toString() {
+ return "FilterNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/LimitNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/LimitNode.java
index f224ba9152f..03fa0b97ec0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/LimitNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/LimitNode.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChil
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
@@ -32,7 +33,7 @@ import java.util.Optional;
public class LimitNode extends SingleChildProcessNode {
private final long count;
- // TODO what's the meaning?
+ // TODO(beyyes) useless variable?
private final Optional<OrderingScheme> tiesResolvingScheme;
// private final boolean partial;
@@ -86,4 +87,23 @@ public class LimitNode extends SingleChildProcessNode {
public long getCount() {
return count;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ LimitNode limitNode = (LimitNode) o;
+ return Objects.equal(count, limitNode.count);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), count);
+ }
+
+ @Override
+ public String toString() {
+ return "LimitNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/MergeSortNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/MergeSortNode.java
index 4bce78f7c9e..a5d388d53ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/MergeSortNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/MergeSortNode.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChild
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
@@ -99,4 +100,24 @@ public class MergeSortNode extends MultiChildProcessNode {
public OrderingScheme getOrderingScheme() {
return orderingScheme;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ MergeSortNode sortNode = (MergeSortNode) o;
+ return Objects.equal(orderingScheme, sortNode.orderingScheme)
+ && Objects.equal(outputSymbols, sortNode.outputSymbols);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), orderingScheme, outputSymbols);
+ }
+
+ @Override
+ public String toString() {
+ return "MergeSortNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OffsetNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OffsetNode.java
index beb680b6842..f08c3ae2dfc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OffsetNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OffsetNode.java
@@ -21,6 +21,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
@@ -77,4 +78,23 @@ public class OffsetNode extends SingleChildProcessNode {
public long getCount() {
return count;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ OffsetNode offsetNode = (OffsetNode) o;
+ return Objects.equal(count, offsetNode.count);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), count);
+ }
+
+ @Override
+ public String toString() {
+ return "OffsetNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OutputNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OutputNode.java
index da8105d741e..87f15146a11 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OutputNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/OutputNode.java
@@ -21,6 +21,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -35,21 +36,22 @@ public class OutputNode extends SingleChildProcessNode {
private final List<String> columnNames;
// column name = symbol
- private final List<Symbol> outputs;
+ private final List<Symbol> outputSymbols;
- public OutputNode(PlanNodeId id, PlanNode child, List<String> columnNames, List<Symbol> outputs) {
+ public OutputNode(
+ PlanNodeId id, PlanNode child, List<String> columnNames, List<Symbol> outputSymbols) {
super(id, child);
this.id = id;
this.child = child;
this.columnNames = ImmutableList.copyOf(columnNames);
- this.outputs = ImmutableList.copyOf(outputs);
+ this.outputSymbols = ImmutableList.copyOf(outputSymbols);
}
- public OutputNode(PlanNodeId id, List<String> columnNames, List<Symbol> outputs) {
+ public OutputNode(PlanNodeId id, List<String> columnNames, List<Symbol> outputSymbols) {
super(id);
this.id = id;
this.columnNames = ImmutableList.copyOf(columnNames);
- this.outputs = ImmutableList.copyOf(outputs);
+ this.outputSymbols = ImmutableList.copyOf(outputSymbols);
}
@Override
@@ -59,7 +61,7 @@ public class OutputNode extends SingleChildProcessNode {
@Override
public PlanNode clone() {
- return new OutputNode(id, child, columnNames, outputs);
+ return new OutputNode(id, child, columnNames, outputSymbols);
}
@Override
@@ -72,8 +74,8 @@ public class OutputNode extends SingleChildProcessNode {
PlanNodeType.TABLE_OUTPUT_NODE.serialize(byteBuffer);
ReadWriteIOUtils.write(columnNames.size(), byteBuffer);
columnNames.forEach(columnName -> ReadWriteIOUtils.write(columnName, byteBuffer));
- ReadWriteIOUtils.write(outputs.size(), byteBuffer);
- outputs.forEach(symbol -> Symbol.serialize(symbol, byteBuffer));
+ ReadWriteIOUtils.write(outputSymbols.size(), byteBuffer);
+ outputSymbols.forEach(symbol -> Symbol.serialize(symbol, byteBuffer));
}
@Override
@@ -83,8 +85,8 @@ public class OutputNode extends SingleChildProcessNode {
for (String columnName : columnNames) {
ReadWriteIOUtils.write(columnName, stream);
}
- ReadWriteIOUtils.write(outputs.size(), stream);
- for (Symbol symbol : outputs) {
+ ReadWriteIOUtils.write(outputSymbols.size(), stream);
+ for (Symbol symbol : outputSymbols) {
Symbol.serialize(symbol, stream);
}
}
@@ -110,6 +112,25 @@ public class OutputNode extends SingleChildProcessNode {
@Override
public List<Symbol> getOutputSymbols() {
- return outputs;
+ return outputSymbols;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ OutputNode outputNode = (OutputNode) o;
+ return Objects.equal(outputSymbols, outputNode.outputSymbols);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), outputSymbols);
+ }
+
+ @Override
+ public String toString() {
+ return "OutputNode-" + this.getPlanNodeId();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ProjectNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ProjectNode.java
index b6124608079..140951a0e97 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ProjectNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/ProjectNode.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import com.google.common.base.Objects;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
@@ -93,4 +94,23 @@ public class ProjectNode extends SingleChildProcessNode {
public List<Symbol> getOutputSymbols() {
return assignments.getOutputs();
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ ProjectNode projectNode = (ProjectNode) o;
+ return Objects.equal(assignments, projectNode.assignments);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), assignments);
+ }
+
+ @Override
+ public String toString() {
+ return "ProjectNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/SortNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/SortNode.java
index c2dd7754d01..64ecf359045 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/SortNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/SortNode.java
@@ -16,19 +16,23 @@ package org.apache.iotdb.db.queryengine.plan.relational.planner.node;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
public class SortNode extends SingleChildProcessNode {
- private final OrderingScheme orderingScheme;
- private final boolean partial;
+ protected final OrderingScheme orderingScheme;
+ protected final boolean partial;
public SortNode(PlanNodeId id, PlanNode child, OrderingScheme scheme, boolean partial) {
super(id, child);
@@ -52,13 +56,24 @@ public class SortNode extends SingleChildProcessNode {
}
@Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {}
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TABLE_SORT_NODE.serialize(byteBuffer);
+ orderingScheme.serialize(byteBuffer);
+ ReadWriteIOUtils.write(partial, byteBuffer);
+ }
@Override
- protected void serializeAttributes(DataOutputStream stream) throws IOException {}
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TABLE_SORT_NODE.serialize(stream);
+ orderingScheme.serialize(stream);
+ ReadWriteIOUtils.write(partial, stream);
+ }
public static SortNode deserialize(ByteBuffer byteBuffer) {
- return null;
+ OrderingScheme orderingScheme = OrderingScheme.deserialize(byteBuffer);
+ boolean partial = ReadWriteIOUtils.readBool(byteBuffer);
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new SortNode(planNodeId, null, orderingScheme, partial);
}
@Override
@@ -69,4 +84,23 @@ public class SortNode extends SingleChildProcessNode {
public OrderingScheme getOrderingScheme() {
return orderingScheme;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ SortNode sortNode = (SortNode) o;
+ return Objects.equal(orderingScheme, sortNode.orderingScheme) && partial == sortNode.partial;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), orderingScheme, partial);
+ }
+
+ @Override
+ public String toString() {
+ return "SortNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/StreamSortNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/StreamSortNode.java
index f90a02df4ea..b0d419b0840 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/StreamSortNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/StreamSortNode.java
@@ -21,9 +21,13 @@ package org.apache.iotdb.db.queryengine.plan.relational.planner.node;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
+import com.google.common.base.Objects;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -52,12 +56,45 @@ public class StreamSortNode extends SortNode {
}
@Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {}
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TABLE_STREAM_SORT_NODE.serialize(byteBuffer);
+ orderingScheme.serialize(byteBuffer);
+ ReadWriteIOUtils.write(partial, byteBuffer);
+ ReadWriteIOUtils.write(streamCompareKeyEndIndex, byteBuffer);
+ }
@Override
- protected void serializeAttributes(DataOutputStream stream) throws IOException {}
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TABLE_STREAM_SORT_NODE.serialize(stream);
+ orderingScheme.serialize(stream);
+ ReadWriteIOUtils.write(partial, stream);
+ ReadWriteIOUtils.write(streamCompareKeyEndIndex, stream);
+ }
public static SortNode deserialize(ByteBuffer byteBuffer) {
- return null;
+ OrderingScheme orderingScheme = OrderingScheme.deserialize(byteBuffer);
+ boolean partial = ReadWriteIOUtils.readBool(byteBuffer);
+ int streamCompareKeyEndIndex = ReadWriteIOUtils.read(byteBuffer);
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new StreamSortNode(planNodeId, null, orderingScheme, partial, streamCompareKeyEndIndex);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ StreamSortNode streamSortNode = (StreamSortNode) o;
+ return Objects.equal(streamCompareKeyEndIndex, streamSortNode.streamCompareKeyEndIndex);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), streamCompareKeyEndIndex);
+ }
+
+ @Override
+ public String toString() {
+ return "StreamSortNode-" + this.getPlanNodeId();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
index d90f5af29a1..d8380bd892f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
@@ -51,7 +51,12 @@ public class TableScanNode extends SourceNode {
private Map<Symbol, ColumnSchema> assignments;
private List<DeviceEntry> deviceEntries;
- private Map<Symbol, Integer> idAndAttributeIndexMap;
+
+ // Indicates the respective index order of ID and Attribute columns in DeviceEntry.
+ // For example, for DeviceEntry `table1.tag1.tag2.attribute1.attribute2.s1.s2`, the content of
+ // `idAndAttributeIndexMap` will
+ // be `tag1: 0, tag2: 1, attribute1: 0, attribute2: 1`.
+ private final Map<Symbol, Integer> idAndAttributeIndexMap;
// The order to traverse the data.
// Currently, we only support TIMESTAMP_ASC and TIMESTAMP_DESC here.
@@ -85,11 +90,13 @@ public class TableScanNode extends SourceNode {
PlanNodeId id,
QualifiedObjectName qualifiedObjectName,
List<Symbol> outputSymbols,
- Map<Symbol, ColumnSchema> assignments) {
+ Map<Symbol, ColumnSchema> assignments,
+ Map<Symbol, Integer> idAndAttributeIndexMap) {
super(id);
this.qualifiedObjectName = qualifiedObjectName;
this.outputSymbols = outputSymbols;
this.assignments = assignments;
+ this.idAndAttributeIndexMap = idAndAttributeIndexMap;
}
public TableScanNode(
@@ -322,28 +329,6 @@ public class TableScanNode extends SourceNode {
@Override
public void close() throws Exception {}
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- TableScanNode that = (TableScanNode) o;
- return Objects.equals(qualifiedObjectName, that.qualifiedObjectName)
- && Objects.equals(outputSymbols, that.outputSymbols)
- && Objects.equals(regionReplicaSet, that.regionReplicaSet);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), qualifiedObjectName, outputSymbols, regionReplicaSet);
- }
-
public QualifiedObjectName getQualifiedObjectName() {
return this.qualifiedObjectName;
}
@@ -368,8 +353,8 @@ public class TableScanNode extends SourceNode {
return this.idAndAttributeIndexMap;
}
- public void setIdAndAttributeIndexMap(Map<Symbol, Integer> idAndAttributeIndexMap) {
- this.idAndAttributeIndexMap = idAndAttributeIndexMap;
+ public void setScanOrder(Ordering scanOrder) {
+ this.scanOrder = scanOrder;
}
public Ordering getScanOrder() {
@@ -427,4 +412,31 @@ public class TableScanNode extends SourceNode {
public void setTimeFilter(Filter timeFilter) {
this.timeFilter = timeFilter;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ TableScanNode that = (TableScanNode) o;
+ return Objects.equals(qualifiedObjectName, that.qualifiedObjectName)
+ && Objects.equals(outputSymbols, that.outputSymbols)
+ && Objects.equals(regionReplicaSet, that.regionReplicaSet);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), qualifiedObjectName, outputSymbols, regionReplicaSet);
+ }
+
+ @Override
+ public String toString() {
+ return "TableScanNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TopKNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TopKNode.java
index 6a78ac84e95..edade08c0b1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TopKNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TopKNode.java
@@ -21,14 +21,19 @@ package org.apache.iotdb.db.queryengine.plan.relational.planner.node;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import com.google.common.base.Objects;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
public class TopKNode extends MultiChildProcessNode {
@@ -70,13 +75,40 @@ public class TopKNode extends MultiChildProcessNode {
}
@Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {}
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TABLE_TOPK_NODE.serialize(byteBuffer);
+ orderingScheme.serialize(byteBuffer);
+ ReadWriteIOUtils.write(count, byteBuffer);
+ ReadWriteIOUtils.write(outputSymbols.size(), byteBuffer);
+ for (Symbol symbol : outputSymbols) {
+ Symbol.serialize(symbol, byteBuffer);
+ }
+ ReadWriteIOUtils.write(childrenDataInOrder, byteBuffer);
+ }
@Override
- protected void serializeAttributes(DataOutputStream stream) throws IOException {}
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TABLE_TOPK_NODE.serialize(stream);
+ orderingScheme.serialize(stream);
+ ReadWriteIOUtils.write(count, stream);
+ ReadWriteIOUtils.write(outputSymbols.size(), stream);
+ for (Symbol symbol : outputSymbols) {
+ Symbol.serialize(symbol, stream);
+ }
+ ReadWriteIOUtils.write(childrenDataInOrder, stream);
+ }
public static TopKNode deserialize(ByteBuffer byteBuffer) {
- return null;
+ OrderingScheme orderingScheme = OrderingScheme.deserialize(byteBuffer);
+ int count = ReadWriteIOUtils.readInt(byteBuffer);
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<Symbol> outputSymbols = new ArrayList<>(size);
+ while (size-- > 0) {
+ outputSymbols.add(Symbol.deserialize(byteBuffer));
+ }
+ boolean childrenDataInOrder = ReadWriteIOUtils.readBool(byteBuffer);
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new TopKNode(planNodeId, orderingScheme, count, outputSymbols, childrenDataInOrder);
}
@Override
@@ -95,4 +127,25 @@ public class TopKNode extends MultiChildProcessNode {
public boolean isChildrenDataInOrder() {
return childrenDataInOrder;
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ if (!super.equals(o)) return false;
+ TopKNode sortNode = (TopKNode) o;
+ return Objects.equal(orderingScheme, sortNode.orderingScheme)
+ && Objects.equal(outputSymbols, sortNode.outputSymbols)
+ && Objects.equal(count, sortNode.count);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), orderingScheme, outputSymbols, count);
+ }
+
+ @Override
+ public String toString() {
+ return "TopKNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PruneUnUsedColumns.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PruneUnUsedColumns.java
index 8b1ce55f7ae..c67a7a59b60 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PruneUnUsedColumns.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PruneUnUsedColumns.java
@@ -14,7 +14,6 @@
package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
-import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
@@ -50,7 +49,7 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.TIME;
* attr1, s1`, but the output columns of TableScanNode in `select s1 from table1` query can only be
* `s1`.
*/
-public class PruneUnUsedColumns implements RelationalPlanOptimizer {
+public class PruneUnUsedColumns implements TablePlanOptimizer {
@Override
public PlanNode optimize(
@@ -126,19 +125,6 @@ public class PruneUnUsedColumns implements RelationalPlanOptimizer {
}
node.setOutputSymbols(newOutputSymbols);
node.setAssignments(newAssignments);
-
- int IDIdx = 0, attributeIdx = 0;
- Map<Symbol, Integer> idAndAttributeIndexMap = new HashMap<>(node.getAssignments().size());
- for (Symbol symbol : node.getOutputSymbols()) {
- ColumnSchema columnSchema = node.getAssignments().get(symbol);
- if (TsTableColumnCategory.ID.equals(columnSchema.getColumnCategory())) {
- idAndAttributeIndexMap.put(symbol, IDIdx++);
- } else if (TsTableColumnCategory.ATTRIBUTE.equals(columnSchema.getColumnCategory())) {
- idAndAttributeIndexMap.put(symbol, attributeIdx++);
- }
- }
- node.setIdAndAttributeIndexMap(idAndAttributeIndexMap);
-
return node;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index e9fb6ead873..a3215c00953 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -43,12 +43,12 @@ import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.utils.Pair;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
+import static java.util.Objects.requireNonNull;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.ATTRIBUTE;
import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory.MEASUREMENT;
import static org.apache.iotdb.db.queryengine.plan.analyze.AnalyzeVisitor.getTimePartitionSlotList;
@@ -71,7 +71,7 @@ import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.GlobalT
* <p>Notice that, when aggregation, multi-table, join are introduced, this optimization rule
* need to be adapted.
*/
-public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
+public class PushPredicateIntoTableScan implements TablePlanOptimizer {
private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
@@ -82,10 +82,10 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
Metadata metadata,
SessionInfo sessionInfo,
MPPQueryContext queryContext) {
- return planNode.accept(new Rewriter(queryContext, analysis, metadata), new RewriterContext());
+ return planNode.accept(new Rewriter(queryContext, analysis, metadata), null);
}
- private static class Rewriter extends PlanVisitor<PlanNode, RewriterContext> {
+ private static class Rewriter extends PlanVisitor<PlanNode, Void> {
private final MPPQueryContext queryContext;
private final Analysis analysis;
private final Metadata metadata;
@@ -98,20 +98,20 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
}
@Override
- public PlanNode visitPlan(PlanNode node, RewriterContext context) {
+ public PlanNode visitPlan(PlanNode node, Void context) {
throw new IllegalArgumentException(
String.format("Unexpected plan node: %s in rule PushPredicateIntoTableScan", node));
}
@Override
- public PlanNode visitSingleChildProcess(SingleChildProcessNode node, RewriterContext context) {
+ public PlanNode visitSingleChildProcess(SingleChildProcessNode node, Void context) {
PlanNode rewrittenChild = node.getChild().accept(this, context);
node.setChild(rewrittenChild);
return node;
}
@Override
- public PlanNode visitMultiChildProcess(MultiChildProcessNode node, RewriterContext context) {
+ public PlanNode visitMultiChildProcess(MultiChildProcessNode node, Void context) {
List<PlanNode> rewrittenChildren = new ArrayList<>();
for (PlanNode child : node.getChildren()) {
rewrittenChildren.add(child.accept(this, context));
@@ -121,7 +121,7 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
}
@Override
- public PlanNode visitFilter(FilterNode node, RewriterContext context) {
+ public PlanNode visitFilter(FilterNode node, Void context) {
if (node.getPredicate() != null) {
@@ -149,14 +149,11 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
}
public PlanNode combineFilterAndScan(TableScanNode tableScanNode) {
- List<List<Expression>> splitPredicates = splitPredicate(tableScanNode);
-
- // exist indexed metadata expressions
- tableMetadataIndexScan(tableScanNode, splitPredicates.get(0));
+ SplitExpression splitExpression = splitPredicate(tableScanNode);
// exist expressions can push down to scan operator
- if (!splitPredicates.get(1).isEmpty()) {
- List<Expression> expressions = splitPredicates.get(1);
+ if (!splitExpression.getExpressionsCanPushDown().isEmpty()) {
+ List<Expression> expressions = splitExpression.getExpressionsCanPushDown();
Expression pushDownPredicate =
expressions.size() == 1
? expressions.get(0)
@@ -174,9 +171,12 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
tableScanNode.setPushDownPredicate(null);
}
+ // do index scan after expressionCanPushDown is processed
+ tableMetadataIndexScan(tableScanNode, splitExpression.getMetadataExpressions());
+
// exist expressions can not push down to scan operator
- if (!splitPredicates.get(2).isEmpty()) {
- List<Expression> expressions = splitPredicates.get(2);
+ if (!splitExpression.getExpressionsCannotPushDown().isEmpty()) {
+ List<Expression> expressions = splitExpression.getExpressionsCannotPushDown();
return new FilterNode(
queryContext.getQueryId().genPlanNodeId(),
tableScanNode,
@@ -188,12 +188,7 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
return tableScanNode;
}
- /**
- * splits predicate expression in table model into three parts, index 0 represents
- * metadataExpressions, index 1 represents expressionsCanPushDownToOperator, index 2 represents
- * expressionsCannotPushDownToOperator
- */
- private List<List<Expression>> splitPredicate(TableScanNode node) {
+ private SplitExpression splitPredicate(TableScanNode node) {
Set<String> idOrAttributeColumnNames =
node.getIdAndAttributeIndexMap().keySet().stream()
@@ -225,7 +220,7 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
}
}
- return Arrays.asList(
+ return new SplitExpression(
metadataExpressions, expressionsCanPushDown, expressionsCannotPushDown);
}
@@ -237,11 +232,12 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
expressionsCannotPushDown.add(predicate);
}
- return Arrays.asList(metadataExpressions, expressionsCanPushDown, expressionsCannotPushDown);
+ return new SplitExpression(
+ metadataExpressions, expressionsCanPushDown, expressionsCannotPushDown);
}
@Override
- public PlanNode visitTableScan(TableScanNode node, RewriterContext context) {
+ public PlanNode visitTableScan(TableScanNode node, Void context) {
tableMetadataIndexScan(node, Collections.emptyList());
return node;
}
@@ -263,11 +259,9 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
analysis.setEmptyDataSource(true);
} else {
Filter timeFilter =
- node.getTimePredicate().isPresent()
- ? node.getTimePredicate()
- .get()
- .accept(new ConvertPredicateToTimeFilterVisitor(), null)
- : null;
+ node.getTimePredicate()
+ .map(value -> value.accept(new ConvertPredicateToTimeFilterVisitor(), null))
+ .orElse(null);
node.setTimeFilter(timeFilter);
String treeModelDatabase = "root." + node.getQualifiedObjectName().getDatabaseName();
DataPartition dataPartition =
@@ -316,7 +310,7 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
}
}
- static boolean containsDiffFunction(Expression expression) {
+ public static boolean containsDiffFunction(Expression expression) {
if (expression instanceof FunctionCall
&& "diff".equalsIgnoreCase(((FunctionCall) expression).getName().toString())) {
return true;
@@ -333,5 +327,35 @@ public class PushPredicateIntoTableScan implements RelationalPlanOptimizer {
return false;
}
- private static class RewriterContext {}
+ private static class SplitExpression {
+ // indexed tag expressions, such as `tag1 = 'A'`
+ List<Expression> metadataExpressions;
+ // expressions can push down into TableScan, such as `time > 1 and s_1 = 1`
+ List<Expression> expressionsCanPushDown;
+ // expressions can not push down into TableScan, such as `s_1 is null`
+ List<Expression> expressionsCannotPushDown;
+
+ public SplitExpression(
+ List<Expression> metadataExpressions,
+ List<Expression> expressionsCanPushDown,
+ List<Expression> expressionsCannotPushDown) {
+ this.metadataExpressions = requireNonNull(metadataExpressions, "metadataExpressions is null");
+ this.expressionsCanPushDown =
+ requireNonNull(expressionsCanPushDown, "expressionsCanPushDown is null");
+ this.expressionsCannotPushDown =
+ requireNonNull(expressionsCannotPushDown, "expressionsCannotPushDown is null");
+ }
+
+ public List<Expression> getMetadataExpressions() {
+ return this.metadataExpressions;
+ }
+
+ public List<Expression> getExpressionsCanPushDown() {
+ return this.expressionsCanPushDown;
+ }
+
+ public List<Expression> getExpressionsCannotPushDown() {
+ return this.expressionsCannotPushDown;
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantIdentityProjections.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantIdentityProjections.java
index dcc1e68368e..7e89c5e1f62 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantIdentityProjections.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RemoveRedundantIdentityProjections.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNod
import java.util.Collections;
import java.util.List;
-public class RemoveRedundantIdentityProjections implements RelationalPlanOptimizer {
+public class RemoveRedundantIdentityProjections implements TablePlanOptimizer {
@Override
public PlanNode optimize(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SimplifyExpressions.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SimplifyExpressions.java
index 76c83a0aa0c..e6bf1ee22c2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SimplifyExpressions.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SimplifyExpressions.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.ExtractCommonPredicatesExpressionRewriter.extractCommonPredicates;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.ir.NormalizeOrExpressionRewriter.normalizeOrExpression;
-public class SimplifyExpressions implements RelationalPlanOptimizer {
+public class SimplifyExpressions implements TablePlanOptimizer {
@Override
public PlanNode optimize(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RelationalPlanOptimizer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TablePlanOptimizer.java
similarity index 96%
rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RelationalPlanOptimizer.java
rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TablePlanOptimizer.java
index 37611e6a35c..ce33f82adc3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/RelationalPlanOptimizer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TablePlanOptimizer.java
@@ -20,7 +20,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
-public interface RelationalPlanOptimizer {
+public interface TablePlanOptimizer {
PlanNode optimize(
PlanNode planNode,
Analysis analysis,
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
index 0369e96d111..4ff1d6c2b03 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector;
import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnHandle;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
@@ -37,7 +38,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectN
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableHandle;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema;
import org.apache.iotdb.db.queryengine.plan.relational.planner.LogicalPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.TableDistributionPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
@@ -45,6 +48,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LogicalExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Statement;
import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser;
@@ -60,6 +64,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector.NOOP;
import static org.apache.iotdb.db.queryengine.plan.statement.component.Ordering.ASC;
@@ -79,7 +85,7 @@ public class AnalyzerTest {
private static final NopAccessControl nopAccessControl = new NopAccessControl();
- QueryId queryId = new QueryId("tmp_query");
+ QueryId queryId = new QueryId("test_query");
SessionInfo sessionInfo =
new SessionInfo(
1L,
@@ -130,28 +136,6 @@ public class AnalyzerTest {
metadata.getTableSchema(Mockito.any(), eq(new QualifiedObjectName("testdb", "table1"))))
.thenReturn(Optional.of(tableSchema));
- // ResolvedFunction lLessThanI =
- // new ResolvedFunction(
- // new BoundSignature("l<i", BOOLEAN, Arrays.asList(INT64, INT32)),
- // new FunctionId("l<i"),
- // FunctionKind.SCALAR,
- // true);
- //
- // ResolvedFunction iAddi =
- // new ResolvedFunction(
- // new BoundSignature("l+i", INT64, Arrays.asList(INT32, INT32)),
- // new FunctionId("l+i"),
- // FunctionKind.SCALAR,
- // true);
- //
- // Mockito.when(
- // metadata.resolveOperator(eq(OperatorType.LESS_THAN), eq(Arrays.asList(INT64,
- // INT32))))
- // .thenReturn(lLessThanI);
- // Mockito.when(metadata.resolveOperator(eq(OperatorType.ADD), eq(Arrays.asList(INT32,
- // INT32))))
- // .thenReturn(iAddi);
-
Mockito.when(
metadata.getOperatorReturnType(
eq(OperatorType.LESS_THAN), eq(Arrays.asList(INT64, INT32))))
@@ -186,11 +170,23 @@ public class AnalyzerTest {
tableScanNode.getOutputColumnNames());
assertEquals(9, tableScanNode.getOutputSymbols().size());
assertEquals(9, tableScanNode.getAssignments().size());
- assertEquals(1, tableScanNode.getDeviceEntries().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
assertEquals(ASC, tableScanNode.getScanOrder());
distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(0)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof CollectNode);
}
@Test
@@ -212,12 +208,36 @@ public class AnalyzerTest {
Arrays.asList("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3"),
tableScanNode.getOutputColumnNames());
assertEquals(9, tableScanNode.getAssignments().size());
- assertEquals(1, tableScanNode.getDeviceEntries().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
- assertEquals("(\"time\" > 1)", tableScanNode.getTimePredicate().get().toString());
+ assertEquals(
+ "(\"time\" > 1)", tableScanNode.getTimePredicate().map(Expression::toString).orElse(null));
assertNull(tableScanNode.getPushDownPredicate());
assertEquals(ASC, tableScanNode.getScanOrder());
+
distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof CollectNode);
+ CollectNode collectNode = (CollectNode) outputNode.getChildren().get(0);
+ assertTrue(
+ collectNode.getChildren().get(0) instanceof ExchangeNode
+ && collectNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(collectNode.getChildren().get(1) instanceof TableScanNode);
+ TableScanNode tableScanNode = (TableScanNode) collectNode.getChildren().get(1);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
}
@Test
@@ -237,8 +257,44 @@ public class AnalyzerTest {
assertNotNull(tableScanNode.getPushDownPredicate());
assertEquals("(\"s1\" > 1)", tableScanNode.getPushDownPredicate().toString());
assertFalse(tableScanNode.getTimePredicate().isPresent());
+ assertTrue(
+ Stream.of(
+ Symbol.of("tag1"),
+ Symbol.of("tag2"),
+ Symbol.of("tag3"),
+ Symbol.of("attr1"),
+ Symbol.of("attr2"))
+ .allMatch(tableScanNode.getIdAndAttributeIndexMap()::containsKey));
assertEquals(
Arrays.asList("time", "tag1", "attr1", "s1", "s2"), tableScanNode.getOutputColumnNames());
+
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof CollectNode);
+ CollectNode collectNode = (CollectNode) outputNode.getChildren().get(0);
+ assertTrue(
+ collectNode.getChildren().get(0) instanceof ExchangeNode
+ && collectNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(collectNode.getChildren().get(1) instanceof TableScanNode);
+ tableScanNode = (TableScanNode) collectNode.getChildren().get(1);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0);
+ assertEquals("(\"s1\" > 1)", tableScanNode.getPushDownPredicate().toString());
}
@Test
@@ -317,6 +373,21 @@ public class AnalyzerTest {
Arrays.asList("time", "tag1", "attr1", "s1", "s2"), tableScanNode.getOutputColumnNames());
}
+ /*
+ * IdentitySinkNode-15
+ * └──OutputNode-3
+ * └──FilterNode-2
+ * └──CollectNode-10
+ * ├──ExchangeNode-11: [SourceAddress:192.0.12.1/test_query.2.0/13]
+ * ├──TableScanNode-8
+ * └──ExchangeNode-12: [SourceAddress:192.0.10.1/test_query.3.0/14]
+ *
+ * IdentitySinkNode-13
+ * └──TableScanNode-7
+ *
+ * IdentitySinkNode-14
+ * └──TableScanNode-9
+ */
@Test
public void singleTableWithFilterTest6() {
// value filter which can not be pushed down
@@ -333,6 +404,33 @@ public class AnalyzerTest {
tableScanNode = (TableScanNode) rootNode.getChildren().get(0).getChildren().get(0);
assertEquals(
Arrays.asList("time", "tag1", "attr1", "s1", "s2"), tableScanNode.getOutputColumnNames());
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof FilterNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof CollectNode);
+ CollectNode collectNode = (CollectNode) outputNode.getChildren().get(0).getChildren().get(0);
+ assertTrue(
+ collectNode.getChildren().get(0) instanceof ExchangeNode
+ && collectNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(collectNode.getChildren().get(1) instanceof TableScanNode);
+ tableScanNode = (TableScanNode) collectNode.getChildren().get(1);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0);
sql = "SELECT tag1, attr1, s2 FROM table1 where diff(s1) + 1 > 1";
actualAnalysis = analyzeSQL(sql, metadata);
@@ -363,7 +461,7 @@ public class AnalyzerTest {
@Test
public void singleTableProjectTest() {
// 1. project without filter
- sql = "SELECT tag1, attr1, s1 FROM table1";
+ sql = "SELECT time, tag1, attr1, s1 FROM table1";
actualAnalysis = analyzeSQL(sql, metadata);
assertNotNull(actualAnalysis);
assertEquals(1, actualAnalysis.getTables().size());
@@ -433,7 +531,7 @@ public class AnalyzerTest {
tableScanNode = (TableScanNode) rootNode.getChildren().get(0);
assertEquals(
Arrays.asList("time", "tag2", "attr2", "s2"), tableScanNode.getOutputColumnNames());
- assertEquals(2, tableScanNode.getIdAndAttributeIndexMap().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
}
@Test
@@ -554,6 +652,37 @@ public class AnalyzerTest {
assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof FilterNode);
FilterNode filterNode = (FilterNode) rootNode.getChildren().get(0).getChildren().get(0);
assertEquals("(DIFF(\"s2\") > 0)", filterNode.getPredicate().toString());
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof FilterNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof CollectNode);
+ CollectNode collectNode =
+ (CollectNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(
+ collectNode.getChildren().get(0) instanceof ExchangeNode
+ && collectNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(collectNode.getChildren().get(1) instanceof TableScanNode);
+ tableScanNode = (TableScanNode) collectNode.getChildren().get(1);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0);
// 2. diff with time filter, tag filter and measurement filter
sql = "SELECT s1 FROM table1 WHERE DIFF(s2) > 0 and time > 5 and tag1 = 'A' and s1 = 1";
@@ -630,18 +759,6 @@ public class AnalyzerTest {
assertEquals(3, offsetNode.getCount());
}
- @Test
- public void sortTest() {
- // when TableScan locates multi regions, use default MergeSortNode
- sql = "SELECT * FROM table1 ";
- context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
- actualAnalysis = analyzeSQL(sql, metadata);
- logicalQueryPlan =
- new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
- .plan(actualAnalysis);
- rootNode = logicalQueryPlan.getRootNode();
- }
-
public static Analysis analyzeSQL(String sql, Metadata metadata) {
try {
SqlParser sqlParser = new SqlParser();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java
new file mode 100644
index 00000000000..90e9d71f46b
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java
@@ -0,0 +1,166 @@
+/*
+ * 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.queryengine.plan.relational.analyzer;
+
+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.partition.DataPartition;
+import org.apache.iotdb.commons.partition.SchemaPartition;
+import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class MockTableModelDataPartition {
+
+ private static final SeriesPartitionExecutor EXECUTOR =
+ SeriesPartitionExecutor.getSeriesPartitionExecutor(
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+
+ private static final String DB_NAME = "testdb";
+
+ static final String DEVICE_1 = "table1.beijing.A1.ZZ";
+ static final String DEVICE_2 = "table1.beijing.A2.XX";
+ static final String DEVICE_3 = "table1.shanghai.A3.YY";
+ static final String DEVICE_4 = "table1.shanghai.B3.YY";
+ static final String DEVICE_5 = "table1.shenzhen.B2.ZZ";
+ static final String DEVICE_6 = "table1.shenzhen.B1.XX";
+
+ static final List<String> DEVICE_1_ATTRIBUTES = Arrays.asList("high", "big");
+ static final List<String> DEVICE_2_ATTRIBUTES = Arrays.asList("high", "small");
+ static final List<String> DEVICE_3_ATTRIBUTES = Arrays.asList("low", "small");
+ static final List<String> DEVICE_4_ATTRIBUTES = Arrays.asList("low", "big");
+ static final List<String> DEVICE_5_ATTRIBUTES = Arrays.asList("mid", "big");
+ static final List<String> DEVICE_6_ATTRIBUTES = Arrays.asList("mid", "small");
+
+ private static final TRegionReplicaSet DATA_REGION_GROUP_1 = genDataRegionGroup(10, 1, 2);
+ private static final TRegionReplicaSet DATA_REGION_GROUP_2 = genDataRegionGroup(11, 3, 2);
+ private static final TRegionReplicaSet DATA_REGION_GROUP_3 = genDataRegionGroup(12, 2, 1);
+ private static final TRegionReplicaSet DATA_REGION_GROUP_4 = genDataRegionGroup(13, 1, 3);
+
+ /*
+ * DataPartition:
+ *
+ * device1(startTime:0): DataRegionGroup_1,
+ * device2(startTime:0): DataRegionGroup_1,
+ * device3(startTime:0): DataRegionGroup_2,
+ * device4(startTime:0): DataRegionGroup_2,
+ * device5(startTime:0): DataRegionGroup_2,
+ * device5(startTime:100): DataRegionGroup_3,
+ * device6(startTime:0): DataRegionGroup_2,
+ * device6(startTime:100): DataRegionGroup_3,
+ */
+ public static DataPartition constructDataPartition() {
+ DataPartition dataPartition =
+ new DataPartition(
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
+ IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
+
+ Map<String, Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>>
+ dbPartitionMap = new HashMap<>();
+ Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>> devicePartitionMap =
+ new HashMap<>();
+
+ List<TRegionReplicaSet> regionGroup1 = Collections.singletonList(DATA_REGION_GROUP_1);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> dataRegionMap1 =
+ Collections.singletonMap(new TTimePartitionSlot(0L), regionGroup1);
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_1), dataRegionMap1);
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_2), dataRegionMap1);
+
+ List<TRegionReplicaSet> regionGroup2 = Collections.singletonList(DATA_REGION_GROUP_2);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> dataRegionMap2 =
+ Collections.singletonMap(new TTimePartitionSlot(0L), regionGroup2);
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_3), dataRegionMap2);
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_4), dataRegionMap2);
+
+ List<TRegionReplicaSet> regionGroup3 = Collections.singletonList(DATA_REGION_GROUP_2);
+ List<TRegionReplicaSet> regionGroup4 = Collections.singletonList(DATA_REGION_GROUP_3);
+ Map<TTimePartitionSlot, List<TRegionReplicaSet>> dataRegionMap3 =
+ ImmutableMap.<TTimePartitionSlot, List<TRegionReplicaSet>>builder()
+ .put(new TTimePartitionSlot(0L), regionGroup3)
+ .put(new TTimePartitionSlot(100L), regionGroup4)
+ .build();
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_5), dataRegionMap3);
+ devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_6), dataRegionMap3);
+
+ dbPartitionMap.put(DB_NAME, devicePartitionMap);
+ dataPartition.setDataPartitionMap(dbPartitionMap);
+
+ return dataPartition;
+ }
+
+ public static SchemaPartition constructSchemaPartition() {
+ 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(DEVICE_1), schemaRegion1);
+ schemaRegionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_2), schemaRegion2);
+ schemaRegionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_3), schemaRegion2);
+ schemaPartitionMap.put(DB_NAME, schemaRegionMap);
+ schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
+
+ return schemaPartition;
+ }
+
+ private static TRegionReplicaSet genDataRegionGroup(
+ int regionGroupId, int dataNodeId1, int dataNodeId2) {
+ return new TRegionReplicaSet(
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, regionGroupId),
+ Arrays.asList(
+ genDataNodeLocation(dataNodeId1, String.format("192.0.%s.1", regionGroupId)),
+ genDataNodeLocation(dataNodeId2, String.format("192.0.%s.2", regionGroupId))));
+ }
+
+ private static TDataNodeLocation genDataNodeLocation(int dataNodeId, String ip) {
+ return new TDataNodeLocation()
+ .setDataNodeId(dataNodeId)
+ .setClientRpcEndPoint(new TEndPoint(ip, 9000))
+ .setMPPDataExchangeEndPoint(new TEndPoint(ip, 9001))
+ .setInternalEndPoint(new TEndPoint(ip, 9002));
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTablePartition.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTablePartition.java
deleted file mode 100644
index ae24f4a219e..00000000000
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTablePartition.java
+++ /dev/null
@@ -1,174 +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.queryengine.plan.relational.analyzer;
-
-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.partition.DataPartition;
-import org.apache.iotdb.commons.partition.SchemaPartition;
-import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class MockTablePartition {
-
- private static final SeriesPartitionExecutor EXECUTOR =
- SeriesPartitionExecutor.getSeriesPartitionExecutor(
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
- IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
-
- private static final String DB_NAME = "root.testdb";
-
- private static final String device1 = "root.testdb.d1";
- private static final String device2 = "root.testdb.d22";
- private static final String device3 = "root.testdb.d333";
- private static final String device4 = "root.testdb.d4444";
- private static final String device5 = "root.testdb.d55555";
- private static final String device6 = "root.testdb.d666666";
-
- public static DataPartition constructDataPartition() {
- 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")));
-
- 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);
-
- List<TRegionReplicaSet> d5DataRegions = new ArrayList<>();
- d5DataRegions.add(dataRegion4);
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d5DataRegionMap = new HashMap<>();
- d5DataRegionMap.put(new TTimePartitionSlot(), d5DataRegions);
-
- List<TRegionReplicaSet> d6DataRegions = new ArrayList<>();
- d6DataRegions.add(dataRegion1);
- d6DataRegions.add(dataRegion2);
- Map<TTimePartitionSlot, List<TRegionReplicaSet>> d6DataRegionMap = new HashMap<>();
- d6DataRegionMap.put(new TTimePartitionSlot(), d6DataRegions);
-
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device1), d1DataRegionMap);
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device2), d2DataRegionMap);
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device3), d3DataRegionMap);
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device4), d4DataRegionMap);
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device5), d5DataRegionMap);
- sgPartitionMap.put(EXECUTOR.getSeriesPartitionSlot(device6), d6DataRegionMap);
-
- dataPartitionMap.put(DB_NAME, sgPartitionMap);
- dataPartition.setDataPartitionMap(dataPartitionMap);
-
- return dataPartition;
- }
-
- public static SchemaPartition constructSchemaPartition() {
- 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(DB_NAME, schemaRegionMap);
- schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
-
- return schemaPartition;
- }
-
- private static TDataNodeLocation genDataNodeLocation(int dataNodeId, String ip) {
- return new TDataNodeLocation()
- .setDataNodeId(dataNodeId)
- .setClientRpcEndPoint(new TEndPoint(ip, 9000))
- .setMPPDataExchangeEndPoint(new TEndPoint(ip, 9001))
- .setInternalEndPoint(new TEndPoint(ip, 9002));
- }
-}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java
new file mode 100644
index 00000000000..b5a71999a3e
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java
@@ -0,0 +1,2251 @@
+/*
+ * 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.queryengine.plan.relational.analyzer;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.LogicalPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.TableDistributionPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OffsetNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PruneUnUsedColumns;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PushPredicateIntoTableScan;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.RemoveRedundantIdentityProjections;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.SimplifyExpressions;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.TablePlanOptimizer;
+
+import org.junit.Test;
+
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.AnalyzerTest.analyzeSQL;
+import static org.apache.iotdb.db.queryengine.plan.statement.component.Ordering.ASC;
+import static org.apache.iotdb.db.queryengine.plan.statement.component.Ordering.DESC;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class SortTest {
+
+ QueryId queryId = new QueryId("test_query");
+ SessionInfo sessionInfo =
+ new SessionInfo(
+ 1L,
+ "iotdb-user",
+ ZoneId.systemDefault(),
+ IoTDBConstant.ClientVersion.V_1_0,
+ "db",
+ IClientSession.SqlDialect.TABLE);
+ Metadata metadata = new TestMatadata();
+ String sql;
+ Analysis actualAnalysis;
+ MPPQueryContext context;
+ LogicalPlanner logicalPlanner;
+ LogicalQueryPlan logicalQueryPlan;
+ PlanNode rootNode;
+ TableDistributionPlanner distributionPlanner;
+ DistributedQueryPlan distributedQueryPlan;
+ TableScanNode tableScanNode;
+ List<TablePlanOptimizer> planOptimizerList =
+ Arrays.asList(
+ new SimplifyExpressions(),
+ new PruneUnUsedColumns(),
+ new PushPredicateIntoTableScan(),
+ new RemoveRedundantIdentityProjections());
+
+ /*
+ * order by time, others, some_ids
+ *
+ * IdentitySinkNode-33
+ * └──OutputNode-8
+ * └──LimitNode-7
+ * └──OffsetNode-6
+ * └──MergeSortNode-25
+ * ├──ExchangeNode-29: [SourceAddress:192.0.12.1/test_query.2.0/31]
+ * ├──SortNode-27
+ * │ └──ProjectNode-23
+ * │ └──ProjectNode-20
+ * │ └──FilterNode-17
+ * │ └──TableScanNode-14
+ * └──ExchangeNode-30: [SourceAddress:192.0.10.1/test_query.3.0/32]
+ *
+ * IdentitySinkNode-31
+ * └──SortNode-26
+ * └──ProjectNode-22
+ * └──ProjectNode-19
+ * └──FilterNode-16
+ * └──TableScanNode-13
+ *
+ * IdentitySinkNode-31
+ * └──SortNode-26
+ * └──ProjectNode-22
+ * └──ProjectNode-19
+ * └──FilterNode-16
+ * └──TableScanNode-13
+ */
+ @Test
+ public void timeOthersSomeIDColumnSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by time desc, s1+s2 asc, tag2 asc, tag1 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ // TODO(beyyes) if these two ProjectNode can be merged into one?
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ // TODO(beyyes) fix the case prune unUsed Columns when the child of SortNode is ProjectNode
+ // assertEquals(
+ // Arrays.asList("time", "tag1", "tag2", "attr1", "s1", "s2"),
+ // tableScanNode.getOutputColumnNames());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("time", "(s1 + s2)", "tag2", "tag1"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // TODO(beyyes) add only one device entry optimization and verifies
+ }
+
+ // order by time, others, all_ids
+ @Test
+ public void timeOthersAllIDColumnSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by time desc, s1+s2 asc, tag2 asc, tag3 desc, tag1 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("time", "(s1 + s2)", "tag2", "tag3", "tag1"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+ }
+
+ // order by time, some_ids, others
+ @Test
+ public void timeSomeIDColumnOthersSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by time desc, tag2 asc, tag3 desc, s1+s2 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("time", "tag2", "tag3", "(s1 + s2)"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+ }
+
+ // order by time, all_ids, others
+ @Test
+ public void timeAllIDColumnOthersSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by time desc, tag2 asc, tag3 desc, tag1 asc, s1+s2 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("time", "tag2", "tag3", "tag1", "(s1 + s2)"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+ }
+
+ // order by some_ids, time, others
+ @Test
+ public void someIDColumnTimeOthersSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by tag2 desc, tag3 asc, time desc, s1+s2 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("tag2", "tag3", "time", "(s1 + s2)"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shenzhen.B1.XX",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B2.ZZ", "table1.shenzhen.B1.XX"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+ }
+
+ // order by all_ids, time, others
+ @Test
+ public void allIDColumnTimeOthersSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by tag2 desc, tag1 desc, tag3 asc, time desc, s1+s2 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("tag2", "tag1", "tag3", "time", "(s1 + s2)"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shenzhen.B1.XX",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B2.ZZ", "table1.shenzhen.B1.XX"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(DESC, tableScanNode.getScanOrder());
+ }
+
+ // order by some_ids, others, time
+ @Test
+ public void someIDColumnOthersTimeSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by tag2 desc, tag1 desc, s1+s2 desc, time desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("tag2", "tag1", "(s1 + s2)", "time"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shenzhen.B1.XX",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B2.ZZ", "table1.shenzhen.B1.XX"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+
+ // order by all_ids, others, time
+ @Test
+ public void allIDColumnOthersTimeSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by tag2 desc, tag1 desc, tag3 asc, s1+s2 desc, time desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("tag2", "tag1", "tag3", "(s1 + s2)", "time"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shenzhen.B1.XX",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B2.ZZ", "table1.shenzhen.B1.XX"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+
+ // order by others, some_ids, time
+ @Test
+ public void othersSomeIDColumnTimeSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by s1+s2 desc, tag2 desc, tag1 desc, time asc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("(s1 + s2)", "tag2", "tag1", "time"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+
+ // order by others, all_ids, time
+ @Test
+ public void othersAllIDColumnTimeSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by s1+s2 desc, tag2 desc, tag1 desc, tag3 desc, time asc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("(s1 + s2)", "tag2", "tag1", "tag3", "time"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+
+ // order by others, time, some_ids
+ @Test
+ public void othersTimeSomeIDColumnSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by s1+s2 desc, time desc, tag2 desc, tag1 desc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("(s1 + s2)", "time", "tag2", "tag1"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+
+ // order by others, time, all_ids
+ @Test
+ public void othersTimeAllIDColumnSortTest() {
+ sql =
+ "SELECT time, tag3, tag1, cast(s2 as double), s2+s3, attr1 FROM table1 "
+ + "where s1>1 and s1+s3>0 and cast(s1 as double)>1.0 order by s1+s2 desc, time desc, tag2 desc, tag1 desc, tag3 asc offset 5 limit 10";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, planOptimizerList, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ // OutputNode - LimitNode - OffsetNode - SortNode - ProjectNode - ProjectNode - FilterNode -
+ // TableScanNode
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ SortNode sortNode =
+ (SortNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ // OutputNode - LimitNode - OffsetNode - MergeSortNode - SortNode - ProjectNode - ProjectNode -
+ // FilterNode -
+ // TableScanNode
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof LimitNode);
+ assertTrue(outputNode.getChildren().get(0).getChildren().get(0) instanceof OffsetNode);
+ assertTrue(
+ outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof MergeSortNode);
+ MergeSortNode mergeSortNode =
+ (MergeSortNode) outputNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals(
+ Arrays.asList("(s1 + s2)", "time", "tag2", "tag1", "tag3"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof SortNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ sortNode = (SortNode) mergeSortNode.getChildren().get(1);
+ assertTrue(sortNode.getChildren().get(0) instanceof ProjectNode);
+ assertTrue(sortNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ assertTrue(
+ sortNode.getChildren().get(0).getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ sortNode
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ",
+ "table1.shanghai.A3.YY"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+
+ // IdentitySinkNode - SortNode - ProjectNode - ProjectNode - FilterNode - TableScanNode
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree() instanceof IdentitySinkNode);
+ assertTrue(
+ distributedQueryPlan.getFragments().get(1).getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof ProjectNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof FilterNode);
+ assertTrue(
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode)
+ distributedQueryPlan
+ .getFragments()
+ .get(1)
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0);
+ assertEquals(2, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList("table1.shenzhen.B1.XX", "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
+ assertEquals(ASC, tableScanNode.getScanOrder());
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
index ae4d4a7315d..66eeecd7510 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
@@ -47,12 +47,23 @@ import org.apache.tsfile.read.common.type.StringType;
import org.apache.tsfile.read.common.type.Type;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_1;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_1_ATTRIBUTES;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_2;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_2_ATTRIBUTES;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_3;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_3_ATTRIBUTES;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_4;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_4_ATTRIBUTES;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_5;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_5_ATTRIBUTES;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_6;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.MockTableModelDataPartition.DEVICE_6_ATTRIBUTES;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl.getFunctionType;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl.isOneNumericType;
import static org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl.isTwoNumericType;
@@ -195,10 +206,13 @@ public class TestMatadata implements Metadata {
QualifiedObjectName tableName,
List<Expression> expressionList,
List<String> attributeColumns) {
- return Collections.singletonList(
- new DeviceEntry(
- new StringArrayDeviceID("root.testdb", "table1", "t1", "t2", "t3"),
- Arrays.asList("a1", "a2")));
+ return Arrays.asList(
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES),
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_6.split("\\.")), DEVICE_6_ATTRIBUTES),
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES),
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES),
+ new DeviceEntry(new StringArrayDeviceID(DEVICE_2.split("\\.")), DEVICE_2_ATTRIBUTES));
}
@Override
@@ -241,9 +255,11 @@ public class TestMatadata implements Metadata {
return DATA_PARTITION;
}
- private static final DataPartition DATA_PARTITION = MockTablePartition.constructDataPartition();
+ private static final DataPartition DATA_PARTITION =
+ MockTableModelDataPartition.constructDataPartition();
+
private static final SchemaPartition SCHEMA_PARTITION =
- MockTablePartition.constructSchemaPartition();
+ MockTableModelDataPartition.constructSchemaPartition();
private static IPartitionFetcher getFakePartitionFetcher() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
index edeed746f13..1fe12378834 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/partition/DataPartition.java
@@ -140,9 +140,20 @@ public class DataPartition extends Partition {
* <p>The device id shall be [table, seg1, ....]
*/
public List<TRegionReplicaSet> getDataRegionReplicaSetWithTimeFilter(
- String database, IDeviceID deviceID, Filter timeFilter) {
- // TODO implement this interface, @Potato
- throw new UnsupportedOperationException();
+ String database, IDeviceID deviceId, Filter timeFilter) {
+ // TODO perfect this interface, @Potato
+ TSeriesPartitionSlot seriesPartitionSlot = calculateDeviceGroupId(deviceId);
+ if (!dataPartitionMap.containsKey(database)
+ || !dataPartitionMap.get(database).containsKey(seriesPartitionSlot)) {
+ return Collections.singletonList(NOT_ASSIGNED);
+ }
+ return dataPartitionMap.get(database).get(seriesPartitionSlot).entrySet().stream()
+ .filter(
+ entry ->
+ TimePartitionUtils.satisfyPartitionStartTime(timeFilter, entry.getKey().startTime))
+ .flatMap(entry -> entry.getValue().stream())
+ .distinct()
+ .collect(toList());
}
public List<TRegionReplicaSet> getDataRegionReplicaSet(