This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 0d81bbce431 Avoid parsing fully matched historical table TsFiles
(#18123) (#18134)
0d81bbce431 is described below
commit 0d81bbce431dbd87f9b1ad6878b754b7dddfde83
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 9 14:58:00 2026 +0800
Avoid parsing fully matched historical table TsFiles (#18123) (#18134)
(cherry picked from commit bbd424f00138819aa5cf70f3757070591681b11a)
---
.../PipeHistoricalDataRegionTsFileSource.java | 20 +++++++++-
.../PipeHistoricalDataRegionTsFileSourceTest.java | 46 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 1 deletion(-)
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 09b121bfae8..00fd08dc2b7 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 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
&& 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 class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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 3b1396de5c1..a00fc2c19db 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.HybridProgressIndex;
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 class PipeHistoricalDataRegionTsFileSourceTest {
}
}
+ @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 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
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);
