You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/10/25 07:25:28 UTC
[iotdb] branch fix-progress-index-recover created (now 181e3bc01b0)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch fix-progress-index-recover
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 181e3bc01b0 Pipe: assign progress index for sealed tsfile recovery
This branch includes the following new commits:
new 181e3bc01b0 Pipe: assign progress index for sealed tsfile recovery
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Pipe: assign progress index for sealed tsfile recovery
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch fix-progress-index-recover
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 181e3bc01b09e6c99ca9b72d007c1391cb7719c8
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Oct 25 15:25:19 2023 +0800
Pipe: assign progress index for sealed tsfile recovery
---
.../org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java | 6 +++---
.../plan/planner/plan/node/load/LoadSingleTsFileNode.java | 2 +-
.../iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java | 2 +-
.../dataregion/wal/recover/file/AbstractTsFileRecoverPerformer.java | 5 +++++
.../dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java | 2 +-
.../src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java | 2 +-
6 files changed, 12 insertions(+), 7 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
index dc50ba94bee..f7a2a5595f7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
@@ -102,14 +102,14 @@ public class PipeRuntimeAgent implements IService {
////////////////////// Recover ProgressIndex Assigner //////////////////////
- public void assignRecoverProgressIndexForTsFileRecovery(TsFileResource tsFileResource) {
- tsFileResource.recoverProgressIndex(
+ public void assignProgressIndexForTsFileLoad(TsFileResource tsFileResource) {
+ tsFileResource.setProgressIndex(
new RecoverProgressIndex(
DATA_NODE_ID,
simpleConsensusProgressIndexAssigner.getSimpleProgressIndexForTsFileRecovery()));
}
- public void assignUpdateProgressIndexForTsFileRecovery(TsFileResource tsFileResource) {
+ public void assignProgressIndexForTsFileRecovery(TsFileResource tsFileResource) {
tsFileResource.updateProgressIndex(
new RecoverProgressIndex(
DATA_NODE_ID,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
index 687a652c363..897522afa04 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
@@ -100,7 +100,7 @@ public class LoadSingleTsFileNode extends WritePlanNode {
needDecodeTsFile = !isDispatchedToLocal(new HashSet<>(partitionFetcher.apply(slotList)));
}
- PipeAgent.runtime().assignRecoverProgressIndexForTsFileRecovery(resource);
+ PipeAgent.runtime().assignProgressIndexForTsFileLoad(resource);
// we serialize the resource file even if the tsfile does not need to be decoded
// or the resource file is already existed because we need to serialize the
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index ce49b700768..4aa7f546df3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -1127,7 +1127,7 @@ public class TsFileResource {
: maxProgressIndex.updateToMinimumIsAfterProgressIndex(progressIndex));
}
- public void recoverProgressIndex(ProgressIndex progressIndex) {
+ public void setProgressIndex(ProgressIndex progressIndex) {
if (progressIndex == null) {
return;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/AbstractTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/AbstractTsFileRecoverPerformer.java
index fbaba378da1..784a8c18045 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/AbstractTsFileRecoverPerformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/AbstractTsFileRecoverPerformer.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file;
import org.apache.iotdb.db.exception.DataRegionException;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.tsfile.exception.NotCompatibleTsFileException;
@@ -117,6 +118,10 @@ public abstract class AbstractTsFileRecoverPerformer implements Closeable {
new TsFileSequenceReader(tsFileResource.getTsFile().getAbsolutePath())) {
FileLoaderUtils.updateTsFileResource(reader, tsFileResource);
}
+
+ // set progress index for pipe to avoid data loss
+ PipeAgent.runtime().assignProgressIndexForTsFileRecovery(tsFileResource);
+
tsFileResource.serialize();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
index 9c883889143..b0aaa8b1378 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
@@ -243,7 +243,7 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
}
// set recover progress index for pipe
- PipeAgent.runtime().assignUpdateProgressIndexForTsFileRecovery(tsFileResource);
+ PipeAgent.runtime().assignProgressIndexForTsFileRecovery(tsFileResource);
// if we put following codes in the 'if' clause above, this file can be continued writing
// into it
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
index 2d6d53809da..eccb532e8df 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
@@ -108,7 +108,7 @@ public class FileLoaderUtils {
}
}
resource.setStatus(TsFileResourceStatus.NORMAL);
- PipeAgent.runtime().assignRecoverProgressIndexForTsFileRecovery(resource);
+ PipeAgent.runtime().assignProgressIndexForTsFileLoad(resource);
return resource;
}