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:11 UTC
(iotdb) 03/03: fit table model write process
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 344301e9008df163bc8f28ab7f833c599ece4508
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Jun 11 15:14:53 2024 +0800
fit table model write process
---
.../TableModelWriteFragmentPlanner.java | 78 ----------------------
.../distribution/WriteFragmentParallelPlanner.java | 3 +-
.../distribute/TableDistributionPlanner.java | 4 +-
3 files changed, 3 insertions(+), 82 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
deleted file mode 100644
index 9c404c411a2..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/TableModelWriteFragmentPlanner.java
+++ /dev/null
@@ -1,78 +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.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/planner/distribution/WriteFragmentParallelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/WriteFragmentParallelPlanner.java
index 9bc526ee4f5..b1d1fe80123 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/WriteFragmentParallelPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/WriteFragmentParallelPlanner.java
@@ -21,7 +21,6 @@ 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.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
import org.apache.iotdb.db.queryengine.plan.planner.IFragmentParallelPlaner;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
@@ -40,7 +39,7 @@ public class WriteFragmentParallelPlanner implements IFragmentParallelPlaner {
private MPPQueryContext queryContext;
public WriteFragmentParallelPlanner(
- SubPlan subPlan, Analysis analysis, MPPQueryContext queryContext) {
+ SubPlan subPlan, IAnalysis analysis, MPPQueryContext queryContext) {
this.subPlan = subPlan;
this.analysis = analysis;
this.queryContext = queryContext;
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 4e821ef7067..59404c03b07 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,7 +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.distribution.WriteFragmentParallelPlanner;
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;
@@ -79,7 +79,7 @@ public class TableDistributionPlanner {
List<FragmentInstance> fragmentInstances =
mppQueryContext.getQueryType() == QueryType.READ
? new TableModelQueryFragmentPlanner(subPlan, analysis, mppQueryContext).parallelPlan()
- : new TableModelWriteFragmentPlanner(subPlan, analysis, mppQueryContext).parallelPlan();
+ : new WriteFragmentParallelPlanner(subPlan, analysis, mppQueryContext).parallelPlan();
// Only execute this step for READ operation
if (mppQueryContext.getQueryType() == QueryType.READ) {