This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/mpp_issues
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/xingtanzjr/mpp_issues by this
push:
new 563fbe2218 fix the issue that distribution planner cannot give correct
result for SeriesScanNode
563fbe2218 is described below
commit 563fbe2218c91fdc5178a326aa234aad5c7dd486
Author: Jinrui.Zhang <[email protected]>
AuthorDate: Mon Apr 18 16:04:52 2022 +0800
fix the issue that distribution planner cannot give correct result for
SeriesScanNode
---
.../iotdb/db/mpp/execution/QueryExecution.java | 4 +--
.../db/mpp/sql/planner/DistributionPlanner.java | 20 +++++++++++
.../db/mpp/sql/plan/DistributionPlannerTest.java | 39 ++++++++++++++++++++++
3 files changed, 60 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
index 6e991039fe..172a82db70 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryExecution.java
@@ -97,8 +97,6 @@ public class QueryExecution implements IQueryExecution {
this.stateMachine = new QueryStateMachine(context.getQueryId(), executor);
this.partitionFetcher = partitionFetcher;
this.schemaFetcher = schemaFetcher;
- // TODO: (xingtanzjr) Initialize the result handle after the
DataBlockManager is merged.
- // resultHandle = xxxx
// We add the abort logic inside the QueryExecution.
// So that the other components can only focus on the state change.
@@ -115,6 +113,7 @@ public class QueryExecution implements IQueryExecution {
doLogicalPlan();
doDistributedPlan();
if (context.getQueryType() == QueryType.READ) {
+ // The ResultHandle could only be initialized after distributed planning
initResultHandle();
}
schedule();
@@ -140,7 +139,6 @@ public class QueryExecution implements IQueryExecution {
context.getQueryType(),
executor,
scheduledExecutor);
- // TODO: (xingtanzjr) how to make the schedule running asynchronously
this.scheduler.start();
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
index ca42ec2336..83b5e950d6 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/DistributionPlanner.java
@@ -174,6 +174,26 @@ public class DistributionPlanner {
return root;
}
+ // TODO: (xingtanzjr) a temporary way to resolve the distribution of
single SeriesScanNode issue
+ @Override
+ public PlanNode visitSeriesScan(SeriesScanNode node,
DistributionPlanContext context) {
+ List<RegionReplicaSet> dataDistribution =
+ analysis.getPartitionInfo(node.getSeriesPath(),
node.getTimeFilter());
+ if (dataDistribution.size() == 1) {
+ node.setRegionReplicaSet(dataDistribution.get(0));
+ return node;
+ }
+ TimeJoinNode timeJoinNode =
+ new TimeJoinNode(context.queryContext.getQueryId().genPlanNodeId(),
node.getScanOrder());
+ for (RegionReplicaSet dataRegion : dataDistribution) {
+ SeriesScanNode split = (SeriesScanNode) node.clone();
+ split.setPlanNodeId(context.queryContext.getQueryId().genPlanNodeId());
+ split.setRegionReplicaSet(dataRegion);
+ timeJoinNode.addChild(split);
+ }
+ return timeJoinNode;
+ }
+
@Override
public PlanNode visitTimeJoin(TimeJoinNode node, DistributionPlanContext
context) {
TimeJoinNode root = (TimeJoinNode) node.clone();
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
index 21ca0ed2d1..78ffd4bb40 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/sql/plan/DistributionPlannerTest.java
@@ -68,6 +68,45 @@ import static org.junit.Assert.assertEquals;
public class DistributionPlannerTest {
+ @Test
+ public void TestSingleSeriesScan() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ SeriesScanNode root =
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ Sets.newHashSet("s1", "s2"),
+ OrderBy.TIMESTAMP_ASC);
+
+ Analysis analysis = constructAnalysis();
+
+ MPPQueryContext context = new MPPQueryContext("", queryId, null, new
Endpoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ DistributedQueryPlan plan = planner.planFragments();
+ plan.getInstances().forEach(System.out::println);
+ assertEquals(2, plan.getInstances().size());
+ }
+
+ @Test
+ public void TestSingleSeriesScanRewriteSource() throws IllegalPathException {
+ QueryId queryId = new QueryId("test_query");
+ SeriesScanNode root =
+ new SeriesScanNode(
+ queryId.genPlanNodeId(),
+ new MeasurementPath("root.sg.d1.s1", TSDataType.INT32),
+ Sets.newHashSet("s1", "s2"),
+ OrderBy.TIMESTAMP_ASC);
+
+ Analysis analysis = constructAnalysis();
+
+ MPPQueryContext context = new MPPQueryContext("", queryId, null, new
Endpoint());
+ DistributionPlanner planner =
+ new DistributionPlanner(analysis, new LogicalQueryPlan(context, root));
+ PlanNode rootAfterRewrite = planner.rewriteSource();
+ assertEquals(2, rootAfterRewrite.getChildren().size());
+ }
+
@Test
public void TestRewriteSourceNode() throws IllegalPathException {
QueryId queryId = new QueryId("test_query");
