[Pipe] Fix stalled region ordering after TsFile-to-Tablet conversion (#19243)

* [Pipe] Fix stalled region ordering after TsFile-to-Tablet conversion

* [Pipe] Track processed tablet descendants for TsFile completion

Author: Caideyipi

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

@@ -286,11 +286,11 @@ private void collectEvent(final Event event) {
     }
 
     if (event instanceof PipeRawTabletInsertionEvent
-        && ((PipeRawTabletInsertionEvent) event).getSourceEvent()
-            instanceof PipeTsFileInsertionEvent) {
+        && ((PipeRawTabletInsertionEvent) event).getSourceTsFileInsertionEvent() != null) {
       // 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())
+      ((PipeRawTabletInsertionEvent) event)
+          .getSourceTsFileInsertionEvent()
           .registerGeneratedTabletInsertionEvent((PipeRawTabletInsertionEvent) event);
     }
 

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

@@ -66,6 +66,7 @@ public class PipeRawTabletInsertionEvent extends PipeInsertionEvent
   private final boolean isAligned;
 
   private final EnrichedEvent sourceEvent;
+  private final PipeTsFileInsertionEvent sourceTsFileInsertionEvent;
   private final AtomicBoolean generatedEventRegisteredWithSource = new AtomicBoolean(false);
   private final AtomicBoolean generatedEventOutcomeReported = new AtomicBoolean(false);
   private boolean needToReport;
@@ -115,6 +116,12 @@ private PipeRawTabletInsertionEvent(
     this.tablet = Objects.requireNonNull(tablet);
     this.isAligned = isAligned;
     this.sourceEvent = sourceEvent;
+    this.sourceTsFileInsertionEvent =
+        sourceEvent instanceof PipeTsFileInsertionEvent
+            ? (PipeTsFileInsertionEvent) sourceEvent
+            : sourceEvent instanceof PipeRawTabletInsertionEvent
+                ? ((PipeRawTabletInsertionEvent) sourceEvent).getSourceTsFileInsertionEvent()
+                : null;
     this.needToReport = needToReport;
     inheritSourceEventReportSkippingIfNecessary();
 
@@ -135,18 +142,17 @@ private PipeRawTabletInsertionEvent(
                 parentMemoryBlock);
     this.allocatedMemoryBlock.setAssigner(this);
 
-    if (needToReport || sourceEvent instanceof PipeTsFileInsertionEvent) {
+    if (needToReport || sourceTsFileInsertionEvent != null) {
       addOnCommittedHook(
           () -> {
             // 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
+            if (sourceTsFileInsertionEvent != null
                 && generatedEventOutcomeReported.compareAndSet(false, true)) {
-              ((PipeTsFileInsertionEvent) sourceEvent)
-                  .markGeneratedTabletInsertionEventAsTransferred();
+              sourceTsFileInsertionEvent.markGeneratedTabletInsertionEventAsTransferred();
             }
           });
     }
@@ -366,8 +372,8 @@ 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();
+        && sourceTsFileInsertionEvent != null) {
+      sourceTsFileInsertionEvent.markGeneratedTabletInsertionEventAsDiscarded();
     }
     return cleared;
   }
@@ -494,6 +500,11 @@ public EnrichedEvent getSourceEvent() {
     return sourceEvent;
   }
 
+  /** Returns the original TsFile, including when a processor has transformed its tablets. */
+  public PipeTsFileInsertionEvent getSourceTsFileInsertionEvent() {
+    return sourceTsFileInsertionEvent;
+  }
+
   @Override
   public boolean isShouldReportOnCommit() {
     return shouldReportOnCommit && needToReport;

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

@@ -592,6 +592,18 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     return isSuccessful;
   }
 
+  @Override
+  public synchronized boolean decreaseReferenceCount(
+      final String holderMessage, final boolean shouldReport) {
+    final boolean decreased = super.decreaseReferenceCount(holderMessage, shouldReport);
+    if (decreased && isReleased() && Objects.isNull(getCommitterKey())) {
+      // A split TsFile may only commit its generated tablets. Releasing that original event ends
+      // its processing, but the transfer hook must still wait for parsing and all tablet commits.
+      markAsTransferredIfGeneratedTabletEventsCompleted();
+    }
+    return decreased;
+  }
+
   @Override
   public boolean clearReferenceCount(final String holderMessage) {
     final boolean cleared = super.clearReferenceCount(holderMessage);
@@ -698,14 +710,12 @@ 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())
+    // The collector only assigns commit ids to the generated tablets when it splits a TsFile.
+    // The original event then finishes processing without entering the commit queue. If the
+    // original event does enter that queue, resource release alone must not bypass its commit.
+    if ((isTsFileEventCommitted.get()
+            || (isReleased() && Objects.isNull(getCommitterKey()))
+            || !generatedTabletInsertionEventsParsingStarted.get())
         && generationBoundaryReached
         && transferredGeneratedTabletInsertionEventCount.get()
                 + discardedGeneratedTabletInsertionEventCount.get()