[Pipe] Support region-level realtime downgrading

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java

@@ -262,6 +262,13 @@ private void collectEvent(final Event event) {
 
     if (pendingQueue.offer(event)) {
       collectInvocationCount.incrementAndGet();
+      if (event instanceof PipeRawTabletInsertionEvent
+          && ((PipeRawTabletInsertionEvent) event).getSourceEvent()
+              instanceof PipeTsFileInsertionEvent) {
+        ((PipeTsFileInsertionEvent) ((PipeRawTabletInsertionEvent) event).getSourceEvent())
+            .registerGeneratedTabletInsertionEvent();
+      }
+      return;
     }
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java

@@ -118,12 +118,18 @@ private PipeRawTabletInsertionEvent(
     this.allocatedMemoryBlock =
         PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
 
-    if (needToReport) {
+    if (needToReport || sourceEvent instanceof PipeTsFileInsertionEvent) {
       addOnCommittedHook(
           () -> {
-            if (shouldReportOnCommit) {
+            // The last generated tablet still owns the progress-index cleanup. Other generated
+            // tablets only contribute to the transfer completion of their source TsFile.
+            if (shouldReportOnCommit && needToReport) {
               eliminateProgressIndex();
             }
+            if (sourceEvent instanceof PipeTsFileInsertionEvent) {
+              ((PipeTsFileInsertionEvent) sourceEvent)
+                  .markGeneratedTabletInsertionEventAsTransferred();
+            }
           });
     }
   }
@@ -308,7 +314,8 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
 
   protected void eliminateProgressIndex() {
     if (sourceEvent instanceof PipeTsFileInsertionEvent) {
-      ((PipeTsFileInsertionEvent) sourceEvent).eliminateProgressIndex();
+      final PipeTsFileInsertionEvent tsFileInsertionEvent = (PipeTsFileInsertionEvent) sourceEvent;
+      tsFileInsertionEvent.eliminateProgressIndex();
     }
   }
 
@@ -391,7 +398,7 @@ public boolean mayEventPathsOverlappedWithPattern() {
   }
 
   public void markAsNeedToReport() {
-    if (!needToReport) {
+    if (!needToReport && !(sourceEvent instanceof PipeTsFileInsertionEvent)) {
       addOnCommittedHook(
           () -> {
             if (shouldReportOnCommit) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -109,6 +109,13 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
   private final AtomicInteger parsedTabletInsertionEventCount = new AtomicInteger(0);
   private final AtomicBoolean isTsFileParsingCompleted = new AtomicBoolean(false);
   private final AtomicLong parsedPointCountForCount = new AtomicLong(0);
+  private final AtomicInteger generatedTabletInsertionEventCount = new AtomicInteger(0);
+  private final AtomicInteger transferredGeneratedTabletInsertionEventCount = new AtomicInteger(0);
+  private final AtomicBoolean generatedTabletInsertionEventsParsingCompleted =
+      new AtomicBoolean(false);
+  private final AtomicBoolean isTsFileEventCommitted = new AtomicBoolean(false);
+  private final AtomicBoolean isTransferred = new AtomicBoolean(false);
+  private final List<Runnable> onTransferredHooks = new ArrayList<>();
 
   // The point count of the TsFile. Used for metrics on IoTConsensusV2' receiver side.
   // May be updated after it is flushed. Should be negative if not set.
@@ -295,6 +302,7 @@ private PipeTsFileInsertionEvent(
           if (shouldReportOnCommit) {
             eliminateProgressIndex();
           }
+          markAsTransferredOnCommit();
         });
   }
 
@@ -520,6 +528,75 @@ public void eliminateProgressIndex() {
     }
   }
 
+  /**
+   * Records a tablet generated from this TsFile that has been accepted by the processor output
+   * collector. The transfer hook must wait for all such tablets, rather than the first one, before
+   * allowing the region-level downgrade to end.
+   */
+  public void registerGeneratedTabletInsertionEvent() {
+    generatedTabletInsertionEventCount.incrementAndGet();
+  }
+
+  /** Records that one tablet generated from this TsFile has been committed downstream. */
+  public void markGeneratedTabletInsertionEventAsTransferred() {
+    transferredGeneratedTabletInsertionEventCount.incrementAndGet();
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  /** Marks the end of tablet generation for this TsFile. */
+  public void markGeneratedTabletInsertionEventsParsingCompleted() {
+    generatedTabletInsertionEventsParsingCompleted.set(true);
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  private void markAsTransferredIfGeneratedTabletEventsCompleted() {
+    if ((generatedTabletInsertionEventsParsingCompleted.get() || isTsFileEventCommitted.get())
+        && generatedTabletInsertionEventCount.get() > 0
+        && transferredGeneratedTabletInsertionEventCount.get()
+            >= generatedTabletInsertionEventCount.get()) {
+      markAsTransferred();
+    }
+  }
+
+  private void markAsTransferredOnCommit() {
+    // The default PipeProcessor implementation iterates over toTabletInsertionEvents() directly,
+    // so it cannot signal parser completion through consumeTabletInsertionEventsWithRetry(). Its
+    // source event is committed only after processor execution has returned, which makes this a
+    // reliable completion boundary for the generated-event counters as well.
+    isTsFileEventCommitted.set(true);
+    if (generatedTabletInsertionEventCount.get() == 0) {
+      markAsTransferred();
+      return;
+    }
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  private void markAsTransferred() {
+    final List<Runnable> hooksToRun;
+    synchronized (onTransferredHooks) {
+      if (!isTransferred.compareAndSet(false, true)) {
+        return;
+      }
+      hooksToRun = new ArrayList<>(onTransferredHooks);
+      onTransferredHooks.clear();
+    }
+    hooksToRun.forEach(Runnable::run);
+  }
+
+  /**
+   * Adds a hook that is invoked after this TsFile, or the last tablet generated from it, is
+   * committed downstream.
+   */
+  public void addOnTransferredHook(final Runnable hook) {
+    synchronized (onTransferredHooks) {
+      if (!isTransferred.get()) {
+        onTransferredHooks.add(hook);
+        return;
+      }
+    }
+    hook.run();
+  }
+
   public PipeTsFileInsertionEvent bindTsFileDedupScopeID(final String tsFileDedupScopeID) {
     this.tsFileDedupScopeID = tsFileDedupScopeID;
     return this;
@@ -776,6 +853,7 @@ public void consumeTabletInsertionEventsWithRetry(
             getNextTabletInsertionEventFromSavedProgress();
         if (parsedEvent == null) {
           isTsFileParsingCompleted.set(true);
+          markGeneratedTabletInsertionEventsParsingCompleted();
           releaseTsFileParserMemoryIfReserved();
           return;
         }