Avoid parsing fully matched historical table TsFiles (#18123)

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

@@ -526,6 +526,24 @@ private boolean isTsFileResourceCoveredByTimeRange(final TsFileResource resource
         && historicalDataExtractionEndTime >= resource.getFileEndTime();
   }
 
+  private boolean isTsFileResourceCoveredByPattern(final TsFileResource resource) {
+    final Set<IDeviceID> deviceSet;
+    try {
+      final Map<IDeviceID, Boolean> deviceIsAlignedMap =
+          PipeDataNodeResourceManager.tsfile()
+              .getDeviceIsAlignedMapFromCache(resource.getTsFile(), false);
+      deviceSet =
+          Objects.nonNull(deviceIsAlignedMap) ? deviceIsAlignedMap.keySet() : resource.getDevices();
+    } catch (final IOException e) {
+      return false;
+    }
+
+    return !deviceSet.isEmpty()
+        && deviceSet.stream()
+            .allMatch(
+                deviceID -> pipePattern.coversDevice(((PlainDeviceID) deviceID).toStringID()));
+  }
+
   @Override
   public synchronized Event supply() {
     if (!hasBeenStarted && StorageEngine.getInstance().isReadyForNonReadWriteFunctions()) {
@@ -598,7 +616,7 @@ public synchronized Event supply() {
             pipePattern,
             historicalDataExtractionStartTime,
             historicalDataExtractionEndTime);
-    if (sloppyPattern || isDbNameCoveredByPattern) {
+    if (sloppyPattern || isDbNameCoveredByPattern || isTsFileResourceCoveredByPattern(resource)) {
       event.skipParsingPattern();
     }
 

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

@@ -24,11 +24,13 @@
 import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 
 import com.google.common.collect.ImmutableMap;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -68,6 +70,38 @@ public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws
     }
   }
 
+  @Test
+  public void testTsFileResourceCoveredByPattern() throws Exception {
+    final File tempDir = Files.createTempDirectory("pipeHistoricalPatternCoverage").toFile();
+
+    try {
+      final PipeHistoricalDataRegionTsFileSource source =
+          new PipeHistoricalDataRegionTsFileSource();
+      final Method method =
+          PipeHistoricalDataRegionTsFileSource.class.getDeclaredMethod(
+              "isTsFileResourceCoveredByPattern", TsFileResource.class);
+      method.setAccessible(true);
+
+      final TsFileResource resource =
+          createClosedTsFileResourceWithDevices(
+              tempDir, "covered-pattern.tsfile", "root.sg.d1", "root.sg.d2");
+
+      setPrivateField(source, "pipePattern", new PrefixPipePattern("root.sg"));
+      Assert.assertTrue((Boolean) method.invoke(source, resource));
+
+      setPrivateField(source, "pipePattern", new PrefixPipePattern("root.sg.d1"));
+      Assert.assertFalse((Boolean) method.invoke(source, resource));
+      Assert.assertFalse(
+          (Boolean)
+              method.invoke(
+                  source,
+                  createClosedTsFileResource(
+                      tempDir, "empty-device.tsfile", new SimpleProgressIndex(0, 1))));
+    } finally {
+      FileUtils.deleteFileOrDirectory(tempDir);
+    }
+  }
+
   private static void assertMayTsFileContainUnprocessedData(
       final File tempDir,
       final String fileName,
@@ -104,6 +138,18 @@ private static TsFileResource createClosedTsFileResource(
     return resource;
   }
 
+  private static TsFileResource createClosedTsFileResourceWithDevices(
+      final File tempDir, final String fileName, final String... devices) throws Exception {
+    final TsFileResource resource =
+        createClosedTsFileResource(tempDir, fileName, new SimpleProgressIndex(0, 1));
+    for (final String device : devices) {
+      final PlainDeviceID deviceID = new PlainDeviceID(device);
+      resource.updateStartTime(deviceID, 0);
+      resource.updateEndTime(deviceID, 1);
+    }
+    return resource;
+  }
+
   private static ProgressIndex hybridProgressIndex(
       final ProgressIndex firstProgressIndex, final ProgressIndex... progressIndexes) {
     ProgressIndex result = new HybridProgressIndex(firstProgressIndex);