You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2023/11/13 02:21:16 UTC
(iotdb) branch load_v2 updated: enable consensus in load
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch load_v2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/load_v2 by this push:
new 309b324b71e enable consensus in load
309b324b71e is described below
commit 309b324b71e323b63b986bf63fb1f95826f56201
Author: Tian Jiang <jt...@163.com>
AuthorDate: Mon Nov 13 10:25:35 2023 +0800
enable consensus in load
---
.../iotdb/db/queryengine/execution/load/LoadTsFileManager.java | 1 +
.../iotdb/db/queryengine/execution/load/TsFileSplitSender.java | 10 ++++++----
.../plan/planner/plan/node/load/LoadCommandNode.java | 6 ++++++
3 files changed, 13 insertions(+), 4 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
index 85c1742f424..baf7b221e82 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/LoadTsFileManager.java
@@ -237,6 +237,7 @@ public class LoadTsFileManager {
this.dataPartition2LastDevice = new HashMap<>();
this.isClosed = false;
this.receivedSplitIds = new HashSet<>();
+ this.loadedPartitionNum = new AtomicInteger();
clearDir(taskDir);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSender.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSender.java
index c843ed1e22a..b09feea35f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSender.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/load/TsFileSplitSender.java
@@ -255,15 +255,17 @@ public class TsFileSplitSender {
}
private boolean secondPhase(boolean isFirstPhaseSuccess) {
- TLoadCommandReq loadCommandReq =
- new TLoadCommandReq(
- (isFirstPhaseSuccess ? LoadCommand.EXECUTE : LoadCommand.ROLLBACK).ordinal(), uuid);
- loadCommandReq.setIsGeneratedByPipe(isGeneratedByPipe);
long p2StartMS = System.currentTimeMillis();
List<Pair<TRegionReplicaSet, Future<Void>>> loadFutures = new ArrayList<>();
AtomicBoolean hasTimeout = new AtomicBoolean();
for (TRegionReplicaSet replicaSet : allReplicaSets) {
+ TLoadCommandReq loadCommandReq =
+ new TLoadCommandReq(
+ (isFirstPhaseSuccess ? LoadCommand.EXECUTE : LoadCommand.ROLLBACK).ordinal(), uuid);
+ loadCommandReq.setIsGeneratedByPipe(isGeneratedByPipe);
+ loadCommandReq.setUseConsensus(true);
+ loadCommandReq.setConsensusGroupId(replicaSet.getRegionId());
loadFutures.add(
new Pair<>(
replicaSet,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadCommandNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadCommandNode.java
index fba81478899..3d42a401b70 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadCommandNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadCommandNode.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler.LoadCommand;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
@@ -135,4 +136,9 @@ public class LoadCommandNode extends PlanNode {
public boolean isGeneratedByPipe() {
return isGeneratedByPipe;
}
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitLoadCommand(this, context);
+ }
}