Pipe: improve progress coverage checks

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

@@ -765,7 +765,7 @@ public boolean persistAllProgressIndex(final long timeoutInMs) {
     persistThread.setDaemon(true);
 
     LOGGER.info(
-        "Start to persist all pipe progress indexes during shutdown, pipe count {}, timeout {} ms.",
+        "Start to persist all pipe progress indexes during shutdown, pipe count: {}, timeout: {} ms.",
         getPipeCount(),
         normalizedTimeoutInMs);
     persistThread.start();
@@ -816,7 +816,8 @@ private boolean persistAllProgressIndexInternal() {
               .mapToInt(ByteBuffer::remaining)
               .sum();
       LOGGER.info(
-          "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, pipe meta size {} bytes, cost {} ms.",
+          "Collected pipe metas for shutdown progress persist, local pipe count: {}, "
+              + "collected pipe meta count: {}, total serialized size: {} bytes, cost: {} ms.",
           pipeCount,
           pipeMetaCount,
           pipeMetaSizeInBytes,
@@ -825,7 +826,8 @@ private boolean persistAllProgressIndexInternal() {
       if (resp.getPipeMetaList().isEmpty()) {
         if (pipeCount != 0) {
           LOGGER.info(
-              "Collected empty pipe metas during shutdown while pipe count is {}.", pipeCount);
+              "Collected empty pipe metas during shutdown while local pipe count is {}.",
+              pipeCount);
           return false;
         }
         return true;
@@ -834,7 +836,8 @@ private boolean persistAllProgressIndexInternal() {
       try (final ConfigNodeClient configNodeClient =
           ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
         LOGGER.info(
-            "Start to push heartbeat shutdown pipe meta to ConfigNode, data node id {}, pipe count {}, pipe meta count {}, pipe meta size {} bytes.",
+            "Start to push shutdown pipe meta heartbeat to ConfigNode, dataNodeId: {}, "
+                + "local pipe count: {}, collected pipe meta count: {}, total serialized size: {} bytes.",
             IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
             pipeCount,
             pipeMetaCount,
@@ -845,20 +848,21 @@ private boolean persistAllProgressIndexInternal() {
                 IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp);
         final long pushCostTime = System.currentTimeMillis() - pushStartTime;
         if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) {
-          LOGGER.warn("Failed to persist progress index to ConfigNode, status: {}", result);
+          LOGGER.warn("Failed to persist progress index to configNode, status: {}", result);
           LOGGER.warn(
-              "Failed to push heartbeat shutdown pipe meta to ConfigNode, status {}, cost {} ms.",
+              "Failed to push shutdown pipe meta heartbeat to ConfigNode, status: {}, cost {} ms.",
               result,
               pushCostTime);
           return false;
         } else {
           LOGGER.info(
-              "Successfully finished pushing heartbeat shutdown pipe meta to ConfigNode, pipe count {}, pipe meta count {}, pipe meta size {} bytes, cost {} ms.",
+              "Successfully pushed shutdown pipe meta heartbeat to ConfigNode, local pipe count: {}, "
+                  + "collected pipe meta count: {}, total serialized size: {} bytes, cost {} ms.",
               pipeCount,
               pipeMetaCount,
               pipeMetaSizeInBytes,
               pushCostTime);
-          LOGGER.info("Successfully persisted all pipe's info to ConfigNode.");
+          LOGGER.info("Successfully persisted all pipe's info to configNode.");
           return true;
         }
       }

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

@@ -406,15 +406,17 @@ public synchronized void start() {
             originalUnSequenceTsFileCount,
             startIndex);
 
+        final HistoricalTsFileExtractionStatistics statistics =
+            new HistoricalTsFileExtractionStatistics();
         final Collection<TsFileResource> sequenceTsFileResources =
             tsFileManager.getTsFileList(true).stream()
                 .peek(originalResourceList::add)
-                .filter(this::shouldExtractTsFileResource)
+                .filter(resource -> shouldExtractTsFileResource(resource, statistics))
                 .collect(Collectors.toList());
         final Collection<TsFileResource> unSequenceTsFileResources =
             tsFileManager.getTsFileList(false).stream()
                 .peek(originalResourceList::add)
-                .filter(this::shouldExtractTsFileResource)
+                .filter(resource -> shouldExtractTsFileResource(resource, statistics))
                 .collect(Collectors.toList());
         filteredTsFileResources.addAll(sequenceTsFileResources);
         filteredTsFileResources.addAll(unSequenceTsFileResources);
@@ -428,6 +430,7 @@ public synchronized void start() {
                     .pinTsFileResource(resource, shouldTransferModFile, pipeName);
                 return false;
               } catch (final IOException e) {
+                ++statistics.pinFailedCount;
                 LOGGER.warn("Pipe: failed to pin TsFileResource {}", resource.getTsFilePath(), e);
                 return true;
               }
@@ -456,6 +459,22 @@ public synchronized void start() {
             filteredTsFileResources.size(),
             originalSequenceTsFileCount + originalUnSequenceTsFileCount,
             System.currentTimeMillis() - startHistoricalExtractionTime);
+        LOGGER.info(
+            "Pipe {}@{}: historical TsFile selection summary, selected by progress uncovered {}, "
+                + "selected by unclosed/closing {}, filtered by time/path {} (time {}, path {}), "
+                + "skipped covered {}, skipped deleted {}, skipped generated by pipe {}, "
+                + "pin failed {}",
+            pipeName,
+            dataRegionId,
+            statistics.selectedByProgressUncoveredCount,
+            statistics.selectedByUnclosedOrClosingCount,
+            statistics.filteredByTimeOrPathCount,
+            statistics.filteredByTimeCount,
+            statistics.filteredByPathCount,
+            statistics.skippedCoveredCount,
+            statistics.skippedDeletedCount,
+            statistics.skippedGeneratedByPipeCount,
+            statistics.pinFailedCount);
       } finally {
         tsFileManager.readUnlock();
       }
@@ -567,43 +586,54 @@ private int compareTsFileResourcesByQueryPriority(
     return resource1.getTsFilePath().compareTo(resource2.getTsFilePath());
   }
 
-  private boolean shouldExtractTsFileResource(final TsFileResource resource) {
+  private boolean shouldExtractTsFileResource(
+      final TsFileResource resource, final HistoricalTsFileExtractionStatistics statistics) {
     if (!isHistoricalSourceEnabled) {
       return false;
     }
 
     // Some resource is marked as deleted but not removed from the list.
     if (resource.isDeleted()) {
+      ++statistics.skippedDeletedCount;
       return false;
     }
 
     // Some resource is generated by pipe. We ignore them if the pipe should not transfer pipe
     // requests.
     if (resource.isGeneratedByPipe() && !isForwardingPipeRequests) {
+      ++statistics.skippedGeneratedByPipeCount;
       return false;
     }
 
-    // Some resource may not be closed due to the control of PIPE_MIN_FLUSH_INTERVAL_IN_MS. We
-    // simply ignore them.
+    // If the tsFile is not already marked closing, it is not captured by the pipe realtime module.
+    // Thus, we can wait for the realtime sync module to handle this, to avoid blocking the pipe
+    // sync process.
     if (!resource.isClosed()
         && Optional.ofNullable(resource.getProcessor())
             .map(TsFileProcessor::alreadyMarkedClosing)
             .orElse(true)) {
+      ++statistics.selectedByUnclosedOrClosingCount;
       return true;
     }
 
     if (!mayTsFileContainUnprocessedData(resource)) {
+      ++statistics.skippedCoveredCount;
       return false;
     }
 
     if (!isTsFileResourceOverlappedWithTimeRange(resource)) {
+      ++statistics.filteredByTimeOrPathCount;
+      ++statistics.filteredByTimeCount;
       return false;
     }
 
     if (!mayTsFileResourceOverlappedWithPattern(resource)) {
+      ++statistics.filteredByTimeOrPathCount;
+      ++statistics.filteredByPathCount;
       return false;
     }
 
+    ++statistics.selectedByProgressUncoveredCount;
     return true;
   }
 
@@ -678,6 +708,19 @@ private TimeWindowStateProgressIndex getTimeWindowStateProgressIndex(
         : progressIndex.getProgressIndexByType(TimeWindowStateProgressIndex.class).orElse(null);
   }
 
+  private static class HistoricalTsFileExtractionStatistics {
+
+    private int selectedByProgressUncoveredCount;
+    private int selectedByUnclosedOrClosingCount;
+    private int filteredByTimeOrPathCount;
+    private int filteredByTimeCount;
+    private int filteredByPathCount;
+    private int skippedCoveredCount;
+    private int skippedDeletedCount;
+    private int skippedGeneratedByPipeCount;
+    private int pinFailedCount;
+  }
+
   private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) {
     // Trimming to avoid unnecessary file device getter
     if (isDbNameCoveredByPattern) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java

@@ -151,7 +151,7 @@ public void run() {
     final long shutdownProgressPersistTimeoutInMs =
         PipeDataNodeAgent.task().getShutdownProgressPersistTimeoutInMs();
     logger.info(
-        "Persisting pipe progress indexes before shutdown, timeout {} ms.",
+        "Persisting pipe progress indexes before shutdown with timeout {} ms.",
         shutdownProgressPersistTimeoutInMs);
     if (!PipeDataNodeAgent.task().persistAllProgressIndex(shutdownProgressPersistTimeoutInMs)) {
       logger.warn("Pipe progress indexes were not confirmed during shutdown.");

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java

@@ -39,7 +39,6 @@
 import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.pipe.api.event.Event;
 
-import com.google.common.collect.ImmutableMap;
 import org.apache.tsfile.file.metadata.PlainDeviceID;
 import org.junit.Assert;
 import org.junit.Test;
@@ -104,21 +103,41 @@ public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws
           tempDir,
           "superset.tsfile",
           hybridProgressIndex(
-              new IoTProgressIndex(ImmutableMap.of(1, 100L, 2, 200L)),
+              iotProgressIndex(1, 100L, 2, 200L),
               new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))),
           hybridProgressIndex(
-              new IoTProgressIndex(1, 100L),
+              iotProgressIndex(1, 100L),
               new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 9))),
           false);
 
       assertMayTsFileContainUnprocessedData(
           tempDir,
           "missing-dimension.tsfile",
-          hybridProgressIndex(new IoTProgressIndex(1, 100L)),
+          hybridProgressIndex(iotProgressIndex(1, 100L)),
           hybridProgressIndex(
-              new IoTProgressIndex(1, 90L),
+              iotProgressIndex(1, 90L),
               new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))),
           true);
