[Pipe] Support region-level realtime downgrading

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java

@@ -33,6 +33,11 @@ public interface TsFileInsertionEvent extends Event, AutoCloseable {
    * The method is used to convert the {@link TsFileInsertionEvent} into several {@link
    * TabletInsertionEvent}s.
    *
+   * <p>The returned iterable represents the lifetime of the conversion. Processors that retain it
+   * for asynchronous work must eventually exhaust the iterable before the event can be considered
+   * complete. A processor should not retain the iterable and silently abandon it, because the
+   * source cannot otherwise determine when all generated tablet events have finished.
+   *
    * @return {@code Iterable<TabletInsertionEvent>} the list of {@link TabletInsertionEvent}
    */
   Iterable<TabletInsertionEvent> toTabletInsertionEvents();

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

@@ -273,8 +273,18 @@ private void collectEvent(final Event event) {
       ((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue);
     }
 
+    if (event instanceof PipeRawTabletInsertionEvent
+        && ((PipeRawTabletInsertionEvent) event).getSourceEvent()
+            instanceof PipeTsFileInsertionEvent) {
+      // Register before publishing to the queue so a concurrent queue cleanup can report a
+      // generated tablet as discarded instead of racing with registration.
+      ((PipeTsFileInsertionEvent) ((PipeRawTabletInsertionEvent) event).getSourceEvent())
+          .registerGeneratedTabletInsertionEvent((PipeRawTabletInsertionEvent) event);
+    }
+
     if (pendingQueue.offer(event)) {
       collectInvocationCount.incrementAndGet();
+      return;
     }
   }
 

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

@@ -64,6 +64,8 @@ public class PipeRawTabletInsertionEvent extends PipeInsertionEvent
   private final boolean isAligned;
 
   private final EnrichedEvent sourceEvent;
+  private final AtomicBoolean generatedEventRegisteredWithSource = new AtomicBoolean(false);
+  private final AtomicBoolean generatedEventOutcomeReported = new AtomicBoolean(false);
   private boolean needToReport;
 
   private final PipeTabletMemoryBlock allocatedMemoryBlock;
@@ -118,12 +120,19 @@ 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
+                && generatedEventOutcomeReported.compareAndSet(false, true)) {
+              ((PipeTsFileInsertionEvent) sourceEvent)
+                  .markGeneratedTabletInsertionEventAsTransferred();
+            }
           });
     }
   }
@@ -306,9 +315,26 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     return true;
   }
 
+  /** Marks this generated tablet as accepted by the processor output collector. */
+  public void markAsGeneratedEventRegisteredWithSource() {
+    generatedEventRegisteredWithSource.set(true);
+  }
+
+  @Override
+  public boolean clearReferenceCount(final String holderMessage) {
+    final boolean cleared = super.clearReferenceCount(holderMessage);
+    if (generatedEventRegisteredWithSource.get()
+        && generatedEventOutcomeReported.compareAndSet(false, true)
+        && sourceEvent instanceof PipeTsFileInsertionEvent) {
+      ((PipeTsFileInsertionEvent) sourceEvent).markGeneratedTabletInsertionEventAsDiscarded();
+    }
+    return cleared;
+  }
+
   protected void eliminateProgressIndex() {
     if (sourceEvent instanceof PipeTsFileInsertionEvent) {
-      ((PipeTsFileInsertionEvent) sourceEvent).eliminateProgressIndex();
+      final PipeTsFileInsertionEvent tsFileInsertionEvent = (PipeTsFileInsertionEvent) sourceEvent;
+      tsFileInsertionEvent.eliminateProgressIndex();
     }
   }
 
@@ -391,7 +417,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

@@ -111,6 +111,18 @@ 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 AtomicInteger discardedGeneratedTabletInsertionEventCount = new AtomicInteger(0);
+  private final AtomicBoolean generatedTabletInsertionEventsParsingStarted =
+      new AtomicBoolean(false);
+  private final AtomicBoolean generatedTabletInsertionEventsParsingCompleted =
+      new AtomicBoolean(false);
+  private final AtomicBoolean isTsFileEventCommitted = new AtomicBoolean(false);
+  private final AtomicBoolean isTransferred = new AtomicBoolean(false);
+  private final AtomicBoolean isDiscarded = new AtomicBoolean(false);
+  private final List<Runnable> onTransferredHooks = new ArrayList<>();
+  private final List<Runnable> onDiscardedHooks = 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.
@@ -298,6 +310,7 @@ private PipeTsFileInsertionEvent(
           if (shouldReportOnCommit) {
             eliminateProgressIndex();
           }
+          markAsTransferredOnCommit();
         });
   }
 
