You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hu...@apache.org on 2023/02/21 07:52:58 UTC
[iotdb] 01/03: finish
This is an automated email from the ASF dual-hosted git repository.
hui pushed a commit to branch lmh/planPushDown
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3a65552ba445f617b78e06b31b8df421775d77b9
Author: Minghui Liu <li...@foxmail.com>
AuthorDate: Tue Feb 21 15:16:04 2023 +0800
finish
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 8 +-
.../apache/iotdb/db/mpp/plan/analyze/Analysis.java | 4 +
.../db/mpp/plan/analyze/ExpressionAnalyzer.java | 33 ++++
.../memory/StatementMemorySourceVisitor.java | 5 +-
.../mpp/plan/optimization/LimitOffsetPushDown.java | 214 +++++++++++++++++++++
.../optimization/LimitOffsetPushDownOptimizer.java | 42 ----
.../db/mpp/plan/optimization/PlanOptimizer.java | 3 +-
.../iotdb/db/mpp/plan/planner/LogicalPlanner.java | 2 +-
.../db/mpp/plan/planner/OperatorTreeGenerator.java | 2 +
.../planner/distribution/DistributionPlanner.java | 27 ++-
.../plan/planner/plan/node/PlanGraphPrinter.java | 4 +
.../db/mpp/plan/planner/plan/node/PlanVisitor.java | 51 +++--
.../plan/node/source/AlignedSeriesScanNode.java | 8 +
13 files changed, 326 insertions(+), 77 deletions(-)
diff --git a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index f213a4187b..d17c74f354 100644
--- a/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++ b/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -530,13 +530,13 @@ paginationClause
;
rowPaginationClause
- : limitClause offsetClause?
- | offsetClause? limitClause
+ : limitClause? offsetClause?
+ | offsetClause? limitClause?
;
seriesPaginationClause
- : slimitClause soffsetClause?
- | soffsetClause? slimitClause
+ : slimitClause? soffsetClause?
+ | soffsetClause? slimitClause?
;
limitClause
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
index 14ee67b535..bf15b2a826 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/Analysis.java
@@ -582,4 +582,8 @@ public class Analysis {
public void setVirtualSource(boolean virtualSource) {
isVirtualSource = virtualSource;
}
+
+ public Map<NodeRef<Expression>, TSDataType> getExpressionTypes() {
+ return expressionTypes;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
index 20588ac297..09aa5bbd72 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/analyze/ExpressionAnalyzer.java
@@ -1331,4 +1331,37 @@ public class ExpressionAnalyzer {
"unsupported expression type: " + expression.getExpressionType());
}
}
+
+ public static boolean check(Expression expression, Analysis analysis) {
+ if (expression instanceof TernaryExpression) {
+ TernaryExpression ternaryExpression = (TernaryExpression) expression;
+ return check(ternaryExpression.getFirstExpression(), analysis)
+ && check(ternaryExpression.getSecondExpression(), analysis)
+ && check(ternaryExpression.getThirdExpression(), analysis);
+ } else if (expression instanceof BinaryExpression) {
+ BinaryExpression binaryExpression = (BinaryExpression) expression;
+ return check(binaryExpression.getLeftExpression(), analysis)
+ && check(binaryExpression.getRightExpression(), analysis);
+ } else if (expression instanceof UnaryExpression) {
+ return check(((UnaryExpression) expression).getExpression(), analysis);
+ } else if (expression instanceof FunctionExpression) {
+ FunctionExpression functionExpression = (FunctionExpression) expression;
+ if (functionExpression.isBuiltInAggregationFunctionExpression()
+ || functionExpression.isMappable(analysis.getExpressionTypes())) {
+ return false;
+ }
+ List<Expression> inputExpressions = functionExpression.getExpressions();
+ for (Expression inputExpression : inputExpressions) {
+ if (!check(inputExpression, analysis)) {
+ return false;
+ }
+ }
+ return true;
+ } else if (expression instanceof LeafOperand) {
+ return true;
+ } else {
+ throw new IllegalArgumentException(
+ "unsupported expression type: " + expression.getExpressionType());
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/memory/StatementMemorySourceVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/memory/StatementMemorySourceVisitor.java
index 44a131ec02..d1f15a9534 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/memory/StatementMemorySourceVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/memory/StatementMemorySourceVisitor.java
@@ -92,8 +92,11 @@ public class StatementMemorySourceVisitor
.plan(context.getAnalysis());
DistributionPlanner planner = new DistributionPlanner(context.getAnalysis(), logicalPlan);
PlanNode rootWithExchange = planner.addExchangeNode(planner.rewriteSource());
+ PlanNode optimizedRootWithExchange = planner.optimize(rootWithExchange);
+
List<String> lines =
- rootWithExchange.accept(new PlanGraphPrinter(), new PlanGraphPrinter.GraphContext());
+ optimizedRootWithExchange.accept(
+ new PlanGraphPrinter(), new PlanGraphPrinter.GraphContext());
TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.TEXT));
lines.forEach(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDown.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDown.java
new file mode 100644
index 0000000000..55fee2de2f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDown.java
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.plan.optimization;
+
+import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
+import org.apache.iotdb.db.mpp.plan.analyze.ExpressionAnalyzer;
+import org.apache.iotdb.db.mpp.plan.expression.Expression;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FillNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.FilterNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildProcessNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SingleChildProcessNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.TransformNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.AlignedSeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
+import org.apache.iotdb.db.mpp.plan.statement.StatementType;
+import org.apache.iotdb.db.mpp.plan.statement.component.FillPolicy;
+import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+
+public class LimitOffsetPushDown 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.isAggregationQuery()
+ || (!queryStatement.hasLimit() && !queryStatement.hasOffset())) {
+ return plan;
+ }
+ return plan.accept(new Rewriter(), new RewriterContext(analysis));
+ }
+
+ private static class Rewriter extends PlanVisitor<PlanNode, RewriterContext> {
+
+ @Override
+ public PlanNode visitPlan(PlanNode node, RewriterContext context) {
+ return null;
+ }
+
+ @Override
+ public PlanNode visitLimit(LimitNode node, RewriterContext context) {
+ PlanNode child = node.getChild();
+ PlanNode parent = context.getParent();
+
+ context.setParent(node);
+ context.setLimit(node.getLimit());
+ child.accept(this, context);
+
+ if (context.isEnablePushDown()) {
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(child);
+ return parent;
+ } else {
+ return node.getChild();
+ }
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitOffset(OffsetNode node, RewriterContext context) {
+ PlanNode child = node.getChild();
+ PlanNode parent = context.getParent();
+
+ context.setParent(node);
+ context.setOffset(node.getOffset());
+ child.accept(this, context);
+
+ if (context.isEnablePushDown()) {
+ if (parent != null) {
+ ((SingleChildProcessNode) parent).setChild(child);
+ return parent;
+ } else {
+ return node.getChild();
+ }
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitFill(FillNode node, RewriterContext context) {
+ FillPolicy fillPolicy = node.getFillDescriptor().getFillPolicy();
+ if (fillPolicy == FillPolicy.VALUE) {
+ node.getChild().accept(this, context);
+ } else {
+ context.setEnablePushDown(false);
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitFilter(FilterNode node, RewriterContext context) {
+ context.setEnablePushDown(false);
+ return node;
+ }
+
+ @Override
+ public PlanNode visitTransform(TransformNode node, RewriterContext context) {
+ Expression[] outputExpressions = node.getOutputExpressions();
+ boolean enablePushDown = true;
+ for (Expression expression : outputExpressions) {
+ if (!ExpressionAnalyzer.check(expression, context.getAnalysis())) {
+ enablePushDown = false;
+ break;
+ }
+ }
+
+ if (enablePushDown) {
+ node.getChild().accept(this, context);
+ } else {
+ context.setEnablePushDown(false);
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitMultiChildProcess(MultiChildProcessNode node, RewriterContext context) {
+ context.setEnablePushDown(false);
+ return node;
+ }
+
+ @Override
+ public PlanNode visitSeriesScan(SeriesScanNode node, RewriterContext context) {
+ if (context.isEnablePushDown()) {
+ node.setLimit(context.getLimit());
+ node.setOffset(context.getOffset());
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitAlignedSeriesScan(AlignedSeriesScanNode node, RewriterContext context) {
+ if (context.isEnablePushDown()) {
+ node.setLimit(context.getLimit());
+ node.setOffset(context.getOffset());
+ }
+ return node;
+ }
+ }
+
+ private static class RewriterContext {
+ private int limit;
+ private int offset;
+
+ private boolean enablePushDown = true;
+
+ private PlanNode parent;
+
+ private final Analysis analysis;
+
+ public RewriterContext(Analysis analysis) {
+ this.analysis = analysis;
+ }
+
+ public int getLimit() {
+ return limit;
+ }
+
+ public void setLimit(int limit) {
+ this.limit = limit;
+ }
+
+ public int getOffset() {
+ return offset;
+ }
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
+ public boolean isEnablePushDown() {
+ return enablePushDown;
+ }
+
+ public void setEnablePushDown(boolean enablePushDown) {
+ this.enablePushDown = enablePushDown;
+ }
+
+ public PlanNode getParent() {
+ return parent;
+ }
+
+ public void setParent(PlanNode parent) {
+ this.parent = parent;
+ }
+
+ public Analysis getAnalysis() {
+ return analysis;
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDownOptimizer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDownOptimizer.java
deleted file mode 100644
index 088d38a020..0000000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/LimitOffsetPushDownOptimizer.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.mpp.plan.optimization;
-
-import org.apache.iotdb.db.mpp.common.MPPQueryContext;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanVisitor;
-
-public class LimitOffsetPushDownOptimizer implements PlanOptimizer {
-
- @Override
- public PlanNode optimize(PlanNode plan, MPPQueryContext context) {
- return null;
- }
-
- private static class Rewriter extends PlanVisitor<PlanNode, RewriterContext> {
- @Override
- public PlanNode visitPlan(PlanNode node, RewriterContext context) {
- return null;
- }
- }
-
- private static class RewriterContext {
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/PlanOptimizer.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/PlanOptimizer.java
index 20d6d37a1d..a2b1f860ab 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/PlanOptimizer.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/optimization/PlanOptimizer.java
@@ -19,8 +19,9 @@
package org.apache.iotdb.db.mpp.plan.optimization;
import org.apache.iotdb.db.mpp.common.MPPQueryContext;
+import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
public interface PlanOptimizer {
- PlanNode optimize(PlanNode plan, MPPQueryContext context);
+ PlanNode optimize(PlanNode plan, Analysis analysis, MPPQueryContext context);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
index b3f65eec7c..f55d695b2d 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/LogicalPlanner.java
@@ -50,7 +50,7 @@ public class LogicalPlanner {
.recordPlanCost(LOGICAL_PLANNER, System.nanoTime() - startTime);
for (PlanOptimizer optimizer : optimizers) {
- rootNode = optimizer.optimize(rootNode, context);
+ rootNode = optimizer.optimize(rootNode, analysis, context);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
index 4e1338420b..35c4ce7e22 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/OperatorTreeGenerator.java
@@ -282,6 +282,8 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
}
seriesScanOptionsBuilder.withAllSensors(
context.getAllSensors(seriesPath.getDevice(), seriesPath.getMeasurement()));
+ seriesScanOptionsBuilder.withLimit(node.getLimit());
+ seriesScanOptionsBuilder.withOffset(node.getOffset());
SeriesScanOperator seriesScanOperator =
new SeriesScanOperator(
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
index e52724453c..c67a96ca7a 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/DistributionPlanner.java
@@ -22,7 +22,8 @@ import org.apache.iotdb.db.mpp.common.MPPQueryContext;
import org.apache.iotdb.db.mpp.common.PlanFragmentId;
import org.apache.iotdb.db.mpp.plan.analyze.Analysis;
import org.apache.iotdb.db.mpp.plan.analyze.QueryType;
-import org.apache.iotdb.db.mpp.plan.optimization.LimitOffsetPushDownOptimizer;
+import org.apache.iotdb.db.mpp.plan.optimization.LimitOffsetPushDown;
+import org.apache.iotdb.db.mpp.plan.optimization.PlanOptimizer;
import org.apache.iotdb.db.mpp.plan.planner.IFragmentParallelPlaner;
import org.apache.iotdb.db.mpp.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance;
@@ -33,8 +34,8 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ExchangeNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.sink.FragmentSinkNode;
-import org.apache.iotdb.db.mpp.plan.statement.crud.QueryStatement;
+import java.util.Collections;
import java.util.List;
public class DistributionPlanner {
@@ -42,12 +43,15 @@ public class DistributionPlanner {
private MPPQueryContext context;
private LogicalQueryPlan logicalPlan;
+ private final List<PlanOptimizer> optimizers;
+
private int planFragmentIndex = 0;
public DistributionPlanner(Analysis analysis, LogicalQueryPlan logicalPlan) {
this.analysis = analysis;
this.logicalPlan = logicalPlan;
this.context = logicalPlan.getContext();
+ this.optimizers = Collections.singletonList(new LimitOffsetPushDown());
}
public PlanNode rewriteSource() {
@@ -66,6 +70,15 @@ public class DistributionPlanner {
return adder.visit(root, new NodeGroupContext(context));
}
+ public PlanNode optimize(PlanNode rootWithExchange) {
+ if (analysis.getStatement().isQuery()) {
+ for (PlanOptimizer optimizer : optimizers) {
+ rootWithExchange = optimizer.optimize(rootWithExchange, analysis, context);
+ }
+ }
+ return rootWithExchange;
+ }
+
public SubPlan splitFragment(PlanNode root) {
FragmentBuilder fragmentBuilder = new FragmentBuilder(context);
return fragmentBuilder.splitToSubPlan(root);
@@ -79,14 +92,8 @@ public class DistributionPlanner {
.getRespDatasetHeader()
.setColumnToTsBlockIndexMap(rootWithExchange.getOutputColumnNames());
}
- if (analysis.getStatement() instanceof QueryStatement) {
- QueryStatement queryStatement = (QueryStatement) analysis.getStatement();
- if (!queryStatement.isAggregationQuery()
- && (queryStatement.hasLimit() || queryStatement.hasOffset())) {
- rootWithExchange = new LimitOffsetPushDownOptimizer().optimize(rootWithExchange, context);
- }
- }
- SubPlan subPlan = splitFragment(rootWithExchange);
+ PlanNode optimizedRootWithExchange = optimize(rootWithExchange);
+ SubPlan subPlan = splitFragment(optimizedRootWithExchange);
// Mark the root Fragment of root SubPlan as `root`
subPlan.getPlanFragment().setRoot(true);
List<FragmentInstance> fragmentInstances = planFragmentInstances(subPlan);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
index 03e5028e26..dcf16e1775 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanGraphPrinter.java
@@ -95,6 +95,8 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
boxValue.add(String.format("SeriesScan-%s", node.getPlanNodeId().getId()));
boxValue.add(String.format("Series: %s", node.getSeriesPath()));
boxValue.add(String.format("TimeFilter: %s", node.getTimeFilter()));
+ boxValue.add(String.format("Limit: %s", node.getLimit()));
+ boxValue.add(String.format("Offset: %s", node.getOffset()));
boxValue.add(printRegion(node.getRegionReplicaSet()));
return render(node, boxValue, context);
}
@@ -108,6 +110,8 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
"Series: %s%s",
node.getAlignedPath().getDevice(), node.getAlignedPath().getMeasurementList()));
boxValue.add(String.format("TimeFilter: %s", node.getTimeFilter()));
+ boxValue.add(String.format("Limit: %s", node.getLimit()));
+ boxValue.add(String.format("Offset: %s", node.getOffset()));
boxValue.add(printRegion(node.getRegionReplicaSet()));
return render(node, boxValue, context);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
index 8181efcfa3..5633a6b2cb 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/PlanVisitor.java
@@ -60,8 +60,10 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.HorizontallyConcat
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.IntoNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.LimitNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MergeSortNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.MultiChildProcessNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.OffsetNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.ProjectNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SingleChildProcessNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SingleDeviceViewNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SlidingWindowAggregationNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.process.SortNode;
@@ -78,6 +80,7 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.LastQueryScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesAggregationScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SeriesScanNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.ShowQueriesNode;
+import org.apache.iotdb.db.mpp.plan.planner.plan.node.source.SourceNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.DeleteDataNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertMultiTabletsNode;
import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
@@ -93,76 +96,88 @@ public abstract class PlanVisitor<R, C> {
public abstract R visitPlan(PlanNode node, C context);
- public R visitSeriesScan(SeriesScanNode node, C context) {
+ public R visitSourceNode(SourceNode node, C context) {
return visitPlan(node, context);
}
- public R visitSeriesAggregationScan(SeriesAggregationScanNode node, C context) {
+ public R visitSingleChildProcess(SingleChildProcessNode node, C context) {
return visitPlan(node, context);
}
- public R visitAlignedSeriesScan(AlignedSeriesScanNode node, C context) {
+ public R visitMultiChildProcess(MultiChildProcessNode node, C context) {
return visitPlan(node, context);
}
+ public R visitSeriesScan(SeriesScanNode node, C context) {
+ return visitSourceNode(node, context);
+ }
+
+ public R visitSeriesAggregationScan(SeriesAggregationScanNode node, C context) {
+ return visitSourceNode(node, context);
+ }
+
+ public R visitAlignedSeriesScan(AlignedSeriesScanNode node, C context) {
+ return visitSourceNode(node, context);
+ }
+
public R visitAlignedSeriesAggregationScan(AlignedSeriesAggregationScanNode node, C context) {
- return visitPlan(node, context);
+ return visitSourceNode(node, context);
}
public R visitDeviceView(DeviceViewNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitDeviceMerge(DeviceMergeNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitFill(FillNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitFilter(FilterNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitGroupByLevel(GroupByLevelNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitGroupByTag(GroupByTagNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitSlidingWindowAggregation(SlidingWindowAggregationNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitLimit(LimitNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitOffset(OffsetNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitAggregation(AggregationNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitSort(SortNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitProject(ProjectNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitTimeJoin(TimeJoinNode node, C context) {
- return visitPlan(node, context);
+ return visitMultiChildProcess(node, context);
}
public R visitExchange(ExchangeNode node, C context) {
- return visitPlan(node, context);
+ return visitSingleChildProcess(node, context);
}
public R visitSchemaQueryMerge(SchemaQueryMergeNode node, C context) {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
index 3adf0b0618..f3f95956b6 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/source/AlignedSeriesScanNode.java
@@ -127,6 +127,14 @@ public class AlignedSeriesScanNode extends SeriesSourceNode {
return offset;
}
+ public void setLimit(int limit) {
+ this.limit = limit;
+ }
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
@Override
public void open() throws Exception {}