+
+      assertMayTsFileContainUnprocessedData(
+          tempDir,
+          "larger-iot.tsfile",
+          hybridProgressIndex(
+              iotProgressIndex(1, 100L, 2, 200L),
+              new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))),
+          hybridProgressIndex(
+              iotProgressIndex(1, 101L),
+              new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))),
+          true);
+
+      final ProgressIndex recoverProgressIndex =
+          new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10));
+      assertMayTsFileContainUnprocessedData(
+          tempDir,
+          "old-sequence-recover.tsfile",
+          hybridProgressIndex(recoverProgressIndex, iotProgressIndex(1, 100L)),
+          recoverProgressIndex,
+          false);
     } finally {
       FileUtils.deleteFileOrDirectory(tempDir);
     }
@@ -543,6 +562,16 @@ private static ProgressIndex hybridProgressIndex(
     return result;
   }
 
+  private static IoTProgressIndex iotProgressIndex(
+      final int firstPeerId, final long firstSearchIndex, final long... peerIdAndSearchIndexPairs) {
+    final Map<Integer, Long> peerId2SearchIndex = new HashMap<>();
+    peerId2SearchIndex.put(firstPeerId, firstSearchIndex);
+    for (int i = 0; i < peerIdAndSearchIndexPairs.length; i += 2) {
+      peerId2SearchIndex.put((int) peerIdAndSearchIndexPairs[i], peerIdAndSearchIndexPairs[i + 1]);
+    }
+    return new IoTProgressIndex(peerId2SearchIndex);
+  }
+
   private static void sortExtractedResources(
       final PipeHistoricalDataRegionTsFileSource source, final List<TsFileResource> resources)
       throws ReflectiveOperationException {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeperTest.java

@@ -35,6 +35,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.Map;
 
 public class PipeTsFileEpochProgressIndexKeeperTest {
 
@@ -97,24 +99,23 @@ public void testProgressIndexCheckDoesNotLeakAcrossTaskScopes() throws IOExcepti
   public void testProgressIndexCheckUsesEqualOrAfterCoverage() throws IOException {
     final ProgressIndex registeredProgressIndex =
         hybridProgressIndex(
-            new IoTProgressIndex(1, 90L),
-            new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10)));
+            iotProgressIndex(1, 90L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10)));
     keeper.registerProgressIndex(
         DATA_REGION_ID,
         TASK_SCOPE_A,
         createTsFileResource("registered-hybrid.tsfile", registeredProgressIndex));
 
     Assert.assertFalse(
         keeper.isProgressIndexAfterOrEquals(
-            DATA_REGION_ID, TASK_SCOPE_A, "current.tsfile", new IoTProgressIndex(1, 100L)));
+            DATA_REGION_ID, TASK_SCOPE_A, "current.tsfile", iotProgressIndex(1, 100L)));
 
     Assert.assertTrue(
         keeper.isProgressIndexAfterOrEquals(
             DATA_REGION_ID,
             TASK_SCOPE_A,
             "current.tsfile",
             hybridProgressIndex(
-                new IoTProgressIndex(1, 100L),
+                iotProgressIndex(1, 100L),
                 new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10)))));
   }
 
@@ -157,4 +158,10 @@ private ProgressIndex hybridProgressIndex(
     }
     return result;
   }
+
+  private IoTProgressIndex iotProgressIndex(final int peerId, final long searchIndex) {
+    final Map<Integer, Long> peerId2SearchIndex = new HashMap<>();
+    peerId2SearchIndex.put(peerId, searchIndex);
+    return new IoTProgressIndex(peerId2SearchIndex);
+  }
 }