[To dev/1.3] Fix historical TsFile filtering and reference acquisition (#19096)

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -23,6 +23,9 @@ public final class DataNodePipeMessages {
 
   private DataNodePipeMessages() {}
 
+  public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D =
+      "Increase reference count error. Holder Message: %s";
+
   // ===================== LEGACY PIPE RECEIVER RESPONSES =====================
 
   public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -23,6 +23,9 @@ public final class DataNodePipeMessages {
 
   private DataNodePipeMessages() {}
 
+  public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D =
+      "增加引用计数失败。持有者信息：%s";
+
   // ===================== LEGACY PIPE RECEIVER RESPONSES =====================
 
   public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = "Pipe 数据传输错误：%s";

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

@@ -556,16 +556,20 @@ public boolean isGeneratedByPipe() {
 
   @Override
   public boolean mayEventTimeOverlappedWithTimeRange() {
-    // Notice that this is only called at realtime extraction, and the tsFile is always closed
-    // Thus we can use the end time to judge the overlap
+    // Only called during realtime extraction, after writer.endFile() has finalized the time bounds.
+    // The resource may not yet be marked closed, but its end time can already be used for overlap.
     return Objects.isNull(resource)
         || startTime <= resource.getFileEndTime() && resource.getFileStartTime() <= endTime;
   }
 
   @Override
   public boolean shouldParseTime() {
+    // Historical extraction may observe incomplete time bounds while an asynchronous close is in
+    // progress. Realtime TsFile events are published after writer.endFile(), when the bounds are
+    // final even if the resource's closed flag has not yet been updated by the close callbacks.
     if (!isTimeParsed
         && Objects.nonNull(resource)
+        && (!isGeneratedByHistoricalExtractor || resource.isClosed())
         && startTime <= resource.getFileStartTime()
         && resource.getFileEndTime() <= endTime) {
       isTimeParsed = true;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java

@@ -35,6 +35,7 @@
 import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
 import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
@@ -50,6 +51,7 @@
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.exception.PipeException;
 import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
@@ -583,8 +585,8 @@ private boolean shouldExtractTsFileResource(final TsFileResource resource) {
       return false;
     }
 
-    // Some resource may not be closed due to the control of PIPE_MIN_FLUSH_INTERVAL_IN_MS. We
-    // simply ignore them.
+    // Closing files may still have incomplete progress and time metadata. Select them
+    // conservatively; the event waits for the file to close before parsing or transfer.
     if (!resource.isClosed()
         && Optional.ofNullable(resource.getProcessor())
             .map(TsFileProcessor::alreadyMarkedClosing)
@@ -730,7 +732,10 @@ private boolean isTsFileResourceOverlappedWithTimeRange(final TsFileResource res
   }
 
   private boolean isTsFileResourceCoveredByTimeRange(final TsFileResource resource) {
-    return historicalDataExtractionStartTime <= resource.getFileStartTime()
+    // In particular, a sequence file's end time is not final until it is closed. Do not disable
+    // row-level filtering based on the bounds observed while an asynchronous close is in flight.
+    return resource.isClosed()
+        && historicalDataExtractionStartTime <= resource.getFileStartTime()
         && historicalDataExtractionEndTime >= resource.getFileEndTime();
   }
 
@@ -763,8 +768,8 @@ public synchronized Event supply() {
         return supplyTerminateEvent();
       }
 
-      pendingQueue.poll();
       if (!filteredTsFileResources.contains(resource)) {
+        pendingQueue.poll();
         if (shouldUseHistoricalTsFileQueryPriorityOrder()) {
           if (shouldReportHistoricalProgressAfterResource(resource)) {
             return supplyHistoricalProgressReportEvent(
@@ -776,6 +781,7 @@ public synchronized Event supply() {
       }
 
       final Event event = supplyTsFileEvent(resource);
+      pendingQueue.poll();
       if (Objects.nonNull(event) && shouldReportHistoricalProgressAfterResource(resource)) {
         pendingHistoricalProgressIndexToReport = getHistoricalProgressIndexAfterResource(resource);
       }
@@ -846,7 +852,6 @@ private Event supplyHistoricalProgressReportEvent(final ProgressIndex progressIn
   }
 
   private Event supplyTsFileEvent(final TsFileResource resource) {
-    filteredTsFileResources.remove(resource);
     final PipeTsFileInsertionEvent event =
         new PipeTsFileInsertionEvent(
             resource,
@@ -873,17 +878,22 @@ private Event supplyTsFileEvent(final TsFileResource resource) {
       event.skipParsingTime();
     }
 
+    final boolean isReferenceCountIncreased =
+        event.increaseReferenceCount(PipeHistoricalDataRegionTsFileSource.class.getName());
+    if (!isReferenceCountIncreased) {
+      // Unlike realtime supply, which already holds a queued event reference, this is the
+      // event's first reference acquisition. The source's pin keeps the file available for retry.
+      // Keep the queue entry and pin until an event owns the resource. A failed reference
+      // acquisition must not advance historical progress past this file.
+      throw new PipeException(
+          String.format(
+              DataNodePipeMessages
+                  .EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D,
+              event.coreReportMessage()));
+    }
+    filteredTsFileResources.remove(resource);
     try {
-      final boolean isReferenceCountIncreased =
-          event.increaseReferenceCount(PipeHistoricalDataRegionTsFileSource.class.getName());
-      if (!isReferenceCountIncreased) {
-        LOGGER.warn(
-            "Pipe {}@{}: failed to increase reference count for historical event {}, will discard it",
-            pipeName,
-            dataRegionId,
-            event);
-      }
-      return isReferenceCountIncreased ? event : null;
+      return event;
     } finally {
       try {
         PipeDataNodeResourceManager.tsfile().unpinTsFileResource(resource, pipeName);

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.DeviceTimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
 
@@ -37,6 +38,20 @@
 
 public class PipeTsFileInsertionEventTest {
 
+  @Test
+  public void testRealtimeTsFileTimeParsingRemainsUnchanged() {
+    final TsFileResource resource = new TsFileResource(new File("1-1-0-0.tsfile"));
+    final IDeviceID device = new PlainDeviceID("root.db1.d1");
+    resource.updateStartTime(device, 0);
+    resource.updateEndTime(device, Long.MIN_VALUE);
+    resource.setStatusForTest(TsFileResourceStatus.UNCLOSED);
+    final PipeTsFileInsertionEvent event =
+        new PipeTsFileInsertionEvent(
+            resource, null, false, false, false, "realtime_time_filter", 1L, null, null, -10, 10);
+    Assert.assertFalse(event.shouldParseTime());
+    event.close();
+  }
+
   @Test(timeout = 60000)
   public void testRealtimeEventCanSkipWaitingForClosedStatusAfterTsFileSealed() throws Exception {
     final File tempDir = Files.createTempDirectory("pipeTsFileSealed").toFile();