@@ -477,6 +490,16 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     }
   }
 
+  @Override
+  public boolean clearReferenceCount(final String holderMessage) {
+    final boolean cleared = super.clearReferenceCount(holderMessage);
+    // clearReferenceCount intentionally bypasses the commit queue. Notify the realtime source even
+    // when the event was already released by a previous holder: its commit hook may still be
+    // waiting behind an earlier commit, while the explicit discard must release the in-flight slot.
+    markAsDiscarded();
+    return cleared;
+  }
+
   @Override
   public void bindProgressIndex(final ProgressIndex overridingProgressIndex) {
     this.overridingProgressIndex = overridingProgressIndex;
@@ -532,6 +555,133 @@ 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();
+  }
+
+  public void registerGeneratedTabletInsertionEvent(final PipeRawTabletInsertionEvent event) {
+    event.markAsGeneratedEventRegisteredWithSource();
+    registerGeneratedTabletInsertionEvent();
+  }
+
+  /** Marks that a processor has obtained the tablet iterable and may consume it later. */
+  public void markGeneratedTabletInsertionEventsParsingStarted() {
+    generatedTabletInsertionEventsParsingStarted.set(true);
+  }
+
+  /** Records that one tablet generated from this TsFile has been committed downstream. */
+  public void markGeneratedTabletInsertionEventAsTransferred() {
+    transferredGeneratedTabletInsertionEventCount.incrementAndGet();
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  /** Records that one generated tablet was discarded before it could be committed downstream. */
+  public void markGeneratedTabletInsertionEventAsDiscarded() {
+    discardedGeneratedTabletInsertionEventCount.incrementAndGet();
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  /** Marks the end of tablet generation for this TsFile. */
+  public void markGeneratedTabletInsertionEventsParsingCompleted() {
+    generatedTabletInsertionEventsParsingCompleted.set(true);
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  private void markAsTransferredIfGeneratedTabletEventsCompleted() {
+    final boolean generationBoundaryReached =
+        !generatedTabletInsertionEventsParsingStarted.get()
+            || generatedTabletInsertionEventsParsingCompleted.get();
+    // The normal processor path does not retain a tablet iterable unless parsing has explicitly
+    // been marked as started.  In that path, registering the generated tablets and reaching the
+    // generation boundary is sufficient to establish the transfer boundary (the source TsFile
+    // commit may be reported separately or may be intentionally skipped).  Once an iterable has
+    // been retained, however, the source TsFile commit must also be observed before the generated
+    // tablet counts can release the TsFile.  This distinction keeps deferred parsing safe without
+    // delaying the established non-deferred path.
+    if ((isTsFileEventCommitted.get() || !generatedTabletInsertionEventsParsingStarted.get())
+        && generationBoundaryReached
+        && transferredGeneratedTabletInsertionEventCount.get()
+                + discardedGeneratedTabletInsertionEventCount.get()
+            >= generatedTabletInsertionEventCount.get()) {
+      markAsTransferred();
+    }
+  }
+
+  private void markAsTransferredOnCommit() {
+    // A processor may retain the iterable returned by toTabletInsertionEvents() and consume it
+    // after process() returns. In that case the iterable wrapper keeps the generation boundary open
+    // until its iterator is exhausted. If no processor ever requests the iterable (for example,
+    // DoNothingProcessor), committing the TsFile itself is the completion boundary.
+    isTsFileEventCommitted.set(true);
+    markAsTransferredIfGeneratedTabletEventsCompleted();
+  }
+
+  private void markAsTransferred() {
+    final List<Runnable> hooksToRun;
+    synchronized (onTransferredHooks) {
+      if (isDiscarded.get() || !isTransferred.compareAndSet(false, true)) {
+        return;
+      }
+      hooksToRun = new ArrayList<>(onTransferredHooks);
+      onTransferredHooks.clear();
+      onDiscardedHooks.clear();
+    }
+    hooksToRun.forEach(Runnable::run);
+  }
+
+  private void markAsDiscarded() {
+    final List<Runnable> hooksToRun;
+    synchronized (onTransferredHooks) {
+      if (isTransferred.get() || !isDiscarded.compareAndSet(false, true)) {
+        return;
+      }
+      hooksToRun = new ArrayList<>(onDiscardedHooks);
+      onDiscardedHooks.clear();
+      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) {
+    boolean runHook = false;
+    synchronized (onTransferredHooks) {
+      if (!isTransferred.get() && !isDiscarded.get()) {
+        onTransferredHooks.add(hook);
+        return;
+      }
+      runHook = isTransferred.get();
+    }
+    if (runHook) {
+      hook.run();
+    }
+  }
+
+  /**
+   * Adds a hook that runs when this TsFile event is explicitly discarded via clearReferenceCount.
+   */
+  public void addOnDiscardedHook(final Runnable hook) {
+    boolean runHook = false;
+    synchronized (onTransferredHooks) {
+      if (!isTransferred.get() && !isDiscarded.get()) {
+        onDiscardedHooks.add(hook);
+        return;
+      }
+      runHook = isDiscarded.get();
+    }
+    if (runHook) {
+      hook.run();
+    }
+  }
+
   public PipeTsFileInsertionEvent bindTsFileDedupScopeID(final String tsFileDedupScopeID) {
     this.tsFileDedupScopeID = tsFileDedupScopeID;
     return this;
@@ -802,13 +952,15 @@ public void consumeTabletInsertionEventsWithRetry(
       final String callerName,
       final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
       throws Exception {
+    markGeneratedTabletInsertionEventsParsingStarted();
     try {
       while (true) {
         processorExecutionGuard.check();
         final PipeRawTabletInsertionEvent parsedEvent =
             getNextTabletInsertionEventFromSavedProgress(processorExecutionGuard);
         if (parsedEvent == null) {
           isTsFileParsingCompleted.set(true);
+          markGeneratedTabletInsertionEventsParsingCompleted();
           releaseTsFileParserMemoryIfReserved();
           return;
         }
@@ -981,14 +1133,18 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents() throws PipeExcep
 
   public Iterable<TabletInsertionEvent> toTabletInsertionEvents(final long timeoutMs)
       throws PipeException {
+    markGeneratedTabletInsertionEventsParsingStarted();
+
     try {
       if (!waitForTsFileClose()) {
         LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH, tsFile);
+        markGeneratedTabletInsertionEventsParsingCompleted();
         return Collections.emptyList();
       }
       waitForResourceEnough4Parsing(timeoutMs);
-      return initEventParser().toTabletInsertionEvents();
+      return wrapGeneratedTabletInsertionEvents(initEventParser().toTabletInsertionEvents());
     } catch (final Exception e) {
+      markGeneratedTabletInsertionEventsParsingCompleted();
       close();
 
       // close() should be called before re-interrupting the thread
@@ -1014,6 +1170,49 @@ public Iterable<TabletInsertionEvent> toTabletInsertionEvents(final long timeout
     }
   }
 
+  private Iterable<TabletInsertionEvent> wrapGeneratedTabletInsertionEvents(
+      final Iterable<TabletInsertionEvent> iterable) {
+    return () -> {
+      final Iterator<TabletInsertionEvent> iterator;
+      try {
+        iterator = iterable.iterator();
+      } catch (final RuntimeException e) {
+        markGeneratedTabletInsertionEventsParsingCompleted();
+        throw e;
+      }
+      return new Iterator<TabletInsertionEvent>() {
+        @Override
+        public boolean hasNext() {
+          try {
+            final boolean hasNext = iterator.hasNext();
+            if (!hasNext) {
+              markGeneratedTabletInsertionEventsParsingCompleted();
+            }
+            return hasNext;
+          } catch (final RuntimeException e) {
+            markGeneratedTabletInsertionEventsParsingCompleted();
+            throw e;
+          }
+        }
+
+        @Override
+        public TabletInsertionEvent next() {
+          try {
+            return iterator.next();
+          } catch (final RuntimeException e) {
+            markGeneratedTabletInsertionEventsParsingCompleted();
+            throw e;
+          }
+        }
+
+        @Override
+        public void remove() {
+          iterator.remove();
+        }
+      };
+    };
+  }
+
   private void reserveResource4Parsing(
       final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
       throws InterruptedException {