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/06/11 07:15:09 UTC
(iotdb) 01/03: fix table model write plan
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch TableModelGrammar0606
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1b4f85bcd52d1ee0a4e515f4ffd1f591eac526db
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Jun 11 14:46:15 2024 +0800
fix table model write plan
---
.../TableModelWriteFragmentPlanner.java | 78 ++++++++++++++++++++++
.../distribute/TableDistributionPlanner.java | 5 +-
...or.java => TableModelQueryFragmentPlanner.java} | 9 +--
3 files changed, 87 insertions(+), 5 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/TableModelWriteFragmentPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/TableModelWriteFragmentPlanner.java
new file mode 100644
index 00000000000..9c404c411a2
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/TableModelWriteFragmentPlanner.java
@@ -0,0 +1,78 @@
+/*
+ * 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.commons.partition.StorageExecutor;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.planner.IFragmentParallelPlaner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.SubPlan;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.TableModelTimePredicate;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.tree.Expression;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TableModelWriteFragmentPlanner implements IFragmentParallelPlaner {
+
+ private final SubPlan subPlan;
+ private final Analysis tableAnalysis;
+ private final MPPQueryContext queryContext;
+
+ public TableModelWriteFragmentPlanner(
+ SubPlan subPlan, Analysis tableAnalysis, MPPQueryContext queryContext) {
+ this.subPlan = subPlan;
+ this.tableAnalysis = tableAnalysis;
+ this.queryContext = queryContext;
+ }
+
+ @Override
+ public List<FragmentInstance> parallelPlan() {
+ PlanFragment fragment = subPlan.getPlanFragment();
+ Expression globalTimePredicate = tableAnalysis.getGlobalTableModelTimePredicate();
+ PlanNode node = fragment.getPlanNodeTree();
+ if (!(node instanceof WritePlanNode)) {
+ throw new IllegalArgumentException("PlanNode should be IWritePlanNode in WRITE operation");
+ }
+ List<WritePlanNode> splits = ((WritePlanNode) node).tableModelSplitByPartition(tableAnalysis);
+ List<FragmentInstance> ret = new ArrayList<>();
+ for (WritePlanNode split : splits) {
+ FragmentInstance instance =
+ new FragmentInstance(
+ new PlanFragment(fragment.getId(), split),
+ fragment.getId().genFragmentInstanceId(),
+ globalTimePredicate == null
+ ? null
+ : new TableModelTimePredicate(tableAnalysis.getGlobalTableModelTimePredicate()),
+ queryContext.getQueryType(),
+ queryContext.getTimeOut(),
+ queryContext.getSession());
+ if (split.getRegionReplicaSet() != null) {
+ instance.setExecutorAndHost(new StorageExecutor(split.getRegionReplicaSet()));
+ }
+ ret.add(instance);
+ }
+ return ret;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
index 5511209a3f2..4e821ef7067 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributionPlanner.java
@@ -17,6 +17,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.execution.exchange.sink.DownStreamChannelLocation;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import org.apache.iotdb.db.queryengine.plan.planner.distribution.TableModelWriteFragmentPlanner;
import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
@@ -76,7 +77,9 @@ public class TableDistributionPlanner {
subPlan.getPlanFragment().setRoot(true);
List<FragmentInstance> fragmentInstances =
- new FragmentInstanceGenerator(subPlan, analysis, mppQueryContext).plan();
+ mppQueryContext.getQueryType() == QueryType.READ
+ ? new TableModelQueryFragmentPlanner(subPlan, analysis, mppQueryContext).parallelPlan()
+ : new TableModelWriteFragmentPlanner(subPlan, analysis, mppQueryContext).parallelPlan();
// Only execute this step for READ operation
if (mppQueryContext.getQueryType() == QueryType.READ) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/FragmentInstanceGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelQueryFragmentPlanner.java
similarity index 97%
rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/FragmentInstanceGenerator.java
rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelQueryFragmentPlanner.java
index a919a46e29d..0e7afda310d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/FragmentInstanceGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableModelQueryFragmentPlanner.java
@@ -46,9 +46,10 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-public class FragmentInstanceGenerator {
+public class TableModelQueryFragmentPlanner {
- private static final Logger LOGGER = LoggerFactory.getLogger(FragmentInstanceGenerator.class);
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(TableModelQueryFragmentPlanner.class);
private final SubPlan subPlan;
@@ -67,13 +68,13 @@ public class FragmentInstanceGenerator {
// Record FragmentInstances dispatched to same DataNode
private final Map<TDataNodeLocation, List<FragmentInstance>> dataNodeFIMap = new HashMap<>();
- FragmentInstanceGenerator(SubPlan subPlan, Analysis analysis, MPPQueryContext queryContext) {
+ TableModelQueryFragmentPlanner(SubPlan subPlan, Analysis analysis, MPPQueryContext queryContext) {
this.subPlan = subPlan;
this.analysis = analysis;
this.queryContext = queryContext;
}
- public List<FragmentInstance> plan() {
+ public List<FragmentInstance> parallelPlan() {
prepare();
calculateNodeTopologyBetweenInstance();
return fragmentInstanceList;