You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/07/03 09:43:56 UTC
(iotdb) 02/02: fix stash error, add CollectNode
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/TableModelGrammar_0627
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 91c87f57b2b03cc347d0109c4771a32a6dd1b5e2
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Wed Jul 3 17:43:36 2024 +0800
fix stash error, add CollectNode
---
.../analyzer/StatementAnalyzerFactory.java | 4 +-
.../distribute/DistributedPlanGenerator.java | 33 ++--
.../plan/relational/analyzer/AnalyzerTest.java | 24 +--
.../plan/relational/analyzer/SortTest.java | 192 +++++++++++++++++++++
4 files changed, 213 insertions(+), 40 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
index b777c91584b..122e317a6a8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzerFactory.java
@@ -30,13 +30,13 @@ import static java.util.Objects.requireNonNull;
public class StatementAnalyzerFactory {
private final Metadata metadata;
- private final SqlParser sqlParser;
+ private SqlParser sqlParser;
private final AccessControl accessControl;
public StatementAnalyzerFactory(
Metadata metadata, SqlParser sqlParser, AccessControl accessControl) {
this.metadata = requireNonNull(metadata, "plannerContext is null");
- this.sqlParser = requireNonNull(sqlParser, "sqlParser is null");
+ // this.sqlParser = requireNonNull(sqlParser, "sqlParser is null");
this.accessControl = requireNonNull(accessControl, "accessControl is null");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java
index 458bc53ca1c..56159f8c7a2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/DistributedPlanGenerator.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LimitNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
@@ -94,20 +95,13 @@ public class DistributedPlanGenerator
@Override
public List<PlanNode> visitOutput(OutputNode outputNode, PlanContext context) {
- // TODO only consider the order of IDs
- context.expectedOrderingScheme =
- new OrderingScheme(
- outputNode.getOutputSymbols(),
- outputNode.getOutputSymbols().stream()
- .collect(Collectors.toMap(symbol -> symbol, symbol -> SortOrder.ASC_NULLS_LAST)));
-
List<PlanNode> childrenNodes = outputNode.getChild().accept(this, context);
if (childrenNodes.size() == 1) {
outputNode.setChild(childrenNodes.get(0));
return Collections.singletonList(outputNode);
}
- return connectViaMergeSort(outputNode, childrenNodes);
+ return connectViaCollectSort(outputNode, childrenNodes);
}
@Override
@@ -118,7 +112,7 @@ public class DistributedPlanGenerator
return Collections.singletonList(limitNode);
}
- return connectViaMergeSort(limitNode, childrenNodes);
+ return connectViaCollectSort(limitNode, childrenNodes);
}
@Override
@@ -129,7 +123,7 @@ public class DistributedPlanGenerator
return Collections.singletonList(offsetNode);
}
- return connectViaMergeSort(offsetNode, childrenNodes);
+ return connectViaCollectSort(offsetNode, childrenNodes);
}
@Override
@@ -142,7 +136,7 @@ public class DistributedPlanGenerator
for (Expression expression : projectNode.getAssignments().getMap().values()) {
if (containsDiffFunction(expression)) {
- return connectViaMergeSort(projectNode, childrenNodes);
+ return connectViaCollectSort(projectNode, childrenNodes);
}
}
@@ -193,7 +187,7 @@ public class DistributedPlanGenerator
}
if (containsDiffFunction(filterNode.getPredicate())) {
- return connectViaMergeSort(filterNode, childrenNodes);
+ return connectViaCollectSort(filterNode, childrenNodes);
}
return childrenNodes.stream()
@@ -338,18 +332,11 @@ public class DistributedPlanGenerator
return tableScanNodeList;
}
- private List<PlanNode> connectViaMergeSort(
+ private List<PlanNode> connectViaCollectSort(
SingleChildProcessNode node, List<PlanNode> childrenNodes) {
- OrderingScheme childrenOrderingScheme =
- planNodeOrderingSchemeMap.get(childrenNodes.get(0).getPlanNodeId());
- MergeSortNode mergeSortNode =
- new MergeSortNode(
- queryContext.getQueryId().genPlanNodeId(),
- childrenOrderingScheme,
- node.getOutputSymbols());
- childrenNodes.forEach(mergeSortNode::addChild);
- node.setChild(mergeSortNode);
- planNodeOrderingSchemeMap.put(node.getPlanNodeId(), childrenOrderingScheme);
+ CollectNode collectNode = new CollectNode(queryContext.getQueryId().genPlanNodeId());
+ childrenNodes.forEach(collectNode::addChild);
+ node.setChild(collectNode);
return Collections.singletonList(node);
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
index 06fd8dc24ef..5af376cea95 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java
@@ -85,7 +85,7 @@ public class AnalyzerTest {
private static final NopAccessControl nopAccessControl = new NopAccessControl();
- QueryId queryId = new QueryId("tmp_query");
+ QueryId queryId = new QueryId("test_query");
SessionInfo sessionInfo =
new SessionInfo(
1L,
@@ -246,8 +246,14 @@ public class AnalyzerTest {
TableScanNode tableScanNode = (TableScanNode) mergeSortNode.getChildren().get(1);
assertEquals(4, tableScanNode.getDeviceEntries().size());
assertEquals(
- Arrays.asList(),
- tableScanNode.getDeviceEntries().stream().map(d -> d.getDeviceID().toString()));
+ Arrays.asList(
+ "table1.shanghai.A3.YY",
+ "table1.shanghai.B3.YY",
+ "table1.shenzhen.B1.XX",
+ "table1.shenzhen.B2.ZZ"),
+ tableScanNode.getDeviceEntries().stream()
+ .map(d -> d.getDeviceID().toString())
+ .collect(Collectors.toList()));
}
@Test
@@ -673,18 +679,6 @@ public class AnalyzerTest {
assertEquals(3, offsetNode.getCount());
}
- @Test
- public void sortTest() {
- // when TableScan locates multi regions, use default MergeSortNode
- sql = "SELECT * FROM table1 ";
- context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
- actualAnalysis = analyzeSQL(sql, metadata);
- logicalQueryPlan =
- new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
- .plan(actualAnalysis);
- rootNode = logicalQueryPlan.getRootNode();
- }
-
public static Analysis analyzeSQL(String sql, Metadata metadata) {
try {
SqlParser sqlParser = new SqlParser();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java
new file mode 100644
index 00000000000..002530cb84d
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/SortTest.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.analyzer;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.execution.warnings.WarningCollector;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.ExchangeNode;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.LogicalPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.distribute.TableDistributionPlanner;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.MergeSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+
+import org.junit.Test;
+
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.AnalyzerTest.analyzeSQL;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class SortTest {
+
+ QueryId queryId = new QueryId("test_query");
+ SessionInfo sessionInfo =
+ new SessionInfo(
+ 1L,
+ "iotdb-user",
+ ZoneId.systemDefault(),
+ IoTDBConstant.ClientVersion.V_1_0,
+ "db",
+ IClientSession.SqlDialect.TABLE);
+ Metadata metadata = new TestMatadata();
+ String sql;
+ Analysis actualAnalysis;
+ MPPQueryContext context;
+ LogicalPlanner logicalPlanner;
+ LogicalQueryPlan logicalQueryPlan;
+ PlanNode rootNode;
+ TableDistributionPlanner distributionPlanner;
+ DistributedQueryPlan distributedQueryPlan;
+ TableScanNode tableScanNode;
+
+ // order by time, others, some_ids
+ @Test
+ public void timeOthersSomeIDColumnSortTest() {
+ sql = "SELECT * FROM table1 order by time desc, s1+s2 asc, tag2 asc, tag1 desc";
+ context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
+ actualAnalysis = analyzeSQL(sql, metadata);
+ logicalQueryPlan =
+ new LogicalPlanner(context, metadata, sessionInfo, WarningCollector.NOOP)
+ .plan(actualAnalysis);
+ rootNode = logicalQueryPlan.getRootNode();
+
+ assertTrue(rootNode instanceof OutputNode);
+ assertTrue(rootNode.getChildren().get(0) instanceof SortNode);
+ assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof ProjectNode);
+ assertTrue(
+ rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0)
+ instanceof TableScanNode);
+ tableScanNode =
+ (TableScanNode) rootNode.getChildren().get(0).getChildren().get(0).getChildren().get(0);
+ assertEquals("testdb.table1", tableScanNode.getQualifiedObjectName().toString());
+ assertEquals(
+ Arrays.asList("time", "tag1", "tag2", "tag3", "attr1", "attr2", "s1", "s2", "s3"),
+ tableScanNode.getOutputColumnNames());
+ assertEquals(9, tableScanNode.getAssignments().size());
+ assertEquals(6, tableScanNode.getDeviceEntries().size());
+ assertEquals(5, tableScanNode.getIdAndAttributeIndexMap().size());
+
+ distributionPlanner = new TableDistributionPlanner(actualAnalysis, logicalQueryPlan, context);
+ distributedQueryPlan = distributionPlanner.plan();
+ assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertTrue(
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0)
+ instanceof OutputNode);
+ OutputNode outputNode =
+ (OutputNode)
+ distributedQueryPlan.getFragments().get(0).getPlanNodeTree().getChildren().get(0);
+ assertTrue(outputNode.getChildren().get(0) instanceof MergeSortNode);
+ MergeSortNode mergeSortNode = (MergeSortNode) outputNode.getChildren().get(0);
+ assertEquals(
+ Arrays.asList("tag1", "tag2", "tag3", "attr1", "attr2"),
+ mergeSortNode.getOrderingScheme().getOrderBy().stream()
+ .map(Symbol::getName)
+ .collect(Collectors.toList()));
+ assertTrue(mergeSortNode.getChildren().get(0) instanceof ExchangeNode);
+ assertTrue(mergeSortNode.getChildren().get(1) instanceof TableScanNode);
+ assertTrue(mergeSortNode.getChildren().get(2) instanceof ExchangeNode);
+ TableScanNode tableScanNode = (TableScanNode) mergeSortNode.getChildren().get(1);
+ assertEquals(4, tableScanNode.getDeviceEntries().size());
+ assertEquals(
+ Arrays.asList(),
+ tableScanNode.getDeviceEntries().stream().map(d -> d.getDeviceID().toString()));
+ }
+
+ // order by time, others, all_ids
+ @Test
+ public void timeOthersAllIDColumnSortTest() {
+ // TODO
+ }
+
+ // order by time, some_ids, others
+ @Test
+ public void timeSomeIDColumnOthersSortTest() {
+ // TODO
+ }
+
+ // order by time, all_ids, others
+ @Test
+ public void timeAllIDColumnOthersSortTest() {
+ // TODO
+ }
+
+ // order by some_ids, time, others
+ @Test
+ public void someIDColumnTimeOthersSortTest() {
+ // TODO
+ }
+
+ // order by all_ids, time, others
+ @Test
+ public void allIDColumnTimeOthersSortTest() {
+ // TODO
+ }
+
+ // order by some_ids, others, time
+ @Test
+ public void someIDColumnOthersTimeSortTest() {
+ // TODO
+ }
+
+ // order by all_ids, others, time
+ @Test
+ public void allIDColumnOthersTimeSortTest() {
+ // TODO
+ }
+
+ // order by others, some_ids, time
+ @Test
+ public void othersSomeIDColumnTimeSortTest() {
+ // TODO
+ }
+
+ // order by others, all_ids, time
+ @Test
+ public void othersAllIDColumnTimeSortTest() {
+ // TODO
+ }
+
+ // order by others, time, some_ids
+ @Test
+ public void othersTimeSomeIDColumnSortTest() {
+ // TODO
+ }
+
+ // order by others, time, all_ids
+ @Test
+ public void othersTimeAllIDColumnSortTest() {
+ // TODO
+ }
+}