This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/pr19094-tsfile-extraction in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f3471f3a13b2716590c463e6a623e1069438c051 Author: Caideyipi <[email protected]> AuthorDate: Tue Nov 3 09:55:44 2026 +0800 Fix historical TsFile filtering and reference acquisition --- .../common/tsfile/PipeTsFileInsertionEvent.java | 1 + ...istoricalDataRegionTsFileAndDeletionSource.java | 23 ++- .../pipe/event/PipeTsFileInsertionEventTest.java | 32 +++ ...ricalDataRegionTsFileAndDeletionSourceTest.java | 222 +++++++++++++++++++++ 4 files changed, 269 insertions(+), 9 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 5d06cf577b4..92674ee8afe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -967,6 +967,7 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent public boolean shouldParseTime() { if (!isTimeParsed && Objects.nonNull(resource) + && (!isGeneratedByHistoricalExtractor || resource.isClosed()) && startTime <= resource.getFileStartTime() && resource.getFileEndTime() <= endTime) { isTimeParsed = true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java index 80730d8f156..fed95151942 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java @@ -68,6 +68,7 @@ import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment 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; @@ -868,8 +869,8 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource 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) @@ -1057,7 +1058,10 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource } 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(); } @@ -1318,16 +1322,17 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource event.increaseReferenceCount( PipeHistoricalDataRegionTsFileAndDeletionSource.class.getName()); if (!isReferenceCountIncreased) { - LOGGER.warn( - DataNodePipeMessages.PIPE_FAILED_TO_INCREASE_REFERENCE_COUNT_FOR_1, - pipeName, - dataRegionId, - event); + // Keep the queue entry, pin, and replicate index until an event owns the resource. A + // failed reference acquisition must not advance historical progress past this file. + throw new PipeException( + String.format( + DataNodePipeMessages.INCREASE_REFERENCE_COUNT_ERROR_HOLDER_FMT, + event.coreReportMessage())); } filteredTsFileResources2TableNames.remove(resource); shouldUnpinResource = true; shouldClearReplicateIndex = true; - return isReferenceCountIncreased ? event : null; + return event; } finally { if (shouldClearReplicateIndex) { clearReplicateIndexForResource(resource); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java index e905e201ec2..b1a88bf7506 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java @@ -78,6 +78,38 @@ import static org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern.bu import static org.apache.iotdb.db.auth.AuthorityChecker.NO_PERMISSION_PROMOTION; public class PipeTsFileInsertionEventTest { + @Test + public void testRealtimeTsFileTimeParsingRemainsUnchanged() { + final TsFileResource resource = new TsFileResource(new File("1-1-0-0.tsfile")); + final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"t1", "d1"}); + resource.updateStartTime(device, 0); + resource.updateEndTime(device, Long.MIN_VALUE); + resource.setStatusForTest(TsFileResourceStatus.UNCLOSED); + final PipeTsFileInsertionEvent event = + new PipeTsFileInsertionEvent( + true, + "db1", + resource, + null, + false, + false, + false, + Collections.singleton("t1"), + "realtime_time_filter", + 1L, + null, + null, + new TablePattern(true, null, null), + null, + null, + null, + true, + -10, + 10); + Assert.assertFalse(event.shouldParseTime()); + event.close(); + } + @Test public void testHistoricalModsSurviveOriginalFileRemoval() throws Exception { assertHistoricalModsAfterOriginalFileRemoval(true, true); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java index 991dab795f9..2ba204d4def 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java @@ -32,17 +32,24 @@ import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; 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.FileTimeIndex; 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.tsfile.file.metadata.IDeviceID; import org.junit.Assert; import org.junit.Test; @@ -54,14 +61,203 @@ import java.nio.file.Files; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; public class PipeHistoricalDataRegionTsFileAndDeletionSourceTest { + @Test + public void testClosingTsFileKeepsTimeFiltering() throws Exception { + // A closing sequence file can temporarily look wholly inside the extraction window even + // though the eventual file also contains rows outside it. + assertSuppliedTsFileTimeFiltering(false, 0, Long.MIN_VALUE, false, true); + } + + @Test + public void testClosedTsFileTimeFiltering() throws Exception { + assertSuppliedTsFileTimeFiltering(true, -10, 10, false, false); + assertSuppliedTsFileTimeFiltering(true, -20, 20, false, true); + } + + @Test + public void testClosingTsFileHonorsLooseTimeRange() throws Exception { + assertSuppliedTsFileTimeFiltering(false, 0, Long.MIN_VALUE, true, false); + } + + @Test + public void testHistoricalTsFileReferenceFailureRetriesWithoutAdvancingProgress() + throws Exception { + assertHistoricalTsFileReferenceFailure(true); + } + + @Test + public void testHistoricalTsFileReferenceFailureReleasesPinOnClose() throws Exception { + assertHistoricalTsFileReferenceFailure(false); + } + + @SuppressWarnings("unchecked") + private void assertHistoricalTsFileReferenceFailure(final boolean retry) throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalReferenceFailure").toFile(); + final String pipeName = "historical_reference_retry"; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, 1L); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final AtomicBoolean failPathResolution = new AtomicBoolean(); + PipeTsFileInsertionEvent event = null; + try { + final File partitionDir = new File(tempDir, "sequence/db/2147483645/0"); + Assert.assertTrue(partitionDir.mkdirs()); + final TsFileResource resource = createTsFileResource(partitionDir, "1-1-0-0.tsfile"); + resource.setFile(new CanonicalPathFailingFile(resource.getTsFilePath(), failPathResolution)); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + resource.updateProgressIndex(new SimpleProgressIndex(0, 1)); + + setPrivateField(source, "pipeName", pipeName); + setPrivateField(source, "pipeNameWithCreationTime", resourcePipeName); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483645); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "isModelDetected", true); + setPrivateField(source, "isTableModel", true); + setPrivateField(source, "tablePattern", new TablePattern(true, null, null)); + setPrivateField(source, "isDbNameCoveredByPattern", true); + final Map<TsFileResource, Set<String>> selectedResources = + (Map<TsFileResource, Set<String>>) + getPrivateField(source, "filteredTsFileResources2TableNames"); + selectedResources.put(resource, Collections.emptySet()); + final Set<PersistentResource> progressReportResources = + (Set<PersistentResource>) getPrivateField(source, "historicalProgressReportResources"); + progressReportResources.add(resource); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque<PersistentResource>(Collections.singletonList(resource))); + + manager.pinTsFileResource(resource, false, resourcePipeName); + final File pinnedFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile(), resourcePipeName); + failPathResolution.set(true); + Assert.assertThrows(PipeException.class, source::supply); + Assert.assertEquals(1, source.getPendingQueueSize()); + Assert.assertTrue(selectedResources.containsKey(resource)); + Assert.assertTrue(progressReportResources.contains(resource)); + Assert.assertNull(getPrivateField(source, "pendingHistoricalProgressIndexToReport")); + Assert.assertEquals(1, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + + failPathResolution.set(false); + if (retry) { + event = (PipeTsFileInsertionEvent) source.supply(); + Assert.assertNotNull(event); + final Field resourceField = PipeTsFileInsertionEvent.class.getDeclaredField("resource"); + resourceField.setAccessible(true); + Assert.assertSame(resource, resourceField.get(event)); + Assert.assertEquals(0, source.getPendingQueueSize()); + Assert.assertFalse(selectedResources.containsKey(resource)); + // The event now owns the only remaining reference; the source has released its pin. + Assert.assertEquals(1, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + final Event progressEvent = source.supply(); + Assert.assertTrue(progressEvent instanceof ProgressReportEvent); + Assert.assertEquals( + new TimePartitionProgressIndex( + resource.getTimePartition(), resource.getMaxProgressIndex()), + ((ProgressReportEvent) progressEvent).getProgressIndex()); + } else { + source.close(); + Assert.assertEquals(0, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + } + } finally { + failPathResolution.set(false); + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + source.close(); + manager.cleanPipeTsFileDir(resourcePipeName); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @SuppressWarnings("unchecked") + private void assertSuppliedTsFileTimeFiltering( + final boolean closed, + final long fileStartTime, + final long fileEndTime, + final boolean looseTimeRange, + final boolean shouldParseTime) + throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalClosingTimeFilter").toFile(); + final String pipeName = "closing_time_filter"; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, 1L); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + TsFileResource resource = null; + PipeTsFileInsertionEvent event = null; + boolean pinned = false; + try { + final File partitionDir = new File(tempDir, "sequence/db/2147483646/0"); + Assert.assertTrue(partitionDir.mkdirs()); + resource = createTsFileResource(partitionDir, "1-1-0-0.tsfile"); + final IDeviceID device = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "device1"}); + resource.updateStartTime(device, fileStartTime); + resource.updateEndTime(device, fileEndTime); + resource.setStatusForTest( + closed ? TsFileResourceStatus.NORMAL : TsFileResourceStatus.UNCLOSED); + manager.pinTsFileResource(resource, false, resourcePipeName); + pinned = true; + + setPrivateField(source, "pipeName", pipeName); + setPrivateField(source, "pipeNameWithCreationTime", resourcePipeName); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483646); + setPrivateField(source, "isModelDetected", true); + setPrivateField(source, "isTableModel", true); + setPrivateField(source, "tablePattern", new TablePattern(true, null, null)); + setPrivateField(source, "isDbNameCoveredByPattern", true); + setPrivateField(source, "historicalDataExtractionStartTime", -10L); + setPrivateField(source, "historicalDataExtractionEndTime", 10L); + setPrivateField(source, "sloppyTimeRange", looseTimeRange); + ((Map<TsFileResource, Set<String>>) + getPrivateField(source, "filteredTsFileResources2TableNames")) + .put(resource, Collections.emptySet()); + + event = (PipeTsFileInsertionEvent) source.supplyTsFileEvent(resource); + pinned = false; + Assert.assertNotNull(event); + // The collector must not send this entire file without filtering after its close completes. + Assert.assertEquals(shouldParseTime, event.shouldParseTime()); + Assert.assertEquals(!shouldParseTime, PipeEventCollector.canSkipParsing4TsFileEvent(event)); + if (!closed && !looseTimeRange) { + resource.updateStartTime(device, -20); + resource.updateEndTime(device, 20); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + Assert.assertTrue(event.shouldParseTime()); + Assert.assertFalse(PipeEventCollector.canSkipParsing4TsFileEvent(event)); + } + } finally { + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + if (pinned) { + manager.unpinTsFileResource(resource, resourcePipeName, false); + } + source.close(); + manager.cleanPipeTsFileDir(resourcePipeName); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testDeletionOnlyCustomizeInitializesSourceContext() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = @@ -865,6 +1061,32 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSourceTest { return field.get(source); } + private static class CanonicalPathFailingFile extends File { + + private final AtomicBoolean failPathResolution; + + private CanonicalPathFailingFile(final String path, final AtomicBoolean failPathResolution) { + super(path); + this.failPathResolution = failPathResolution; + } + + @Override + public File getParentFile() { + final File parent = super.getParentFile(); + return parent == null + ? null + : new CanonicalPathFailingFile(parent.getPath(), failPathResolution); + } + + @Override + public String getCanonicalPath() throws IOException { + if (failPathResolution.get()) { + throw new IOException("mock path resolution failure"); + } + return super.getCanonicalPath(); + } + } + private static class TestablePipeHistoricalDataRegionTsFileAndDeletionSource extends PipeHistoricalDataRegionTsFileAndDeletionSource {
