diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index 09b121bfae89..00fd08dc2b7b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/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 deviceSet; + try { + final Map 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(); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java index 3b1396de5c18..a00fc2c19dbf 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java +++ b/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);