From d9c7b8696ed64cd4693c78587e432edce4528d1a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 3 Nov 2026 18:30:16 +0800 Subject: [PATCH] Fix historical TsFile filtering and reference acquisition (#19096) * Fix historical TsFile filtering and reference acquisition * Clarify realtime TsFile filtering and reference ownership (cherry picked from commit 814b77028ad42035ae4a8645481d3db84a40f953) --- .../iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../tsfile/PipeTsFileInsertionEvent.java | 8 +- .../PipeHistoricalDataRegionTsFileSource.java | 40 ++-- .../event/PipeTsFileInsertionEventTest.java | 15 ++ ...eHistoricalDataRegionTsFileSourceTest.java | 207 ++++++++++++++++++ 6 files changed, 259 insertions(+), 17 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index db51ff09b0020..02bd37fbfe66c 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -23,6 +23,9 @@ public final class DataNodePipeMessages { private DataNodePipeMessages() {} + public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D = + "Increase reference count error. Holder Message: %s"; + // ===================== LEGACY PIPE RECEIVER RESPONSES ===================== public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 053fc9cf69f8b..8a99dd3592191 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -23,6 +23,9 @@ public final class DataNodePipeMessages { private DataNodePipeMessages() {} + public static final String EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D = + "增加引用计数失败。持有者信息：%s"; + // ===================== LEGACY PIPE RECEIVER RESPONSES ===================== public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 = "Pipe 数据传输错误：%s"; 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 0d379db7831b5..1c76d6577972a 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 @@ -556,16 +556,20 @@ public boolean isGeneratedByPipe() { @Override public boolean mayEventTimeOverlappedWithTimeRange() { - // Notice that this is only called at realtime extraction, and the tsFile is always closed - // Thus we can use the end time to judge the overlap + // Only called during realtime extraction, after writer.endFile() has finalized the time bounds. + // The resource may not yet be marked closed, but its end time can already be used for overlap. return Objects.isNull(resource) || startTime <= resource.getFileEndTime() && resource.getFileStartTime() <= endTime; } @Override public boolean shouldParseTime() { + // Historical extraction may observe incomplete time bounds while an asynchronous close is in + // progress. Realtime TsFile events are published after writer.endFile(), when the bounds are + // final even if the resource's closed flag has not yet been updated by the close callbacks. 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/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index 1d984f9e18f45..9518253a5b7ec 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 @@ -35,6 +35,7 @@ import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment; import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; @@ -50,6 +51,7 @@ 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; @@ -583,8 +585,8 @@ private boolean shouldExtractTsFileResource(final TsFileResource resource) { 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) @@ -730,7 +732,10 @@ private boolean isTsFileResourceOverlappedWithTimeRange(final TsFileResource res } 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(); } @@ -763,8 +768,8 @@ public synchronized Event supply() { return supplyTerminateEvent(); } - pendingQueue.poll(); if (!filteredTsFileResources.contains(resource)) { + pendingQueue.poll(); if (shouldUseHistoricalTsFileQueryPriorityOrder()) { if (shouldReportHistoricalProgressAfterResource(resource)) { return supplyHistoricalProgressReportEvent( @@ -776,6 +781,7 @@ public synchronized Event supply() { } final Event event = supplyTsFileEvent(resource); + pendingQueue.poll(); if (Objects.nonNull(event) && shouldReportHistoricalProgressAfterResource(resource)) { pendingHistoricalProgressIndexToReport = getHistoricalProgressIndexAfterResource(resource); } @@ -846,7 +852,6 @@ private Event supplyHistoricalProgressReportEvent(final ProgressIndex progressIn } private Event supplyTsFileEvent(final TsFileResource resource) { - filteredTsFileResources.remove(resource); final PipeTsFileInsertionEvent event = new PipeTsFileInsertionEvent( resource, @@ -873,17 +878,22 @@ private Event supplyTsFileEvent(final TsFileResource resource) { event.skipParsingTime(); } + final boolean isReferenceCountIncreased = + event.increaseReferenceCount(PipeHistoricalDataRegionTsFileSource.class.getName()); + if (!isReferenceCountIncreased) { + // Unlike realtime supply, which already holds a queued event reference, this is the + // event's first reference acquisition. The source's pin keeps the file available for retry. + // Keep the queue entry and pin until an event owns the resource. A failed reference + // acquisition must not advance historical progress past this file. + throw new PipeException( + String.format( + DataNodePipeMessages + .EXCEPTION_INCREASE_REFERENCE_COUNT_ERROR_HOLDER_MESSAGE_ARG_7E1D032D, + event.coreReportMessage())); + } + filteredTsFileResources.remove(resource); try { - final boolean isReferenceCountIncreased = - event.increaseReferenceCount(PipeHistoricalDataRegionTsFileSource.class.getName()); - if (!isReferenceCountIncreased) { - LOGGER.warn( - "Pipe {}@{}: failed to increase reference count for historical event {}, will discard it", - pipeName, - dataRegionId, - event); - } - return isReferenceCountIncreased ? event : null; + return event; } finally { try { PipeDataNodeResourceManager.tsfile().unpinTsFileResource(resource, pipeName); 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 b30cf4f98e3f5..b5a42056c7c0b 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 @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; 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.DeviceTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; @@ -37,6 +38,20 @@ public class PipeTsFileInsertionEventTest { + @Test + public void testRealtimeTsFileTimeParsingRemainsUnchanged() { + final TsFileResource resource = new TsFileResource(new File("1-1-0-0.tsfile")); + final IDeviceID device = new PlainDeviceID("root.db1.d1"); + resource.updateStartTime(device, 0); + resource.updateEndTime(device, Long.MIN_VALUE); + resource.setStatusForTest(TsFileResourceStatus.UNCLOSED); + final PipeTsFileInsertionEvent event = + new PipeTsFileInsertionEvent( + resource, null, false, false, false, "realtime_time_filter", 1L, null, null, -10, 10); + Assert.assertFalse(event.shouldParseTime()); + event.close(); + } + @Test(timeout = 60000) public void testRealtimeEventCanSkipWaitingForClosedStatusAfterTsFileSealed() throws Exception { final File tempDir = Files.createTempDirectory("pipeTsFileSealed").toFile(); 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 7562e54e5811b..32225750a65be 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 @@ -32,12 +32,17 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern; 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.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 com.google.common.collect.ImmutableMap; import org.apache.tsfile.file.metadata.PlainDeviceID; @@ -52,14 +57,190 @@ 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 PipeHistoricalDataRegionTsFileSourceTest { + @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 PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + final AtomicBoolean failPathResolution = new AtomicBoolean(); + PipeTsFileInsertionEvent event = null; + try { + final File partitionDir = new File(tempDir, "sequence/root.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, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483645); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "isDbNameCoveredByPattern", true); + final Set selectedResources = + (Set) getPrivateField(source, "filteredTsFileResources"); + selectedResources.add(resource); + final Set progressReportResources = + (Set) getPrivateField(source, "historicalProgressReportResources"); + progressReportResources.add(resource); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque(Collections.singletonList(resource))); + + manager.pinTsFileResource(resource, false, pipeName); + final File pinnedFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile(), pipeName); + failPathResolution.set(true); + Assert.assertThrows(PipeException.class, source::supply); + Assert.assertEquals(1, source.getPendingQueueSize()); + Assert.assertTrue(selectedResources.contains(resource)); + Assert.assertTrue(progressReportResources.contains(resource)); + Assert.assertNull(getPrivateField(source, "pendingHistoricalProgressIndexToReport")); + Assert.assertEquals(1, manager.getFileReferenceCount(pinnedFile, pipeName)); + + 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.contains(resource)); + // In 1.3, the source pin and event reference use separate pipe directories. + Assert.assertEquals(0, manager.getFileReferenceCount(pinnedFile, pipeName)); + Assert.assertEquals(1, manager.getFileReferenceCount(event.getTsFile(), 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, pipeName)); + } + } finally { + failPathResolution.set(false); + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + source.close(); + manager.cleanPipeTsFileDir(pipeName); + 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 PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + PipeTsFileInsertionEvent event = null; + try { + final File partitionDir = new File(tempDir, "sequence/root.db/2147483646/0"); + Assert.assertTrue(partitionDir.mkdirs()); + final TsFileResource resource = createTsFileResource(partitionDir, "1-1-0-0.tsfile"); + final PlainDeviceID device = new PlainDeviceID("root.db.device1"); + resource.updateStartTime(device, fileStartTime); + resource.updateEndTime(device, fileEndTime); + resource.setStatusForTest( + closed ? TsFileResourceStatus.NORMAL : TsFileResourceStatus.UNCLOSED); + + setPrivateField(source, "pipeName", pipeName); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483646); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "isDbNameCoveredByPattern", true); + setPrivateField(source, "historicalDataExtractionStartTime", -10L); + setPrivateField(source, "historicalDataExtractionEndTime", 10L); + setPrivateField(source, "sloppyTimeRange", looseTimeRange); + ((Set) getPrivateField(source, "filteredTsFileResources")).add(resource); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque(Collections.singletonList(resource))); + manager.pinTsFileResource(resource, false, pipeName); + + event = (PipeTsFileInsertionEvent) source.supply(); + 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()); + } + source.close(); + manager.cleanPipeTsFileDir(pipeName); + manager.cleanPipeTsFileDir(resourcePipeName); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testGlobalTimeRangeRespectsHistoryEnable() throws Exception { final Map attributes = new HashMap<>(); @@ -443,6 +624,32 @@ public void testMissingTsFileResourceDoesNotBlockHistoricalExtraction() throws E } } + 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 void assertMayTsFileContainUnprocessedData( final File tempDir, final String fileName,