[IOTDB-6176] Pipe: record SimpleProgressIndex for insertion when usin…

…g SimpleConsensus to avoid wrong advance progress report (#11255)
apache · Oct 10, 2023 · 7c976f1 · 7c976f1
1 parent 18a4d91
commit 7c976f1
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 13 deletions.
diff --git a/...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java b/...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.pipe.agent.PipeAgent;
 import org.apache.iotdb.db.pipe.resource.PipeHardlinkFileDirStartupCleaner;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
 import org.apache.iotdb.db.service.ResourcesInformationHolder;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 
@@ -95,8 +96,8 @@ public ServiceType getID() {
 
   ////////////////////// SimpleConsensus ProgressIndex Assigner //////////////////////
 
-  public void assignSimpleProgressIndexIfNeeded(TsFileResource tsFileResource) {
-    simpleConsensusProgressIndexAssigner.assignIfNeeded(tsFileResource);
+  public void assignSimpleProgressIndexIfNeeded(InsertNode insertNode) {
+    simpleConsensusProgressIndexAssigner.assignIfNeeded(insertNode);
   }
 
   ////////////////////// Recover ProgressIndex Assigner //////////////////////

diff --git a/...ain/java/org/apache/iotdb/db/pipe/agent/runtime/SimpleConsensusProgressIndexAssigner.java b/...ain/java/org/apache/iotdb/db/pipe/agent/runtime/SimpleConsensusProgressIndexAssigner.java
@@ -24,7 +24,7 @@
 import org.apache.iotdb.commons.file.SystemFileFactory;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
 
 import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
@@ -54,7 +54,7 @@ public class SimpleConsensusProgressIndexAssigner {
   private boolean isSimpleConsensusEnable = false;
 
   private int rebootTimes = 0;
-  private final AtomicLong memtableFlushOrderId = new AtomicLong(0);
+  private final AtomicLong insertionRequestId = new AtomicLong(0);
 
   public void start() throws StartupException {
     isSimpleConsensusEnable =
@@ -98,16 +98,16 @@ private void recordRebootTimes() throws IOException {
     FileUtils.writeStringToFile(file, String.valueOf(rebootTimes + 1), StandardCharsets.UTF_8);
   }
 
-  public void assignIfNeeded(TsFileResource tsFileResource) {
+  public void assignIfNeeded(InsertNode insertNode) {
     if (!isSimpleConsensusEnable) {
       return;
     }
 
-    tsFileResource.updateProgressIndex(
-        new SimpleProgressIndex(rebootTimes, memtableFlushOrderId.getAndIncrement()));
+    insertNode.setProgressIndex(
+        new SimpleProgressIndex(rebootTimes, insertionRequestId.getAndIncrement()));
   }
 
   public SimpleProgressIndex getSimpleProgressIndexForTsFileRecovery() {
-    return new SimpleProgressIndex(rebootTimes, memtableFlushOrderId.getAndIncrement());
+    return new SimpleProgressIndex(rebootTimes, insertionRequestId.getAndIncrement());
   }
 }
diff --git a/...va/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java b/...va/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java
@@ -148,10 +148,7 @@ private void extractTsFileInsertion(PipeRealtimeEvent event) {
         break;
       case USING_TABLET:
         // All the tablet events have been extracted, so we can ignore the tsFile event.
-        // Report this event for SimpleProgressIndex, which does not have progressIndex for wal.
-        // This report won't affect IoTProgressIndex since the previous wal events have been
-        // successfully transferred here.
-        event.decreaseReferenceCount(PipeRealtimeDataRegionHybridExtractor.class.getName(), true);
+        event.decreaseReferenceCount(PipeRealtimeDataRegionHybridExtractor.class.getName(), false);
         break;
       default:
         throw new UnsupportedOperationException(

diff --git a/.../src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/.../src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -277,6 +277,7 @@ public void insert(InsertRowNode insertRowNode) throws WriteProcessException {
 
     startTime = System.nanoTime();
 
+    PipeAgent.runtime().assignSimpleProgressIndexIfNeeded(insertRowNode);
     PipeInsertionDataNodeListener.getInstance()
         .listenToInsertNode(
             dataRegionInfo.getDataRegion().getDataRegionId(),
@@ -387,6 +388,7 @@ public void insertTablet(
 
     startTime = System.nanoTime();
 
+    PipeAgent.runtime().assignSimpleProgressIndexIfNeeded(insertTabletNode);
     PipeInsertionDataNodeListener.getInstance()
         .listenToInsertNode(
             dataRegionInfo.getDataRegion().getDataRegionId(),
@@ -896,7 +898,6 @@ public void asyncClose() {
       IMemTable tmpMemTable = workMemTable == null ? new NotifyFlushMemTable() : workMemTable;
 
       try {
-        PipeAgent.runtime().assignSimpleProgressIndexIfNeeded(tsFileResource);
         PipeInsertionDataNodeListener.getInstance()
             .listenToTsFile(
                 dataRegionInfo.getDataRegion().getDataRegionId(), tsFileResource, false);