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/01/10 08:29:08 UTC
(iotdb) branch master updated: [IOTDB-6286] Add new query optimize rule: Use TopKNode to replace SortNode when order by expression + limit N
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6a478f67012 [IOTDB-6286] Add new query optimize rule: Use TopKNode to replace SortNode when order by expression + limit N
6a478f67012 is described below
commit 6a478f670120af04e5e05139a4f1945833e9f27b
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Jan 10 16:29:03 2024 +0800
[IOTDB-6286] Add new query optimize rule: Use TopKNode to replace SortNode when order by expression + limit N
---
.../IoTDBOrderByLimitOffsetAlignByDeviceIT.java | 37 +
.../db/queryengine/plan/analyze/Analysis.java | 10 +-
.../OrderByExpressionWithLimitChangeToTopK.java | 309 +++++++
.../plan/planner/LogicalPlanBuilder.java | 100 ++-
.../plan/planner/LogicalPlanVisitor.java | 21 +-
.../plan/planner/TemplatedLogicalPlan.java | 22 +-
.../planner/distribution/DistributionPlanner.java | 11 +-
.../planner/distribution/ExchangeNodeAdder.java | 2 +-
.../plan/planner/distribution/SourceRewriter.java | 4 +-
.../plan/planner/plan/node/process/FillNode.java | 5 +
.../plan/planner/plan/node/process/OffsetNode.java | 5 +
.../plan/planner/plan/node/process/SortNode.java | 5 +
.../plan/planner/plan/node/process/TopKNode.java | 6 +-
.../plan/statement/crud/QueryStatement.java | 2 +-
.../AlignByDeviceOrderByLimitOffsetTest.java | 988 ++++++++++++++++++---
.../AlignByTimeOrderByLimitOffsetTest.java | 316 +++++++
16 files changed, 1644 insertions(+), 199 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java
index 1b88c38b585..670d35c9ca0 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/alignbydevice/IoTDBOrderByLimitOffsetAlignByDeviceIT.java
@@ -85,6 +85,33 @@ public class IoTDBOrderByLimitOffsetAlignByDeviceIT {
retArray);
}
+ @Test
+ public void aggregationWithHavingTest() {
+ // when aggregation with having, can only use MergeSortNode but not use TopKNode
+ String[] expectedHeader = new String[] {"Time,Device,sum(s1)"};
+ String[] retArray = new String[] {"3,root.db.d2,222.0,", "3,root.db.d3,333.0,"};
+ resultSetEqualTest(
+ "select sum(s1) from root.db.** group by ((1,5],1ms) having(sum(s1)>111) order by time limit 2 align by device;",
+ expectedHeader,
+ retArray);
+ }
+
+ @Test
+ public void fillTest() {
+ String[] expectedHeader = new String[] {"Time,Device,s1,s2"};
+ String[] retArray =
+ new String[] {
+ "1,root.fill.d1,1,null,",
+ "1,root.fill.d2,2,null,",
+ "1,root.fill.d3,3,null,",
+ "2,root.fill.d1,12,11.0,",
+ };
+ resultSetEqualTest(
+ "select * from root.fill.** order by time fill(linear) limit 4 align by device;",
+ expectedHeader,
+ retArray);
+ }
+
private static final String[] SQL_LIST =
new String[] {
"CREATE DATABASE root.db;",
@@ -94,6 +121,16 @@ public class IoTDBOrderByLimitOffsetAlignByDeviceIT {
"INSERT INTO root.db.d1(timestamp,s1) VALUES(1, 1), (2, 11), (3, 111);",
"INSERT INTO root.db.d2(timestamp,s1) VALUES(1, 2), (2, 22), (3, 222);",
"INSERT INTO root.db.d3(timestamp,s1) VALUES(1, 3), (2, 33), (3, 333);",
+ "CREATE DATABASE root.fill;",
+ "CREATE TIMESERIES root.fill.d1.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "CREATE TIMESERIES root.fill.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE;",
+ "CREATE TIMESERIES root.fill.d2.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "CREATE TIMESERIES root.fill.d2.s2 WITH DATATYPE=FLOAT, ENCODING=RLE;",
+ "CREATE TIMESERIES root.fill.d3.s1 WITH DATATYPE=INT32, ENCODING=RLE;",
+ "CREATE TIMESERIES root.fill.d3.s2 WITH DATATYPE=FLOAT, ENCODING=RLE;",
+ "INSERT INTO root.fill.d1(timestamp,s1,s2) VALUES(1, 1, null), (2, null, 11), (3, 111, 111.1);",
+ "INSERT INTO root.fill.d2(timestamp,s1,s2) VALUES(1, 2, null), (2, 22, 22.2), (3, 222, null);",
+ "INSERT INTO root.fill.d3(timestamp,s1,s2) VALUES(1, 3, null), (2, 33, null), (3, 333, 333.3);",
};
protected static void insertData3() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
index 86e58e33f17..4e9c5ec6b15 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/Analysis.java
@@ -195,7 +195,7 @@ public class Analysis {
// one.
private Set<Expression> orderByExpressions;
- private boolean hasSort = false;
+ private boolean hasSortNode = false;
// parameter of `FILL` clause
private FillDescriptor fillDescriptor;
@@ -728,12 +728,12 @@ public class Analysis {
this.deviceToOrderByExpressions = deviceToOrderByExpressions;
}
- public void setHasSort(boolean hasSort) {
- this.hasSort = hasSort;
+ public void setHasSortNode(boolean hasSortNode) {
+ this.hasSortNode = hasSortNode;
}
- public boolean isHasSort() {
- return hasSort;
+ public boolean isHasSortNode() {
+ return hasSortNode;
}
public Map<String, List<SortItem>> getDeviceToSortItems() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/OrderByExpressionWithLimitChangeToTopK.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/OrderByExpressionWithLimitChangeToTopK.java
new file mode 100644
index 00000000000..9bdf5effc74
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/OrderByExpressionWithLimitChangeToTopK.java
@@ -0,0 +1,309 @@
+/*
+ * 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.optimization;
+
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+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.process.FillNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.OffsetNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
+import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement;
+
+import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_VALUE_USE_TOP_K;
+import static org.apache.iotdb.db.queryengine.plan.statement.component.FillPolicy.LINEAR;
+
+/**
+ * Replace `SortNode`+`LimitNode` to `TopKNode` and replace `MergeSortNode`+`LimitNode` to
+ * `TopKNode` in below cases. Notice that, `TransformNode` below is only used to transform the
+ * projection columns, an example query used `TransformNode`: `select s1 from root.** order by s2`.
+ *
+ * <p>The cases which can use this optimize rule.
+ * <li>`LimitNode + SortNode` ==> `TopKNode`.
+ * <li>`LimitNode + MergeSortNode` ==> `TopKNode`.
+ * <li>`LimitNode + OffsetNode + SortNode` ==> `LimitNode + OffsetNode + TopKNode(where topValue =
+ * limitValue+offsetValue)`.
+ * <li>`LimitNode + OffsetNode + MergeSortNode` ==> `LimitNode + OffsetNode + TopKNode(where
+ * topValue = limitValue+offsetValue)`.
+ * <li>`LimitNode + TransformNode/FillNode + SortNode` ==> `TransformNode/FillNode + TopKNode`.
+ * <li>`LimitNode + TransformNode/FillNode + MergeSortNode` ==> `TransformNode/FillNode + TopKNode`.
+ * <li>`LimitNode + OffsetNode + TransformNode/FillNode + SortNode` ==> `LimitNode + OffsetNode +
+ * TransformNode/FillNode + TopKNode(where topValue = limitValue+offsetValue)`.
+ * <li>`LimitNode + OffsetNode + TransformNode/FillNode + MergeSortNode` ==> `LimitNode + OffsetNode
+ * + TransformNode/FillNode + TopKNode(where topValue = limitValue+offsetValue)`.
+ */
+public class OrderByExpressionWithLimitChangeToTopK implements PlanOptimizer {
+
+ @Override
+ public PlanNode optimize(PlanNode plan, Analysis analysis, MPPQueryContext context) {
+ if (analysis.getStatement().getType() != StatementType.QUERY) {
+ return plan;
+ }
+
+ QueryStatement queryStatement = (QueryStatement) analysis.getStatement();
+ if (queryStatement.isLastQuery() || !queryStatement.hasLimit()) {
+ return plan;
+ }
+
+ // when align by time, only order by expression can use this optimize rule
+ if (!queryStatement.isAlignByDevice() && !queryStatement.hasOrderByExpression()) {
+ return plan;
+ }
+
+ // align by device,
+ // when use TopKNode (because MergeSortNode with LimitNode has been replaced to TopKNode),
+ // or order_based_on_device (because it does not need TopKNode),
+ // will not use this optimize rule
+ if (queryStatement.isAlignByDevice()
+ && (analysis.isUseTopKNode()
+ || !queryStatement.hasOrderBy()
+ || queryStatement.isOrderByBasedOnDevice())) {
+ return plan;
+ }
+
+ return plan.accept(new Rewriter(), new RewriterContext(context));
+ }
+
+ private static class Rewriter extends PlanVisitor<PlanNode, RewriterContext> {
+
+ @Override
+ public PlanNode visitPlan(PlanNode node, RewriterContext context) {
+ for (PlanNode child : node.getChildren()) {
+ context.setParent(node);
+ child.accept(this, context);
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitLimit(LimitNode limitNode, RewriterContext rewriterContext) {
+ PlanNode parent = rewriterContext.getParent();
+
+ if (limitNode.getChild() instanceof OffsetNode) {
+ rewriterContext.setParent(limitNode);
+ limitNode.getChild().accept(this, rewriterContext);
+ }
+
+ if (limitNode.getLimit() > LIMIT_VALUE_USE_TOP_K) {
+ return limitNode;
+ }
+
+ if (limitNode.getChild() instanceof SortNode) {
+ return rewriterContext.returnSortNode(limitNode, parent);
+ } else if (limitNode.getChild() instanceof MergeSortNode) {
+ return rewriterContext.returnMergeSortNode(limitNode, parent);
+ } else if (limitNode.getChild() instanceof TransformNode
+ && !(limitNode.getChild() instanceof FilterNode)) {
+ return rewriterContext.returnTransformNodeFillNode(limitNode, parent);
+ } else if (limitNode.getChild() instanceof FillNode
+ && !LINEAR.equals(
+ ((FillNode) limitNode.getChild()).getFillDescriptor().getFillPolicy())) {
+ return rewriterContext.returnTransformNodeFillNode(limitNode, parent);
+ }
+
+ return limitNode;
+ }
+
+ @Override
+ public PlanNode visitOffset(OffsetNode offsetNode, RewriterContext rewriterContext) {
+ PlanNode parent = rewriterContext.getParent();
+
+ if (!(parent instanceof LimitNode)) {
+ return offsetNode;
+ }
+
+ LimitNode limitNode = (LimitNode) parent;
+ if (limitNode.getLimit() + offsetNode.getOffset() > LIMIT_VALUE_USE_TOP_K) {
+ return offsetNode;
+ }
+
+ if (offsetNode.getChild() instanceof SortNode) {
+ rewriterContext.processSortNode(offsetNode, parent);
+ } else if (offsetNode.getChild() instanceof MergeSortNode) {
+ rewriterContext.processMergeSortNode(offsetNode, parent);
+ } else if (offsetNode.getChild() instanceof TransformNode
+ && !(offsetNode.getChild() instanceof FilterNode)) {
+ rewriterContext.processTransformNodeFillNode(offsetNode, parent);
+ } else if (offsetNode.getChild() instanceof FillNode
+ && !LINEAR.equals(
+ ((FillNode) offsetNode.getChild()).getFillDescriptor().getFillPolicy())) {
+ rewriterContext.processTransformNodeFillNode(offsetNode, parent);
+ }
+
+ return offsetNode;
+ }
+ }
+
+ static class RewriterContext {
+ private PlanNode parent;
+
+ private final MPPQueryContext mppQueryContext;
+
+ public RewriterContext(MPPQueryContext mppQueryContext) {
+ this.mppQueryContext = mppQueryContext;
+ }
+
+ public PlanNode getParent() {
+ return parent;
+ }
+
+ public void setParent(PlanNode parent) {
+ this.parent = parent;
+ }
+
+ public MPPQueryContext getMppQueryContext() {
+ return this.mppQueryContext;
+ }
+
+ private PlanNode returnSortNode(LimitNode limitNode, PlanNode parent) {
+ SortNode sortNode = (SortNode) limitNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) limitNode.getLimit(),
+ sortNode.getOrderByParameter(),
+ sortNode.getOutputColumnNames());
+ topKNode.setChildren(sortNode.getChildren());
+
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(topKNode);
+ return parent;
+ } else {
+ return topKNode;
+ }
+ }
+
+ private PlanNode returnMergeSortNode(LimitNode limitNode, PlanNode parent) {
+ MergeSortNode mergeSortNode = (MergeSortNode) limitNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) limitNode.getLimit(),
+ mergeSortNode.getMergeOrderParameter(),
+ mergeSortNode.getOutputColumnNames());
+ topKNode.setChildren(mergeSortNode.getChildren());
+
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(topKNode);
+ return parent;
+ } else {
+ return topKNode;
+ }
+ }
+
+ private PlanNode returnTransformNodeFillNode(LimitNode limitNode, PlanNode parent) {
+ SingleChildProcessNode singleNode = (SingleChildProcessNode) limitNode.getChild();
+ if (singleNode.getChild() instanceof SortNode) {
+ SortNode sortNode = (SortNode) singleNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) limitNode.getLimit(),
+ sortNode.getOrderByParameter(),
+ sortNode.getOutputColumnNames());
+ topKNode.setChildren(sortNode.getChildren());
+ singleNode.setChild(topKNode);
+
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(singleNode);
+ return parent;
+ } else {
+ return singleNode;
+ }
+ } else if (singleNode.getChild() instanceof MergeSortNode) {
+ MergeSortNode mergeSortNode = (MergeSortNode) singleNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) limitNode.getLimit(),
+ mergeSortNode.getMergeOrderParameter(),
+ mergeSortNode.getOutputColumnNames());
+ topKNode.setChildren(mergeSortNode.getChildren());
+ singleNode.setChild(topKNode);
+
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(singleNode);
+ return parent;
+ } else {
+ return singleNode;
+ }
+ }
+
+ return limitNode;
+ }
+
+ private void processSortNode(OffsetNode offsetNode, PlanNode parent) {
+ SortNode sortNode = (SortNode) offsetNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) ((int) ((LimitNode) parent).getLimit() + offsetNode.getOffset()),
+ sortNode.getOrderByParameter(),
+ sortNode.getOutputColumnNames());
+ topKNode.setChildren(sortNode.getChildren());
+ offsetNode.setChild(topKNode);
+ }
+
+ private void processMergeSortNode(OffsetNode offsetNode, PlanNode parent) {
+ MergeSortNode sortNode = (MergeSortNode) offsetNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) ((int) ((LimitNode) parent).getLimit() + offsetNode.getOffset()),
+ sortNode.getMergeOrderParameter(),
+ sortNode.getOutputColumnNames());
+ topKNode.setChildren(sortNode.getChildren());
+ offsetNode.setChild(topKNode);
+ }
+
+ private void processTransformNodeFillNode(OffsetNode offsetNode, PlanNode parent) {
+ SingleChildProcessNode singleNode = (SingleChildProcessNode) offsetNode.getChild();
+ if (singleNode.getChild() instanceof SortNode) {
+ SortNode sortNode = (SortNode) singleNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) ((int) ((LimitNode) parent).getLimit() + offsetNode.getOffset()),
+ sortNode.getOrderByParameter(),
+ sortNode.getOutputColumnNames());
+ topKNode.setChildren(sortNode.getChildren());
+ singleNode.setChild(topKNode);
+ } else if (singleNode.getChild() instanceof MergeSortNode) {
+ MergeSortNode mergeSortNode = (MergeSortNode) singleNode.getChild();
+ TopKNode topKNode =
+ new TopKNode(
+ getMppQueryContext().getQueryId().genPlanNodeId(),
+ (int) ((int) ((LimitNode) parent).getLimit() + offsetNode.getOffset()),
+ mergeSortNode.getMergeOrderParameter(),
+ mergeSortNode.getOutputColumnNames());
+ topKNode.setChildren(mergeSortNode.getChildren());
+ singleNode.setChild(topKNode);
+ }
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
index 0ca934c3282..038149b2b9a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanBuilder.java
@@ -129,7 +129,8 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDC
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.DEVICE;
import static org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant.ENDTIME;
import static org.apache.iotdb.db.queryengine.plan.analyze.ExpressionTypeAnalyzer.analyzeExpression;
-import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE;
+import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_VALUE_USE_TOP_K;
+import static org.apache.iotdb.db.queryengine.plan.statement.component.FillPolicy.LINEAR;
import static org.apache.iotdb.db.utils.constant.SqlConstant.COUNT_TIME;
import static org.apache.iotdb.db.utils.constant.SqlConstant.LAST_VALUE;
import static org.apache.iotdb.db.utils.constant.SqlConstant.MAX_TIME;
@@ -739,7 +740,8 @@ public class LogicalPlanBuilder {
Set<Expression> deviceViewOutputExpressions,
Map<String, List<Integer>> deviceToMeasurementIndexesMap,
Set<Expression> selectExpression,
- QueryStatement queryStatement) {
+ QueryStatement queryStatement,
+ Analysis analysis) {
List<String> outputColumnNames =
deviceViewOutputExpressions.stream()
.map(Expression::getExpressionString)
@@ -764,16 +766,7 @@ public class LogicalPlanBuilder {
? queryStatement.getRowOffset() + queryStatement.getRowLimit()
: queryStatement.getRowLimit();
- // 1. LIMIT and LIMIT_VALUE is smaller than 1000000,
- // 2. `order by based on time` or `order by based on expression`,
- // 3. no aggregation,
- // when satisfy all above requirements use ToKNode.
- if (!queryStatement.isAggregationQuery()
- && queryStatement.hasLimit()
- && queryStatement.getOrderByComponent() != null
- && !queryStatement.isOrderByBasedOnDevice()
- && limitValue <= LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE) {
-
+ if (canUseTopKNode(queryStatement, limitValue)) {
TopKNode topKNode =
new TopKNode(
context.getQueryId().genPlanNodeId(),
@@ -784,7 +777,7 @@ public class LogicalPlanBuilder {
// if value filter exists, need add a LIMIT-NODE as the child node of TopKNode
long valueFilterLimit = queryStatement.hasWhere() ? limitValue : -1;
- // order by based on time, use TopKNode + SingleDeviceViewNode
+ // only order by based on time, use TopKNode + SingleDeviceViewNode
if (queryStatement.isOrderByBasedOnTime() && !queryStatement.hasOrderByExpression()) {
addSingleDeviceViewNodes(
topKNode,
@@ -793,7 +786,7 @@ public class LogicalPlanBuilder {
deviceToMeasurementIndexesMap,
valueFilterLimit);
} else {
- // order by based on expression, use TopKNode + DeviceViewNode
+ // has order by expression, use TopKNode + DeviceViewNode
topKNode.addChild(
addDeviceViewNode(
orderByParameter,
@@ -803,35 +796,22 @@ public class LogicalPlanBuilder {
valueFilterLimit));
}
+ analysis.setUseTopKNode();
this.root = topKNode;
- }
- // 1. `order by based on time` + `no order by expression`,
- // 2. no LIMIT or LIMIT_VALUE is larger than 1000000,
- // when satisfy all above requirements use MergeSortNode.
- else if (queryStatement.isOrderByBasedOnTime() && !queryStatement.hasOrderByExpression()) {
- if (deviceNameToSourceNodesMap.size() == 1) {
- // only one device, use DeviceViewNode, no need MergeSortNode
- this.root =
- addDeviceViewNode(
- orderByParameter,
- outputColumnNames,
- deviceToMeasurementIndexesMap,
- deviceNameToSourceNodesMap,
- -1);
- } else {
- // otherwise use MergeSortNode + SingleDeviceViewNode
- MergeSortNode mergeSortNode =
- new MergeSortNode(
- context.getQueryId().genPlanNodeId(), orderByParameter, outputColumnNames);
- addSingleDeviceViewNodes(
- mergeSortNode,
- deviceNameToSourceNodesMap,
- outputColumnNames,
- deviceToMeasurementIndexesMap,
- -1);
- this.root = mergeSortNode;
- }
+ } else if (canUseMergeSortNode(queryStatement, deviceNameToSourceNodesMap.size())) {
+ // otherwise use MergeSortNode + SingleDeviceViewNode
+ MergeSortNode mergeSortNode =
+ new MergeSortNode(
+ context.getQueryId().genPlanNodeId(), orderByParameter, outputColumnNames);
+ addSingleDeviceViewNodes(
+ mergeSortNode,
+ deviceNameToSourceNodesMap,
+ outputColumnNames,
+ deviceToMeasurementIndexesMap,
+ -1);
+ this.root = mergeSortNode;
} else {
+ // order by based on device, use DeviceViewNode
this.root =
addDeviceViewNode(
orderByParameter,
@@ -859,6 +839,31 @@ public class LogicalPlanBuilder {
return this;
}
+ private boolean canUseTopKNode(QueryStatement queryStatement, long limitValue) {
+ // 1. has LIMIT and LIMIT_VALUE is smaller than 1000000.
+ // 2. `order by based on time` or `order by based on expression`.
+ // 3. no aggregation or has aggregation but no having.
+ // 4. no fill or has fill but not LINEAR fill.
+ // when satisfy all cases above will use ToKNode.
+ return queryStatement.hasLimit()
+ && limitValue <= LIMIT_VALUE_USE_TOP_K
+ && queryStatement.hasOrderBy()
+ && !queryStatement.isOrderByBasedOnDevice()
+ && (!queryStatement.isAggregationQuery()
+ || (queryStatement.isAggregationQuery() && !queryStatement.hasHaving()))
+ && (!queryStatement.hasFill()
+ || !LINEAR.equals(queryStatement.getFillComponent().getFillPolicy()));
+ }
+
+ private boolean canUseMergeSortNode(QueryStatement queryStatement, int deviceSize) {
+ // 1. `order by based on time` + `no order by expression`.
+ // 2. deviceSize is larger than 1.
+ // when satisfy all above cases use MergeSortNode.
+ return queryStatement.isOrderByBasedOnTime()
+ && !queryStatement.hasOrderByExpression()
+ && deviceSize > 1;
+ }
+
private void addSingleDeviceViewNodes(
MultiChildProcessNode parent,
Map<String, PlanNode> deviceNameToSourceNodesMap,
@@ -1189,10 +1194,6 @@ public class LogicalPlanBuilder {
return this;
}
- if (this.getRoot() instanceof TopKNode) {
- return this;
- }
-
this.root = new LimitNode(context.getQueryId().genPlanNodeId(), this.getRoot(), rowLimit);
return this;
}
@@ -1557,10 +1558,7 @@ public class LogicalPlanBuilder {
return this;
}
- public LogicalPlanBuilder planOrderBy(
- QueryStatement queryStatement,
- Set<Expression> orderByExpressions,
- Set<Expression> selectExpression) {
+ public LogicalPlanBuilder planOrderBy(QueryStatement queryStatement, Analysis analysis) {
// only the order by clause having expression needs a sortNode
if (!queryStatement.hasOrderByExpression()) {
return this;
@@ -1571,13 +1569,16 @@ public class LogicalPlanBuilder {
return this;
}
+ Set<Expression> orderByExpressions = analysis.getOrderByExpressions();
updateTypeProvider(orderByExpressions);
OrderByParameter orderByParameter = new OrderByParameter(queryStatement.getSortItemList());
if (orderByParameter.isEmpty()) {
return this;
}
+
this.root = new SortNode(context.getQueryId().genPlanNodeId(), root, orderByParameter);
+ Set<Expression> selectExpression = analysis.getSelectExpressions();
if (root.getOutputColumnNames().size() != selectExpression.size()) {
this.root =
new TransformNode(
@@ -1588,6 +1589,7 @@ public class LogicalPlanBuilder {
queryStatement.getSelectComponent().getZoneId(),
queryStatement.getResultTimeOrder());
}
+
return this;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
index 0b131538942..24324831b31 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
@@ -47,7 +47,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.vie
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedDeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedInsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWriteSchemaNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -168,7 +167,7 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
: null,
analysis.getDeviceViewInputIndexesMap().get(deviceName),
context));
- // sortOperator push down
+ // order by device, expression, push down sortOperator
if (queryStatement.needPushDownSort()) {
subPlanBuilder =
subPlanBuilder.planOrderBy(
@@ -185,11 +184,8 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
analysis.getDeviceViewOutputExpressions(),
analysis.getDeviceViewInputIndexesMap(),
analysis.getSelectExpressions(),
- queryStatement);
-
- if (planBuilder.getRoot() instanceof TopKNode) {
- analysis.setUseTopKNode();
- }
+ queryStatement,
+ analysis);
} else {
planBuilder =
planBuilder.withNewRoot(
@@ -216,17 +212,18 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
}
if (!queryStatement.needPushDownSort()) {
- planBuilder =
- planBuilder.planOrderBy(
- queryStatement, analysis.getOrderByExpressions(), analysis.getSelectExpressions());
+ planBuilder = planBuilder.planOrderBy(queryStatement, analysis);
}
// other upstream node
planBuilder =
planBuilder
.planFill(analysis.getFillDescriptor(), queryStatement.getResultTimeOrder())
- .planOffset(queryStatement.getRowOffset())
- .planLimit(queryStatement.getRowLimit());
+ .planOffset(queryStatement.getRowOffset());
+
+ if (!analysis.isUseTopKNode() || queryStatement.hasOffset()) {
+ planBuilder = planBuilder.planLimit(queryStatement.getRowLimit());
+ }
// plan select into
if (queryStatement.isAlignByDevice()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TemplatedLogicalPlan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TemplatedLogicalPlan.java
index 05ef92dbd60..2819b592ad8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TemplatedLogicalPlan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TemplatedLogicalPlan.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.db.queryengine.plan.analyze.TemplatedInfo;
import org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -158,7 +157,7 @@ public class TemplatedLogicalPlan {
new TemplatedLogicalPlanBuilder(analysis, context, measurementList, schemaList)
.withNewRoot(rootNode);
- // sortOperator push down
+ // order by device, expression, push down sortOperator
if (queryStatement.needPushDownSort()) {
subPlanBuilder =
subPlanBuilder.planOrderBy(
@@ -175,24 +174,21 @@ public class TemplatedLogicalPlan {
analysis.getDeviceViewOutputExpressions(),
analysis.getDeviceViewInputIndexesMap(),
analysis.getSelectExpressions(),
- queryStatement);
-
- if (planBuilder.getRoot() instanceof TopKNode) {
- analysis.setUseTopKNode();
- }
+ queryStatement,
+ analysis);
if (!queryStatement.needPushDownSort()) {
- planBuilder =
- planBuilder.planOrderBy(
- queryStatement, analysis.getOrderByExpressions(), analysis.getSelectExpressions());
+ planBuilder = planBuilder.planOrderBy(queryStatement, analysis);
}
- // other upstream node
planBuilder =
planBuilder
.planFill(analysis.getFillDescriptor(), queryStatement.getResultTimeOrder())
- .planOffset(queryStatement.getRowOffset())
- .planLimit(queryStatement.getRowLimit());
+ .planOffset(queryStatement.getRowOffset());
+
+ if (!analysis.isUseTopKNode() || queryStatement.hasOffset()) {
+ planBuilder = planBuilder.planLimit(queryStatement.getRowLimit());
+ }
return planBuilder.getRoot();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
index a0ba04cba5f..aa7d14608c7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.optimization.ColumnInjectionPushDown;
import org.apache.iotdb.db.queryengine.plan.optimization.LimitOffsetPushDown;
+import org.apache.iotdb.db.queryengine.plan.optimization.OrderByExpressionWithLimitChangeToTopK;
import org.apache.iotdb.db.queryengine.plan.optimization.PlanOptimizer;
import org.apache.iotdb.db.queryengine.plan.planner.IFragmentParallelPlaner;
import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
@@ -53,7 +54,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE;
+import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode.LIMIT_VALUE_USE_TOP_K;
public class DistributionPlanner {
private final Analysis analysis;
@@ -67,7 +68,11 @@ public class DistributionPlanner {
this.logicalPlan = logicalPlan;
this.context = logicalPlan.getContext();
- this.optimizers = Arrays.asList(new LimitOffsetPushDown(), new ColumnInjectionPushDown());
+ this.optimizers =
+ Arrays.asList(
+ new LimitOffsetPushDown(),
+ new ColumnInjectionPushDown(),
+ new OrderByExpressionWithLimitChangeToTopK());
}
public PlanNode rewriteSource() {
@@ -166,7 +171,7 @@ public class DistributionPlanner {
// TopKNode will use IdentityNode but not ShuffleSinkNode
if (queryStatement.hasLimit()
&& !queryStatement.isOrderByBasedOnDevice()
- && queryStatement.getRowLimit() <= LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE) {
+ && queryStatement.getRowLimit() <= LIMIT_VALUE_USE_TOP_K) {
return false;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
index d6111a947f7..06043bd2e41 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/ExchangeNodeAdder.java
@@ -406,7 +406,7 @@ public class ExchangeNodeAdder extends PlanVisitor<PlanNode, NodeGroupContext> {
return newNode;
}
- // optimize `order by time limit N align by device` query,
+ // optimize `order by time|expression limit N align by device` query,
// to ensure that the number of ExchangeNode equals to DataRegionNum but not equals to DeviceNum
if (node instanceof TopKNode) {
return processTopNode(node, visitedChildren, context, newNode, dataRegion);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
index 5e0403bb0a5..751258628f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
@@ -212,7 +212,7 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
deviceViewNodeList.add(regionDeviceViewNode);
}
- if (deviceViewNodeList.size() == 1 || analysis.isHasSort() || analysis.isUseTopKNode()) {
+ if (deviceViewNodeList.size() == 1 || analysis.isHasSortNode() || analysis.isUseTopKNode()) {
return deviceViewNodeList;
}
@@ -288,7 +288,7 @@ public class SourceRewriter extends SimplePlanNodeRewriter<DistributionPlanConte
@Override
public List<PlanNode> visitSort(SortNode node, DistributionPlanContext context) {
- analysis.setHasSort(true);
+ analysis.setHasSortNode(true);
List<PlanNode> children = rewrite(node.getChild(), context);
if (children.size() == 1) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/FillNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/FillNode.java
index 5fff9dd5ec2..17eae2c61eb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/FillNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/FillNode.java
@@ -109,6 +109,11 @@ public class FillNode extends SingleChildProcessNode {
return Objects.hash(super.hashCode(), fillDescriptor, scanOrder);
}
+ @Override
+ public String toString() {
+ return "FillNode-" + this.getPlanNodeId();
+ }
+
public FillDescriptor getFillDescriptor() {
return fillDescriptor;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/OffsetNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/OffsetNode.java
index 4eadbc219b4..8854a754d04 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/OffsetNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/OffsetNode.java
@@ -85,6 +85,11 @@ public class OffsetNode extends SingleChildProcessNode {
return offset;
}
+ @Override
+ public String toString() {
+ return "OffsetNode-" + this.getPlanNodeId();
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/SortNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/SortNode.java
index 416ab666ca2..d5bae27895c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/SortNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/SortNode.java
@@ -95,4 +95,9 @@ public class SortNode extends SingleChildProcessNode {
public int hashCode() {
return Objects.hashCode(super.hashCode(), orderByParameter);
}
+
+ @Override
+ public String toString() {
+ return "SortNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java
index 8b5845d0467..b32bbc2bd9b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/TopKNode.java
@@ -36,10 +36,8 @@ import java.util.Objects;
/** TopNode is optimized for `order by time|expression limit N align by device` query. */
public class TopKNode extends MultiChildProcessNode {
- // when LIMIT value in `order by time|expression LIMIT N align by device` query is less this
- // value,
- // use TopKNode, otherwise use MergeSortNode
- public static final int LIMIT_USE_TOP_K_FOR_ALIGN_BY_DEVICE = 1000000;
+ // when LIMIT value is less this value, can use TopKNode
+ public static final int LIMIT_VALUE_USE_TOP_K = 1000000;
private final int topValue;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
index 556a4dad374..80d3dec1a11 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/QueryStatement.java
@@ -418,7 +418,7 @@ public class QueryStatement extends AuthorityInformationStatement {
return orderByComponent.getDeviceOrder();
}
- // push down only support raw data query currently
+ // align by device + order by device, expression
public boolean needPushDownSort() {
return !isAggregationQuery() && hasOrderByExpression() && isOrderByBasedOnDevice();
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByDeviceOrderByLimitOffsetTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByDeviceOrderByLimitOffsetTest.java
index 88206dc97d2..750a86c4b3a 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByDeviceOrderByLimitOffsetTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByDeviceOrderByLimitOffsetTest.java
@@ -26,15 +26,21 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
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.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.DeviceViewNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FilterNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.HorizontallyConcatNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.MergeSortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SingleDeviceViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.sink.IdentitySinkNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.source.SeriesScanNode;
import org.junit.Test;
@@ -46,23 +52,287 @@ public class AlignByDeviceOrderByLimitOffsetTest {
private static final long LIMIT_VALUE = 10;
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ String sql;
+ Analysis analysis;
+ PlanNode logicalPlanNode;
+ DistributionPlanner planner;
+ DistributedQueryPlan plan;
+ PlanNode firstFiRoot;
+ PlanNode firstFiTopNode;
+ PlanNode mergeSortNode;
+
+ /*
+ * IdentitySinkNode-27
+ * └──LimitNode-22
+ * └──MergeSort-21
+ * ├──DeviceView-12
+ * │ └──FullOuterTimeJoinNode-11
+ * │ ├──SeriesScanNode-9:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-10:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-23: [SourceAddress:192.0.3.1/test.2.0/25]
+ * └──ExchangeNode-24: [SourceAddress:192.0.2.1/test.3.0/26]
+ *
+ * IdentitySinkNode-25
+ * └──DeviceView-16
+ * └──FullOuterTimeJoinNode-15
+ * ├──SeriesScanNode-13:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-14:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-26
+ * └──DeviceView-20
+ * └──FullOuterTimeJoinNode-19
+ * ├──SeriesScanNode-17:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-18:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ */
+ @Test
+ public void orderByDeviceTest1() {
+ // no order by
+ sql = "select * from root.sg.d1, root.sg.d22 LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertTrue(firstFiRoot.getChildren().get(0) instanceof LimitNode);
+ mergeSortNode = ((LimitNode) firstFiRoot.getChildren().get(0)).getChild();
+ assertTrue(mergeSortNode instanceof MergeSortNode);
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(
+ mergeSortNode.getChildren().get(0).getChildren().get(0) instanceof FullOuterTimeJoinNode);
+
+ // order by device, time
+ sql =
+ "select * from root.sg.d1, root.sg.d22 order by device asc, time desc LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertTrue(firstFiRoot.getChildren().get(0) instanceof LimitNode);
+ mergeSortNode = ((LimitNode) firstFiRoot.getChildren().get(0)).getChild();
+ assertTrue(mergeSortNode instanceof MergeSortNode);
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(
+ mergeSortNode.getChildren().get(0).getChildren().get(0) instanceof FullOuterTimeJoinNode);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ }
+
+ /*
+ * IdentitySinkNode-32
+ * └──LimitNode-27
+ * └──MergeSort-26
+ * ├──DeviceView-15
+ * │ └──SortNode-14
+ * │ └──FullOuterTimeJoinNode-13
+ * │ ├──SeriesScanNode-11:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-12:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-28: [SourceAddress:192.0.3.1/test.2.0/30]
+ * └──ExchangeNode-29: [SourceAddress:192.0.2.1/test.3.0/31]
+ *
+ * IdentitySinkNode-30
+ * └──DeviceView-20
+ * └──SortNode-19
+ * └──FullOuterTimeJoinNode-18
+ * ├──SeriesScanNode-16:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-17:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-31
+ * └──DeviceView-25
+ * └──SortNode-24
+ * └──FullOuterTimeJoinNode-23
+ * ├──SeriesScanNode-21:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-22:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ */
@Test
- public void orderByTimeNoValueFilterTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ public void orderByDeviceTest2() {
+ // order by device, expression
+ sql =
+ "select * from root.sg.d1, root.sg.d22 order by device asc, s1 desc LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertTrue(firstFiRoot.getChildren().get(0) instanceof LimitNode);
+ mergeSortNode = ((LimitNode) firstFiRoot.getChildren().get(0)).getChild();
+ assertTrue(mergeSortNode instanceof MergeSortNode);
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(mergeSortNode.getChildren().get(0).getChildren().get(0) instanceof SortNode);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ assertScanNodeLimitValue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE);
+ }
+
+ /*
+ * IdentitySinkNode-38
+ * └──LimitNode-33
+ * └──TransformNode-12
+ * └──MergeSort-32
+ * ├──DeviceView-19
+ * │ └──SortNode-18
+ * │ └──FilterNode-17
+ * │ └──FullOuterTimeJoinNode-16
+ * │ ├──SeriesScanNode-14:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-15:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-34: [SourceAddress:192.0.3.1/test.2.0/36]
+ * └──ExchangeNode-35: [SourceAddress:192.0.2.1/test.3.0/37]
+ *
+ * IdentitySinkNode-36
+ * └──DeviceView-25
+ * └──SortNode-24
+ * └──FilterNode-23
+ * └──FullOuterTimeJoinNode-22
+ * ├──SeriesScanNode-20:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-21:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-37
+ * └──DeviceView-31
+ * └──SortNode-30
+ * └──FilterNode-29
+ * └──FullOuterTimeJoinNode-28
+ * ├──SeriesScanNode-26:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-27:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ */
+ @Test
+ public void orderByDeviceTest3() {
+ // order by device, expression; with value filter
+ sql =
+ "select s1 from root.sg.d1, root.sg.d22 WHERE s2=1 order by device asc, s2 desc LIMIT 5 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertTrue(firstFiRoot.getChildren().get(0) instanceof LimitNode);
+ PlanNode transformNode = ((LimitNode) firstFiRoot.getChildren().get(0)).getChild();
+ assertTrue(transformNode instanceof TransformNode);
+ assertTrue(transformNode.getChildren().get(0) instanceof MergeSortNode);
+ assertTrue(transformNode.getChildren().get(0).getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(
+ transformNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SortNode);
+ }
- // 1. order by time, no value filter
- String sql =
+ /*
+ * IdentitySinkNode-27
+ * └──LimitNode-22
+ * └──FilterNode-12
+ * └──DeviceView-14
+ * ├──AggregationNode-5
+ * │ └──FilterNode-4
+ * │ └──FullOuterTimeJoinNode-3
+ * │ ├──SeriesScanNode-15:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ ├──SeriesScanNode-17:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-23: [SourceAddress:192.0.2.1/test.2.0/25]
+ * └──ExchangeNode-24: [SourceAddress:192.0.3.1/test.3.0/26]
+ *
+ * IdentitySinkNode-25
+ * └──FullOuterTimeJoinNode-19
+ * ├──SeriesScanNode-16:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-18:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-26
+ * └──AggregationNode-10
+ * └──FilterNode-9
+ * └──FullOuterTimeJoinNode-8
+ * ├──SeriesScanNode-20:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-21:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ */
+ @Test
+ public void orderByDeviceTest4() {
+ // aggregation + order by device, expression; with value filter
+ sql =
+ "select count(s1) from root.sg.d1, root.sg.d22 WHERE s2=1 having(count(s1)>1) LIMIT 5 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ assertTrue(firstFiRoot instanceof IdentitySinkNode);
+ assertTrue(firstFiRoot.getChildren().get(0) instanceof LimitNode);
+ PlanNode filterNode = ((LimitNode) firstFiRoot.getChildren().get(0)).getChild();
+ assertTrue(filterNode instanceof FilterNode);
+ assertTrue(filterNode.getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(filterNode.getChildren().get(0).getChildren().get(0) instanceof AggregationNode);
+ assertTrue(
+ filterNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FilterNode);
+ PlanNode thirdFiRoot = plan.getInstances().get(2).getFragment().getPlanNodeTree();
+ assertTrue(thirdFiRoot instanceof IdentitySinkNode);
+ assertTrue(thirdFiRoot.getChildren().get(0) instanceof AggregationNode);
+ assertTrue(thirdFiRoot.getChildren().get(0).getChildren().get(0) instanceof FilterNode);
+ }
+
+ /*
+ * IdentitySinkNode-44
+ * └──TopK-10
+ * ├──TopK-34
+ * │ ├──SingleDeviceView-17
+ * │ │ └──FullOuterTimeJoinNode-16
+ * │ │ ├──SeriesScanNode-14:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──SeriesScanNode-15:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SingleDeviceView-29
+ * │ └──FullOuterTimeJoinNode-28
+ * │ ├──SeriesScanNode-26:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-27:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-38: [SourceAddress:192.0.3.1/test.2.0/41]
+ * ├──ExchangeNode-39: [SourceAddress:192.0.2.1/test.3.0/42]
+ * └──ExchangeNode-40: [SourceAddress:192.0.4.1/test.4.0/43]
+ *
+ * IdentitySinkNode-41
+ * └──TopK-36
+ * └──SingleDeviceView-25
+ * └──FullOuterTimeJoinNode-24
+ * ├──SeriesScanNode-22:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-23:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-42
+ * └──TopK-35
+ * └──SingleDeviceView-21
+ * └──FullOuterTimeJoinNode-20
+ * ├──SeriesScanNode-18:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-19:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-43
+ * └──TopK-37
+ * └──SingleDeviceView-33
+ * └──FullOuterTimeJoinNode-32
+ * ├──SeriesScanNode-30:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-31:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByTimeTest1() {
+ // only order by time, no filter
+ sql =
String.format(
- "select * from root.sg.** ORDER BY TIME DESC LIMIT %s align by device", LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan = planner.planFragments();
+ "select * from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY TIME DESC LIMIT %s align by device",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
assertEquals(4, plan.getInstances().size());
- PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
assertTrue(firstFiRoot instanceof IdentitySinkNode);
assertEquals(4, firstFiRoot.getChildren().get(0).getChildren().size());
PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
@@ -83,25 +353,67 @@ public class AlignByDeviceOrderByLimitOffsetTest {
plan.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE);
}
+ /*
+ * IdentitySinkNode-60
+ * └──TopK-13
+ * ├──TopK-50
+ * │ ├──SingleDeviceView-25
+ * │ │ └──LimitNode-24
+ * │ │ └──FilterNode-23
+ * │ │ └──FullOuterTimeJoinNode-22
+ * │ │ ├──SeriesScanNode-20:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──SeriesScanNode-21:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SingleDeviceView-43
+ * │ └──LimitNode-42
+ * │ └──FilterNode-41
+ * │ └──FullOuterTimeJoinNode-40
+ * │ ├──SeriesScanNode-38:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-39:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-54: [SourceAddress:192.0.3.1/test.2.0/57]
+ * ├──ExchangeNode-55: [SourceAddress:192.0.2.1/test.3.0/58]
+ * └──ExchangeNode-56: [SourceAddress:192.0.4.1/test.4.0/59]
+ *
+ * IdentitySinkNode-57
+ * └──TopK-52
+ * └──SingleDeviceView-37
+ * └──LimitNode-36
+ * └──FilterNode-35
+ * └──FullOuterTimeJoinNode-34
+ * ├──SeriesScanNode-32:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-33:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-58
+ * └──TopK-51
+ * └──SingleDeviceView-31
+ * └──LimitNode-30
+ * └──FilterNode-29
+ * └──FullOuterTimeJoinNode-28
+ * ├──SeriesScanNode-26:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-27:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-59
+ * └──TopK-53
+ * └──SingleDeviceView-49
+ * └──LimitNode-48
+ * └──FilterNode-47
+ * └──FullOuterTimeJoinNode-46
+ * ├──SeriesScanNode-44:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-45:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
@Test
- public void orderByTimeWithValueFilterTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
-
- // 2. order by time, has value filter
+ public void orderByTimeTest2() {
+ // only order by time, has value filter
// put LIMIT-NODE below of SingleDeviceViewNode
- String sql =
+ sql =
String.format(
- "select * from root.sg.** where s1>1 ORDER BY TIME DESC LIMIT %s align by device",
+ "select s1 from root.sg.d1,root.sg.d22,root.sg.d333 where s2>1 ORDER BY TIME DESC LIMIT %s align by device",
LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan = planner.planFragments();
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
assertEquals(4, plan.getInstances().size());
- PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
assertTrue(firstFiTopNode instanceof TopKNode);
for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
@@ -118,28 +430,69 @@ public class AlignByDeviceOrderByLimitOffsetTest {
assertScanNodeLimitValue(plan.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
}
+ /*
+ * IdentitySinkNode-98
+ * └──TopK-54
+ * ├──TopK-88
+ * │ └──DeviceView-69
+ * │ ├──LimitNode-63
+ * │ │ └──FilterNode-62
+ * │ │ └──FullOuterTimeJoinNode-61
+ * │ │ ├──SeriesScanNode-59:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──SeriesScanNode-60:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──LimitNode-68
+ * │ └──FilterNode-67
+ * │ └──FullOuterTimeJoinNode-66
+ * │ ├──SeriesScanNode-64:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-65:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-92: [SourceAddress:192.0.3.1/test.6.0/95]
+ * ├──ExchangeNode-93: [SourceAddress:192.0.2.1/test.7.0/96]
+ * └──ExchangeNode-94: [SourceAddress:192.0.4.1/test.8.0/97]
+ *
+ * IdentitySinkNode-95
+ * └──TopK-89
+ * └──DeviceView-75
+ * └──LimitNode-74
+ * └──FilterNode-73
+ * └──FullOuterTimeJoinNode-72
+ * ├──SeriesScanNode-70:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-71:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-96
+ * └──TopK-90
+ * └──DeviceView-81
+ * └──LimitNode-80
+ * └──FilterNode-79
+ * └──FullOuterTimeJoinNode-78
+ * ├──SeriesScanNode-76:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-77:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-97
+ * └──TopK-91
+ * └──DeviceView-87
+ * └──LimitNode-86
+ * └──FilterNode-85
+ * └──FullOuterTimeJoinNode-84
+ * ├──SeriesScanNode-82:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-83:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
@Test
- public void orderByTimeAndExpressionNoValueFilterTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
-
- // 3. order by time, expression
+ public void orderByTimeTest3() {
+ // order by time and expression, no filter
// need read all data, use DeviceViewNode instead of SingleDeviceViewNode
- // no LIMIT-NODE below DeviceViewNode
+ // no LimitNode below DeviceViewNode
// can push down LIMIT value to ScanNode
- String sql =
+ sql =
String.format(
- "select * from root.sg.** ORDER BY TIME DESC, s1 DESC LIMIT %s align by device",
+ "select s1 from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY TIME DESC, s2 DESC LIMIT %s align by device",
LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan = planner.planFragments();
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
assertEquals(4, plan.getInstances().size());
- PlanNode firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
- PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
assertTrue(firstFiTopNode instanceof TopKNode);
for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
assertTrue(node instanceof DeviceViewNode);
@@ -155,29 +508,22 @@ public class AlignByDeviceOrderByLimitOffsetTest {
plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE);
assertScanNodeLimitValue(
plan.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE);
- }
-
- @Test
- public void orderByTimeAndExpressionWithValueFilterTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
- // 4. order by time, expression, has value filter
+ // order by time and expression, has value filter
// need read all data, use DeviceViewNode instead of SingleDeviceViewNode
- // no LIMIT-NODE below DeviceViewNode
- String sql =
+ // has LimitNode below DeviceViewNode
+ // can not push down LIMIT value to ScanNode
+ sql =
String.format(
- "select * from root.sg.** where s1>1 ORDER BY TIME DESC, s1 DESC LIMIT %s align by device",
+ "select s1 from root.sg.d1,root.sg.d22,root.sg.d333 where s2>1 ORDER BY TIME DESC, s2 DESC LIMIT %s align by device",
LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan4 = planner.planFragments();
- assertEquals(4, plan4.getInstances().size());
- PlanNode firstFiRoot = plan4.getInstances().get(0).getFragment().getPlanNodeTree();
- PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
assertTrue(firstFiTopNode instanceof TopKNode);
for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
assertTrue(node instanceof DeviceViewNode);
@@ -187,63 +533,225 @@ public class AlignByDeviceOrderByLimitOffsetTest {
assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
- assertScanNodeLimitValue(plan4.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan4.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan4.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan4.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
+ assertScanNodeLimitValue(plan.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
}
+ /*
+ * IdentitySinkNode-21
+ * └──MergeSort-8
+ * ├──SingleDeviceView-5
+ * │ └──AggregationNode-9
+ * │ ├──SeriesAggregationScanNode-10:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-15: [SourceAddress:192.0.2.1/test.2.0/18]
+ * ├──ExchangeNode-17: [SourceAddress:192.0.3.1/test.3.0/19]
+ * └──SingleDeviceView-7
+ * └──AggregationNode-12
+ * ├──SeriesAggregationScanNode-13:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * └──ExchangeNode-16: [SourceAddress:192.0.4.1/test.4.0/20]
+ *
+ * ShuffleSinkNode-18
+ * └──SeriesAggregationScanNode-11:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * ShuffleSinkNode-19
+ * └──SingleDeviceView-6
+ * └──SeriesAggregationScanNode-2:[SeriesPath: root.sg.d22.s1, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * ShuffleSinkNode-20
+ * └──SeriesAggregationScanNode-14:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
@Test
- public void orderByExpressionTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ public void orderByTimeTest4() {
+ // aggregation + order by time, no LIMIT
+ // SingleDeviceViewNode + MergeSortNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY TIME DESC align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof MergeSortNode);
+ assertEquals(3, firstFiTopNode.getChildren().size());
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof SingleDeviceViewNode);
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof SingleDeviceViewNode);
- // 5. only order by expression
- String sql =
- String.format(
- "select * from root.sg.** ORDER BY s1 DESC LIMIT %s align by device", LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan5 = planner.planFragments();
- assertEquals(4, plan5.getInstances().size());
- PlanNode firstFiRoot = plan5.getInstances().get(0).getFragment().getPlanNodeTree();
- PlanNode firstFiTopNode = firstFiRoot.getChildren().get(0);
+ // aggregation + order by time + group by time, no LIMIT
+ // SingleDeviceViewNode + MergeSortNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 group by ((1,10], 1ms) ORDER BY TIME DESC align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof MergeSortNode);
+ assertEquals(3, firstFiTopNode.getChildren().size());
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof SingleDeviceViewNode);
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof SingleDeviceViewNode);
+ }
+
+ /*
+ * IdentitySinkNode-22
+ * └──TopK-4
+ * ├──TopK-16
+ * │ ├──SingleDeviceView-5
+ * │ │ └──AggregationNode-8
+ * │ │ ├──SeriesAggregationScanNode-9:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──ExchangeNode-14: [SourceAddress:192.0.2.1/test.2.0/19]
+ * │ └──SingleDeviceView-7
+ * │ └──AggregationNode-11
+ * │ ├──SeriesAggregationScanNode-12:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-15: [SourceAddress:192.0.4.1/test.3.0/20]
+ * └──ExchangeNode-18: [SourceAddress:192.0.3.1/test.4.0/21]
+ *
+ * IdentitySinkNode-19
+ * └──SeriesAggregationScanNode-10:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-20
+ * └──SeriesAggregationScanNode-13:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ *
+ * IdentitySinkNode-21
+ * └──TopK-17
+ * └──SingleDeviceView-6
+ * └──SeriesAggregationScanNode-2:[SeriesPath: root.sg.d22.s1, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ */
+ @Test
+ public void orderByTimeTest5() {
+ // aggregation + order by time, has LIMIT
+ // SingleDeviceViewNode + TopKNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY TIME DESC LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
assertTrue(firstFiTopNode instanceof TopKNode);
- for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
- assertTrue(node instanceof DeviceViewNode);
- assertTrue(node.getChildren().get(0) instanceof FullOuterTimeJoinNode);
- }
+ assertEquals(2, firstFiTopNode.getChildren().size());
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof TopKNode);
assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
- assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
- assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
- assertScanNodeLimitValue(plan5.getInstances().get(0).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan5.getInstances().get(1).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan5.getInstances().get(2).getFragment().getPlanNodeTree(), 0);
- assertScanNodeLimitValue(plan5.getInstances().get(3).getFragment().getPlanNodeTree(), 0);
+ assertTrue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ assertTrue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ assertTrue(
+ plan.getInstances().get(3).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof TopKNode);
+
+ // aggregation + order by time + group by time, has LIMIT
+ // SingleDeviceViewNode + TopKNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 group by ((1,10], 1ms) ORDER BY TIME DESC LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof TopKNode);
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ assertTrue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ assertTrue(
+ plan.getInstances().get(3).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof TopKNode);
}
+ /*
+ * IdentitySinkNode-24
+ * └──LimitNode-17
+ * └──FilterNode-8
+ * └──MergeSort-10
+ * ├──SingleDeviceView-5
+ * │ └──AggregationNode-11
+ * │ ├──SeriesAggregationScanNode-12:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-18: [SourceAddress:192.0.2.1/test.2.0/21]
+ * ├──ExchangeNode-20: [SourceAddress:192.0.3.1/test.3.0/22]
+ * └──SingleDeviceView-7
+ * └──AggregationNode-14
+ * ├──SeriesAggregationScanNode-15:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * └──ExchangeNode-19: [SourceAddress:192.0.4.1/test.4.0/23]
+ *
+ * IdentitySinkNode-21
+ * └──SeriesAggregationScanNode-13:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-22
+ * └──SingleDeviceView-6
+ * └──SeriesAggregationScanNode-2:[SeriesPath: root.sg.d22.s1, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-23
+ * └──SeriesAggregationScanNode-16:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
@Test
- public void orderByTimeWithOffsetTest() {
- QueryId queryId = new QueryId("test");
- MPPQueryContext context =
- new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ public void orderByTimeTest6() {
+ // aggregation + order by time + having, has LIMIT
+ // SingleDeviceViewNode + MergeSortNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 having(count(s1)>1) ORDER BY TIME DESC LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof LimitNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof FilterNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof MergeSortNode);
+ assertEquals(3, firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().size());
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof SingleDeviceViewNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(1)
+ instanceof ExchangeNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(2)
+ instanceof SingleDeviceViewNode);
+ assertTrue(
+ plan.getInstances().get(1).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ assertTrue(
+ plan.getInstances().get(2).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SingleDeviceViewNode);
+ assertTrue(
+ plan.getInstances().get(3).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SeriesAggregationScanNode);
+ }
- // 6. order by time, offset + limit
+ @Test
+ public void orderByTimeWithOffsetTest() {
+ // order by time, offset + limit
// on top of TOP-K NODE, LIMIT-NODE is necessary
- String sql =
+ sql =
String.format(
"select * from root.sg.** ORDER BY time DESC OFFSET %s LIMIT %s align by device",
LIMIT_VALUE, LIMIT_VALUE);
- Analysis analysis = Util.analyze(sql, context);
- PlanNode logicalPlanNode = Util.genLogicalPlan(analysis, context);
- DistributionPlanner planner =
- new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
- DistributedQueryPlan plan6 = planner.planFragments();
- assertEquals(4, plan6.getInstances().size());
- PlanNode firstFiRoot = plan6.getInstances().get(0).getFragment().getPlanNodeTree();
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
PlanNode firstFIFirstNode = firstFiRoot.getChildren().get(0);
assertTrue(firstFIFirstNode instanceof LimitNode);
PlanNode firstFiTopNode = ((LimitNode) firstFIFirstNode).getChild().getChildren().get(0);
@@ -254,13 +762,275 @@ public class AlignByDeviceOrderByLimitOffsetTest {
assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
assertScanNodeLimitValue(
- plan6.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ plan.getInstances().get(0).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
assertScanNodeLimitValue(
- plan6.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ plan.getInstances().get(1).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
assertScanNodeLimitValue(
- plan6.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ plan.getInstances().get(2).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
assertScanNodeLimitValue(
- plan6.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ plan.getInstances().get(3).getFragment().getPlanNodeTree(), LIMIT_VALUE * 2);
+ }
+
+ /*
+ * IdentitySinkNode-43
+ * └──TransformNode-12
+ * └──MergeSort-32
+ * ├──SortNode-33
+ * │ └──DeviceView-19
+ * │ ├──FullOuterTimeJoinNode-15
+ * │ │ ├──SeriesScanNode-13:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──SeriesScanNode-14:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──FullOuterTimeJoinNode-18
+ * │ ├──SeriesScanNode-16:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-17:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-37: [SourceAddress:192.0.3.1/test.2.0/40]
+ * ├──ExchangeNode-38: [SourceAddress:192.0.2.1/test.3.0/41]
+ * └──ExchangeNode-39: [SourceAddress:192.0.4.1/test.4.0/42]
+ *
+ * IdentitySinkNode-40
+ * └──SortNode-34
+ * └──DeviceView-23
+ * └──FullOuterTimeJoinNode-22
+ * ├──SeriesScanNode-20:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-21:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-41
+ * └──SortNode-35
+ * └──DeviceView-27
+ * └──FullOuterTimeJoinNode-26
+ * ├──SeriesScanNode-24:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-25:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-42
+ * └──SortNode-36
+ * └──DeviceView-31
+ * └──FullOuterTimeJoinNode-30
+ * ├──SeriesScanNode-28:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-29:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByExpressionTest1() {
+ // only order by expression, no LIMIT
+ // use MergeSortNode + SortNode + TransformNode
+ sql = "select s1 from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY s2 DESC align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TransformNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof MergeSortNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof SortNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof DeviceViewNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(3) instanceof ExchangeNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof SortNode);
+ assertTrue(
+ plan.getInstances()
+ .get(i)
+ .getFragment()
+ .getPlanNodeTree()
+ .getChildren()
+ .get(0)
+ .getChildren()
+ .get(0)
+ instanceof DeviceViewNode);
+ }
+ for (int i = 0; i < 4; i++) {
+ assertScanNodeLimitValue(plan.getInstances().get(i).getFragment().getPlanNodeTree(), 0);
+ }
+ }
+
+ /*
+ * IdentitySinkNode-41
+ * └──TopK-10
+ * ├──TopK-31
+ * │ └──DeviceView-18
+ * │ ├──FullOuterTimeJoinNode-14
+ * │ │ ├──SeriesScanNode-12:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ │ └──SeriesScanNode-13:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──FullOuterTimeJoinNode-17
+ * │ ├──SeriesScanNode-15:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──SeriesScanNode-16:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-35: [SourceAddress:192.0.3.1/test.2.0/38]
+ * ├──ExchangeNode-36: [SourceAddress:192.0.2.1/test.3.0/39]
+ * └──ExchangeNode-37: [SourceAddress:192.0.4.1/test.4.0/40]
+ *
+ * IdentitySinkNode-38
+ * └──TopK-32
+ * └──DeviceView-22
+ * └──FullOuterTimeJoinNode-21
+ * ├──SeriesScanNode-19:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-20:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-39
+ * └──TopK-33
+ * └──DeviceView-26
+ * └──FullOuterTimeJoinNode-25
+ * ├──SeriesScanNode-23:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-24:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-40
+ * └──TopK-34
+ * └──DeviceView-30
+ * └──FullOuterTimeJoinNode-29
+ * ├──SeriesScanNode-27:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-28:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByExpressionTest2() {
+ // only order by expression, has LIMIT
+ // use TopKNode
+ sql =
+ "select s1 from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY s2 DESC LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ for (PlanNode node : firstFiTopNode.getChildren().get(0).getChildren()) {
+ assertTrue(node instanceof DeviceViewNode);
+ assertTrue(node.getChildren().get(0) instanceof FullOuterTimeJoinNode);
+ }
+ assertTrue(firstFiTopNode.getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(2) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(3) instanceof ExchangeNode);
+ for (int i = 0; i < 4; i++) {
+ assertScanNodeLimitValue(plan.getInstances().get(i).getFragment().getPlanNodeTree(), 0);
+ }
+ }
+
+ /*
+ * IdentitySinkNode-35
+ * └──TransformNode-12
+ * └──SortNode-11
+ * └──DeviceView-13
+ * ├──AggregationNode-18
+ * │ ├──SeriesAggregationScanNode-14:[SeriesPath: root.sg.d1.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ ├──SeriesAggregationScanNode-16:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-29: [SourceAddress:192.0.2.1/test.2.0/32]
+ * ├──ExchangeNode-31: [SourceAddress:192.0.3.1/test.3.0/33]
+ * └──AggregationNode-27
+ * ├──SeriesAggregationScanNode-23:[SeriesPath: root.sg.d333.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesAggregationScanNode-25:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * └──ExchangeNode-30: [SourceAddress:192.0.4.1/test.4.0/34]
+ *
+ * IdentitySinkNode-32
+ * └──HorizontallyConcatNode-19
+ * ├──SeriesAggregationScanNode-15:[SeriesPath: root.sg.d1.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesAggregationScanNode-17:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-33
+ * └──HorizontallyConcatNode-22
+ * ├──SeriesAggregationScanNode-20:[SeriesPath: root.sg.d22.s1, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesAggregationScanNode-21:[SeriesPath: root.sg.d22.s2, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-34
+ * └──HorizontallyConcatNode-28
+ * ├──SeriesAggregationScanNode-24:[SeriesPath: root.sg.d333.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesAggregationScanNode-26:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByExpressionTest3() {
+ // aggregation, order by expression, no LIMIT
+ // use MergeSortNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY count(s2) DESC align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TransformNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof SortNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof AggregationNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(1)
+ instanceof ExchangeNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(2)
+ instanceof AggregationNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof HorizontallyConcatNode);
+ }
+ for (int i = 0; i < 4; i++) {
+ assertScanNodeLimitValue(plan.getInstances().get(i).getFragment().getPlanNodeTree(), 0);
+ }
+ }
+
+ /*
+ * IdentitySinkNode-34
+ * └──TopK-10
+ * └──DeviceView-12
+ * ├──AggregationNode-17
+ * │ ├──SeriesAggregationScanNode-13:[SeriesPath: root.sg.d1.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ ├──SeriesAggregationScanNode-15:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * │ └──ExchangeNode-28: [SourceAddress:192.0.2.1/test.2.0/31]
+ * ├──ExchangeNode-30: [SourceAddress:192.0.3.1/test.3.0/32]
+ * └──AggregationNode-26
+ * ├──SeriesAggregationScanNode-22:[SeriesPath: root.sg.d333.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesAggregationScanNode-24:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * └──ExchangeNode-29: [SourceAddress:192.0.4.1/test.4.0/33]
+ *
+ * IdentitySinkNode-31
+ * └──HorizontallyConcatNode-18
+ * ├──SeriesAggregationScanNode-14:[SeriesPath: root.sg.d1.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesAggregationScanNode-16:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-32
+ * └──HorizontallyConcatNode-21
+ * ├──SeriesAggregationScanNode-19:[SeriesPath: root.sg.d22.s1, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesAggregationScanNode-20:[SeriesPath: root.sg.d22.s2, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-33
+ * └──HorizontallyConcatNode-27
+ * ├──SeriesAggregationScanNode-23:[SeriesPath: root.sg.d333.s2, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesAggregationScanNode-25:[SeriesPath: root.sg.d333.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByExpressionTest4() {
+ // aggregation, order by expression, has LIMIT
+ // use TopKNode, not need MergeSortNode and LimitNode
+ sql =
+ "select count(s1) from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY count(s2) DESC LIMIT 10 align by device";
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof DeviceViewNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof AggregationNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(1) instanceof ExchangeNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(2) instanceof AggregationNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof HorizontallyConcatNode);
+ }
+ for (int i = 0; i < 4; i++) {
+ assertScanNodeLimitValue(plan.getInstances().get(i).getFragment().getPlanNodeTree(), 0);
+ }
}
private void assertScanNodeLimitValue(PlanNode root, long limitValue) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByTimeOrderByLimitOffsetTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByTimeOrderByLimitOffsetTest.java
new file mode 100644
index 00000000000..e1a77a2d8ce
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/AlignByTimeOrderByLimitOffsetTest.java
@@ -0,0 +1,316 @@
+/*
+ * 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.planner.distribution;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+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.AggregationNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.FillNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.OffsetNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.SortNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TransformNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.join.FullOuterTimeJoinNode;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AlignByTimeOrderByLimitOffsetTest {
+
+ private static final long LIMIT_VALUE = 10;
+
+ QueryId queryId = new QueryId("test");
+ MPPQueryContext context =
+ new MPPQueryContext("", queryId, null, new TEndPoint(), new TEndPoint());
+ String sql;
+ Analysis analysis;
+ PlanNode logicalPlanNode;
+ DistributionPlanner planner;
+ DistributedQueryPlan plan;
+ PlanNode firstFiRoot;
+ PlanNode firstFiTopNode;
+ PlanNode mergeSortNode;
+
+ /*
+ * IdentitySinkNode-63
+ * └──LimitNode-55
+ * └──OffsetNode-40
+ * └──TopK-62
+ * └──FullOuterTimeJoinNode-38
+ * ├──SeriesScanNode-42:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-46:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-48:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-50:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-56: [SourceAddress:192.0.3.1/test.6.0/59]
+ * ├──ExchangeNode-57: [SourceAddress:192.0.2.1/test.7.0/60]
+ * └──ExchangeNode-58: [SourceAddress:192.0.4.1/test.8.0/61]
+ *
+ * IdentitySinkNode-59
+ * └──FullOuterTimeJoinNode-52
+ * ├──SeriesScanNode-44:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ * └──SeriesScanNode-45:[SeriesPath: root.sg.d22.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-60
+ * └──FullOuterTimeJoinNode-53
+ * ├──SeriesScanNode-47:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ * └──SeriesScanNode-49:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ *
+ * IdentitySinkNode-61
+ * └──FullOuterTimeJoinNode-54
+ * ├──SeriesScanNode-43:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ * └──SeriesScanNode-51:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:4)]
+ */
+ @Test
+ public void orderByExpressionTest1() {
+ // select * order by expression + limit N
+ // use TopKNode to replace SortNode + LimitNode
+ sql =
+ String.format(
+ "select * from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY root.sg.d1.s1 DESC LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TopKNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof FullOuterTimeJoinNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ }
+
+ // select * order by expression + offset M + limit N
+ // use TopKNode to replace SortNode
+ sql =
+ String.format(
+ "select * from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY root.sg.d1.s1 DESC OFFSET 5 LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof LimitNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof OffsetNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof TopKNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ }
+ }
+
+ /*
+ * IdentitySinkNode-33
+ * └──LimitNode-27
+ * └──OffsetNode-22
+ * └──TransformNode-21
+ * └──TopK-32
+ * └──FullOuterTimeJoinNode-19
+ * ├──SeriesScanNode-25:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-28: [SourceAddress:192.0.3.1/test.5.0/30]
+ * └──ExchangeNode-29: [SourceAddress:192.0.2.1/test.6.0/31]
+ *
+ * IdentitySinkNode-30
+ * └──SeriesScanNode-24:[SeriesPath: root.sg.d22.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-31
+ * └──SeriesScanNode-26:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ */
+ @Test
+ public void orderByExpressionTest2() {
+ // select s1 order by s2 + limit N
+ sql =
+ String.format(
+ "select s1 from root.sg.d1 ORDER BY root.sg.d22.s2 DESC LIMIT %s", LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TransformNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof TopKNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof FullOuterTimeJoinNode);
+
+ // select s1 order by s2 + offset M + limit N
+ sql =
+ String.format(
+ "select s1 from root.sg.d1 ORDER BY root.sg.d22.s2 DESC OFFSET 5 LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof LimitNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof OffsetNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof TransformNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TopKNode);
+ }
+
+ /*
+ * IdentitySinkNode-17
+ * └──TransformNode-5
+ * └──TopK-16
+ * └──AggregationNode-10
+ * ├──SeriesAggregationScanNode-7:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-12: [SourceAddress:192.0.3.1/test.2.0/14]
+ * └──ExchangeNode-13: [SourceAddress:192.0.2.1/test.3.0/15]
+ *
+ * IdentitySinkNode-14
+ * └──SeriesAggregationScanNode-9:[SeriesPath: root.sg.d22.s2, Descriptor: [AggregationDescriptor(count, SINGLE)], DataRegion: TConsensusGroupId(type:DataRegion, id:3)]
+ *
+ * IdentitySinkNode-15
+ * └──SeriesAggregationScanNode-8:[SeriesPath: root.sg.d1.s1, Descriptor: [AggregationDescriptor(count, PARTIAL)], DataRegion: TConsensusGroupId(type:DataRegion, id:2)]
+ */
+ @Test
+ public void orderByExpressionTest3() {
+ // aggregation, select count(root.sg.d1.s1) order by count(root.sg.d22.s2), LIMIT N
+ sql =
+ String.format(
+ "select count(s1) from root.sg.d1 ORDER BY count(root.sg.d22.s2) DESC LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof TransformNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof TopKNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof AggregationNode);
+
+ // aggregation, select count(root.sg.d1.s1) order by count(root.sg.d22.s2), OFFSET M LIMIT N
+ sql =
+ String.format(
+ "select count(s1) from root.sg.d1 ORDER BY count(root.sg.d22.s2) DESC OFFSET 5 LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(3, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof LimitNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof OffsetNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof TransformNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TopKNode);
+ }
+
+ /*
+ * IdentitySinkNode-32
+ * └──FillNode-9
+ * └──TopK-31
+ * └──FullOuterTimeJoinNode-7
+ * ├──SeriesScanNode-11:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-15:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-17:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-19:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-25: [SourceAddress:192.0.3.1/test.2.0/28]
+ * ├──ExchangeNode-26: [SourceAddress:192.0.2.1/test.3.0/29]
+ * └──ExchangeNode-27: [SourceAddress:192.0.4.1/test.4.0/30]
+ */
+ @Test
+ public void orderByFillTest() {
+ sql =
+ String.format(
+ "select * from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY root.sg.d1.s1 DESC fill(previous) LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof FillNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof TopKNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ }
+
+ /*
+ * IdentitySinkNode-63
+ * └──LimitNode-56
+ * └──FillNode-41
+ * └──SortNode-40
+ * └──FullOuterTimeJoinNode-39
+ * ├──SeriesScanNode-43:[SeriesPath: root.sg.d333.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-47:[SeriesPath: root.sg.d1.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-49:[SeriesPath: root.sg.d1.s2, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──SeriesScanNode-51:[SeriesPath: root.sg.d333.s1, DataRegion: TConsensusGroupId(type:DataRegion, id:1)]
+ * ├──ExchangeNode-57: [SourceAddress:192.0.3.1/test.6.0/60]
+ * ├──ExchangeNode-58: [SourceAddress:192.0.2.1/test.7.0/61]
+ * └──ExchangeNode-59: [SourceAddress:192.0.4.1/test.8.0/62]
+ */
+ sql =
+ String.format(
+ "select * from root.sg.d1,root.sg.d22,root.sg.d333 ORDER BY root.sg.d1.s1 DESC fill(linear) LIMIT %s",
+ LIMIT_VALUE);
+ analysis = Util.analyze(sql, context);
+ logicalPlanNode = Util.genLogicalPlan(analysis, context);
+ planner = new DistributionPlanner(analysis, new LogicalQueryPlan(context, logicalPlanNode));
+ plan = planner.planFragments();
+ assertEquals(4, plan.getInstances().size());
+ firstFiRoot = plan.getInstances().get(0).getFragment().getPlanNodeTree();
+ firstFiTopNode = firstFiRoot.getChildren().get(0);
+ assertTrue(firstFiTopNode instanceof LimitNode);
+ assertTrue(firstFiTopNode.getChildren().get(0) instanceof FillNode);
+ assertTrue(firstFiTopNode.getChildren().get(0).getChildren().get(0) instanceof SortNode);
+ assertTrue(
+ firstFiTopNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ for (int i = 1; i < 4; i++) {
+ assertTrue(
+ plan.getInstances().get(i).getFragment().getPlanNodeTree().getChildren().get(0)
+ instanceof FullOuterTimeJoinNode);
+ }
+ }
+}