From 3a32f1ea9653af3b9a8b7c3383148d9759492af8 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:44:08 +0800 Subject: [PATCH 1/4] Pipe: order historical TsFiles by flush time --- .../tablet/PipeRawTabletInsertionEvent.java | 13 ++ .../tsfile/PipeTsFileInsertionEvent.java | 22 ++- .../dataregion/IoTDBDataRegionSource.java | 12 ++ ...icalDataRegionTsFileAndDeletionSource.java | 164 ++++++++++++++--- .../event/PipeTabletInsertionEventTest.java | 59 ++++++ ...DataRegionTsFileAndDeletionSourceTest.java | 168 ++++++++++++++++++ .../config/constant/PipeSourceConstant.java | 5 + 7 files changed, 420 insertions(+), 23 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java index 5bdef252a2f4a..dc2ab1d381fd8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java @@ -112,6 +112,7 @@ private PipeRawTabletInsertionEvent( this.isAligned = isAligned; this.sourceEvent = sourceEvent; this.needToReport = needToReport; + inheritSourceEventReportSkippingIfNecessary(); // Allocate empty memory block, will be resized later. this.allocatedMemoryBlock = @@ -399,6 +400,18 @@ public void markAsNeedToReport() { }); } this.needToReport = true; + inheritSourceEventReportSkippingIfNecessary(); + } + + private void inheritSourceEventReportSkippingIfNecessary() { + if (needToReport && shouldSkipReportOnCommitBecauseOfSourceEvent()) { + skipReportOnCommit(); + } + } + + private boolean shouldSkipReportOnCommitBecauseOfSourceEvent() { + return sourceEvent instanceof PipeTsFileInsertionEvent + && !((PipeTsFileInsertionEvent) sourceEvent).shouldReportGeneratedEventsOnCommit(); } // This getter is reserved for user-defined plugins 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 8f664761aa032..b4e1df26dbecd 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 @@ -105,6 +105,8 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent protected volatile ProgressIndex overridingProgressIndex; private Set tableNames; private String tsFileDedupScopeID; + // False when generated tablet events should wait for an external progress report. + private volatile boolean shouldReportGeneratedEventsOnCommit = true; // This is set to check the tsFile paths by privilege private Map treeSchemaMap; @@ -462,6 +464,23 @@ public ProgressIndex forceGetProgressIndex() { return resource.getMaxProgressIndex(); } + public PipeTsFileInsertionEvent skipReportOnCommitAndGeneratedEvents() { + return setShouldReportGeneratedEventsOnCommit(false); + } + + public boolean shouldReportGeneratedEventsOnCommit() { + return shouldReportGeneratedEventsOnCommit; + } + + private PipeTsFileInsertionEvent setShouldReportGeneratedEventsOnCommit( + final boolean shouldReportGeneratedEventsOnCommit) { + this.shouldReportGeneratedEventsOnCommit = shouldReportGeneratedEventsOnCommit; + if (!shouldReportGeneratedEventsOnCommit) { + skipReportOnCommit(); + } + return this; + } + public void eliminateProgressIndex() { if (Objects.isNull(overridingProgressIndex) && Objects.nonNull(resource) @@ -517,7 +536,8 @@ public PipeTsFileInsertionEvent shallowCopySelfAndBindPipeTaskMetaForProgressRep startTime, endTime, isTsFileSealed) - .bindTsFileDedupScopeID(tsFileDedupScopeID); + .bindTsFileDedupScopeID(tsFileDedupScopeID) + .setShouldReportGeneratedEventsOnCommit(shouldReportGeneratedEventsOnCommit); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java index c5dc7662bbc72..66944f776d0f8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java @@ -75,6 +75,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STREAMING_DEFAULT_VALUE; @@ -107,6 +108,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STREAMING_KEY; @@ -186,6 +188,16 @@ public void validate(final PipeParameterValidator validator) throws Exception { SOURCE_HISTORY_ENABLE_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString()) .validateAttributeValueRange( SOURCE_REALTIME_ENABLE_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString()) + .validateAttributeValueRange( + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + true, + Boolean.TRUE.toString(), + Boolean.FALSE.toString()) + .validateAttributeValueRange( + SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + true, + Boolean.TRUE.toString(), + Boolean.FALSE.toString()) .validate( args -> (boolean) args[0] || (boolean) args[1], "Should not set both history.enable and realtime.enable to false.", 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 36b84e1e12712..b1faf4efa7f6e 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 @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.ProgressIndexType; import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; @@ -98,6 +99,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_TIME_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STRICT_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STRICT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_DEFAULT_VALUE; @@ -110,6 +113,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STRICT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_KEY; @@ -149,6 +153,8 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource private boolean sloppyTimeRange; // true to disable time range filter after extraction private boolean sloppyPattern; // true to disable pattern filter after extraction + private boolean shouldOrderHistoricalTsFileByFlushTime = + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE; private Pair listeningOptionPair; private boolean shouldExtractInsertion; @@ -170,6 +176,8 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource new HashMap<>(); private final Map pendingResource2ReplicateIndexForIoTV2 = new HashMap<>(); + private ProgressIndex maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + private boolean shouldReportMaxHistoricalProgressIndex = false; private int extractedHistoricalTsFileCount = 0; private int extractedHistoricalDeletionCount = 0; @@ -185,6 +193,13 @@ public void validate(final PipeParameterValidator validator) { throw new PipeParameterNotValidException(e.getMessage()); } + shouldOrderHistoricalTsFileByFlushTime = + parameters.getBooleanOrDefault( + Arrays.asList( + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY), + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE); + if (parameters.hasAnyAttributes(EXTRACTOR_MODE_STRICT_KEY, SOURCE_MODE_STRICT_KEY)) { final boolean isStrictMode = parameters.getBooleanOrDefault( @@ -311,6 +326,12 @@ public void customize( throws IllegalPathException { shouldExtractInsertion = listeningOptionPair.getLeft(); shouldExtractDeletion = listeningOptionPair.getRight(); + shouldOrderHistoricalTsFileByFlushTime = + parameters.getBooleanOrDefault( + Arrays.asList( + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY), + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE); final PipeRuntimeEnvironment environment = configuration.getRuntimeEnvironment(); @@ -496,6 +517,8 @@ public synchronized void start() { hasBeenStarted = true; extractedHistoricalTsFileCount = 0; extractedHistoricalDeletionCount = 0; + maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + shouldReportMaxHistoricalProgressIndex = false; final DataRegion dataRegion = StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId)); @@ -519,15 +542,15 @@ public synchronized void start() { .ifPresent(manager -> extractDeletions(manager, originalResourceList)); } + if (shouldUseHistoricalTsFileFlushTimeOrder()) { + prepareResourcesForHistoricalTsFileFlushTimeOrder(originalResourceList); + } + // Sort tsFileResource and deletionResource long startTime = System.currentTimeMillis(); LOGGER.info( DataNodePipeMessages.PIPE_START_TO_SORT_ALL_EXTRACTED_RESOURCES, pipeName, dataRegionId); - originalResourceList.sort( - (o1, o2) -> - startIndex instanceof TimeWindowStateProgressIndex - ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime()) - : o1.getProgressIndex().topologicalCompareTo(o2.getProgressIndex())); + sortExtractedResources(originalResourceList); pendingQueue = new ArrayDeque<>(originalResourceList); PipeTerminateEvent.initializeHistoricalTransferSummary( pipeName, @@ -546,6 +569,88 @@ public synchronized void start() { } } + private boolean shouldUseHistoricalTsFileFlushTimeOrder() { + // Deletion resources only carry progressIndex. Keep the old progressIndex order when deletions + // are extracted together with TsFiles so insertion/deletion ordering semantics are unchanged. + return shouldOrderHistoricalTsFileByFlushTime + && shouldExtractInsertion + && !shouldExtractDeletion; + } + + private void prepareResourcesForHistoricalTsFileFlushTimeOrder( + final List resourceList) { + // Flush-time order is intentionally not compatible with progressIndex order, so report progress + // only after all selected historical TsFiles are supplied. This prefers possible retransmission + // over losing overwrite semantics. + resourceList.removeIf( + resource -> + resource instanceof TsFileResource + && !filteredTsFileResources2TableNames.containsKey(resource)); + updateMaxHistoricalProgressIndex(resourceList); + shouldReportMaxHistoricalProgressIndex = !resourceList.isEmpty(); + } + + private void updateMaxHistoricalProgressIndex(final List resourceList) { + for (final PersistentResource resource : resourceList) { + final ProgressIndex progressIndex = resource.getProgressIndex(); + if (Objects.nonNull(progressIndex)) { + maxHistoricalProgressIndex = + maxHistoricalProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex); + } + } + } + + private void sortExtractedResources(final List resourceList) { + if (shouldUseHistoricalTsFileFlushTimeOrder()) { + // Send TsFiles in source-side file creation order. For duplicated points, older files are + // loaded first on the receiver and newer files are loaded later to preserve overwrite + // semantics. + resourceList.sort( + (o1, o2) -> + o1 instanceof TsFileResource && o2 instanceof TsFileResource + ? compareTsFileResourcesByFlushTime((TsFileResource) o1, (TsFileResource) o2) + : comparePersistentResourcesByProgressIndex(o1, o2)); + return; + } + + resourceList.sort( + (o1, o2) -> + startIndex instanceof TimeWindowStateProgressIndex + ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime()) + : comparePersistentResourcesByProgressIndex(o1, o2)); + } + + private int comparePersistentResourcesByProgressIndex( + final PersistentResource resource1, final PersistentResource resource2) { + return resource1.getProgressIndex().topologicalCompareTo(resource2.getProgressIndex()); + } + + private int compareTsFileResourcesByFlushTime( + final TsFileResource resource1, final TsFileResource resource2) { + if (resource1.isSeq() != resource2.isSeq()) { + return resource1.isSeq() ? -1 : 1; + } + + int result = Long.compare(resource1.getTsFileID().timestamp, resource2.getTsFileID().timestamp); + if (result != 0) { + return result; + } + + result = Long.compare(resource1.getVersion(), resource2.getVersion()); + if (result != 0) { + return result; + } + + result = + Long.compare( + resource1.getTsFileID().compactionVersion, resource2.getTsFileID().compactionVersion); + if (result != 0) { + return result; + } + + return resource1.getTsFilePath().compareTo(resource2.getTsFilePath()); + } + private void flushTsFilesForExtraction(DataRegion dataRegion) { LOGGER.info(DataNodePipeMessages.PIPE_START_TO_FLUSH_DATA_REGION, pipeName, dataRegionId); @@ -554,7 +659,8 @@ private void flushTsFilesForExtraction(DataRegion dataRegion) { // Since a large number of consensus pipes are not created at the same time, resulting in no // serious waiting for locks. Therefore, the flush operation is always performed for the // consensus pipe, and the lastFlushed timestamp is not updated here. - if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) { + if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX) + || shouldUseHistoricalTsFileFlushTimeOrder()) { dataRegion.syncCloseAllWorkingTsFileProcessors(); } else { dataRegion.asyncCloseAllWorkingTsFileProcessors(); @@ -881,27 +987,36 @@ public synchronized Event supply() { return null; } - final PersistentResource resource = pendingQueue.peek(); - if (resource == null) { - return supplyTerminateEvent(); - } + while (true) { + final PersistentResource resource = pendingQueue.peek(); + if (resource == null) { + if (shouldReportMaxHistoricalProgressIndex) { + shouldReportMaxHistoricalProgressIndex = false; + return supplyProgressReportEvent(maxHistoricalProgressIndex); + } + return supplyTerminateEvent(); + } - if (resource instanceof TsFileResource) { - final TsFileResource tsFileResource = (TsFileResource) resource; - if (consumeSkippedHistoricalTsFileEventIfNecessary(tsFileResource)) { - clearReplicateIndexForResource(tsFileResource); + if (resource instanceof TsFileResource) { + final TsFileResource tsFileResource = (TsFileResource) resource; + if (consumeSkippedHistoricalTsFileEventIfNecessary(tsFileResource)) { + clearReplicateIndexForResource(tsFileResource); + pendingQueue.poll(); + if (shouldUseHistoricalTsFileFlushTimeOrder()) { + continue; + } + return supplyProgressReportEvent(tsFileResource.getMaxProgressIndex()); + } + + final Event event = supplyTsFileEvent(tsFileResource); pendingQueue.poll(); - return supplyProgressReportEvent(tsFileResource.getMaxProgressIndex()); + return event; } - final Event event = supplyTsFileEvent(tsFileResource); + final Event event = supplyDeletionEvent((DeletionResource) resource); pendingQueue.poll(); return event; } - - final Event event = supplyDeletionEvent((DeletionResource) resource); - pendingQueue.poll(); - return event; } private Event supplyTerminateEvent() { @@ -982,7 +1097,9 @@ protected Event supplyProgressReportEvent(final ProgressIndex progressIndex) { protected Event supplyTsFileEvent(final TsFileResource resource) { if (!filteredTsFileResources2TableNames.containsKey(resource)) { clearReplicateIndexForResource(resource); - return supplyProgressReportEvent(resource.getMaxProgressIndex()); + return shouldUseHistoricalTsFileFlushTimeOrder() + ? null + : supplyProgressReportEvent(resource.getMaxProgressIndex()); } boolean shouldUnpinResource = false; @@ -1010,6 +1127,10 @@ protected Event supplyTsFileEvent(final TsFileResource resource) { historicalDataExtractionStartTime, historicalDataExtractionEndTime); + if (shouldUseHistoricalTsFileFlushTimeOrder()) { + event.skipReportOnCommitAndGeneratedEvents(); + } + // if using IoTV2, assign a replicateIndex for this event if (shouldAssignReplicateIndexForIoTV2(event)) { event.setReplicateIndexForIoTV2(assignReplicateIndexForResource(resource)); @@ -1081,7 +1202,6 @@ private Event supplyDeletionEvent(final DeletionResource deletionResource) { cliHostname, skipIfNoPrivileges, false); - // if using IoTV2, assign a replicateIndex for this historical deletion event if (shouldAssignReplicateIndexForIoTV2(event)) { event.setReplicateIndexForIoTV2(assignReplicateIndexForResource(deletionResource)); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java index cfe35ef041de9..f5c3f1fa05c8c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java @@ -37,6 +37,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTablePatternParser; import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTreePatternParser; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; @@ -53,6 +54,7 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; import java.time.LocalDate; import java.util.ArrayList; @@ -262,6 +264,63 @@ private void createTablet() { new Tablet(deviceId, Arrays.asList(schemas), times, values, null, times.length); } + @Test + public void markAsNeedToReportShouldInheritSourceTsFileGeneratedReportSkipping() + throws Exception { + final PipeTsFileInsertionEvent sourceEvent = Mockito.mock(PipeTsFileInsertionEvent.class); + Mockito.when(sourceEvent.shouldReportGeneratedEventsOnCommit()).thenReturn(true); + final PipeRawTabletInsertionEvent tabletEvent = + new PipeRawTabletInsertionEvent( + false, + null, + null, + null, + tabletForInsertTabletNode, + false, + null, + 0, + null, + sourceEvent, + false); + + tabletEvent.markAsNeedToReport(); + Assert.assertTrue(tabletEvent.isShouldReportOnCommit()); + + Mockito.when(sourceEvent.shouldReportGeneratedEventsOnCommit()).thenReturn(false); + final PipeRawTabletInsertionEvent skippedTabletEvent = + new PipeRawTabletInsertionEvent( + false, + null, + null, + null, + tabletForInsertTabletNode, + false, + null, + 0, + null, + sourceEvent, + false); + + skippedTabletEvent.markAsNeedToReport(); + Assert.assertFalse(skippedTabletEvent.isShouldReportOnCommit()); + + final PipeRawTabletInsertionEvent constructorSkippedTabletEvent = + new PipeRawTabletInsertionEvent( + false, + null, + null, + null, + tabletForInsertTabletNode, + false, + null, + 0, + null, + sourceEvent, + true); + + Assert.assertFalse(constructorSkippedTabletEvent.isShouldReportOnCommit()); + } + @Test public void convertToTabletForTest() throws Exception { TabletInsertionEventTreePatternParser container1 = 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 edffbb5b31c1f..53cf442c5239f 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 @@ -181,6 +181,152 @@ public void testSupplyRetriesSameTsFileAfterEventCreationFailure() throws Except } } + @Test + public void testHistoricalTsFileFlushTimeOrderDefaultsToTrue() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + + source.validate(new PipeParameterValidator(new PipeParameters(new HashMap<>()))); + + Assert.assertTrue((Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime")); + } + + @Test + public void testHistoricalTsFileFlushTimeOrderSortsOlderFlushTimeFirst() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileOrder").toFile(); + + try { + final TsFileResource seqOlderFlushTimeNewerVersion = + createTsFileResource(tempDir, "50-3-0-0.tsfile"); + seqOlderFlushTimeNewerVersion.setSeq(true); + final TsFileResource seqSameFlushTimeOlderVersion = + createTsFileResource(tempDir, "100-1-0-0.tsfile"); + seqSameFlushTimeOlderVersion.setSeq(true); + final TsFileResource seqSameFlushTimeNewerVersion = + createTsFileResource(tempDir, "100-2-0-0.tsfile"); + seqSameFlushTimeNewerVersion.setSeq(true); + final TsFileResource unseqOldestFlushTime = createTsFileResource(tempDir, "1-1-0-0.tsfile"); + unseqOldestFlushTime.setSeq(false); + + setPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldExtractDeletion", false); + setPrivateField(source, "startIndex", MinimumProgressIndex.INSTANCE); + + final List resources = + new ArrayList<>( + Arrays.asList( + unseqOldestFlushTime, + seqSameFlushTimeNewerVersion, + seqSameFlushTimeOlderVersion, + seqOlderFlushTimeNewerVersion)); + sortExtractedResources(source, resources); + + Assert.assertEquals( + Arrays.asList( + seqOlderFlushTimeNewerVersion, + seqSameFlushTimeOlderVersion, + seqSameFlushTimeNewerVersion, + unseqOldestFlushTime), + resources); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + public void testHistoricalTsFileFlushTimeOrderCanBeDisabled() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final PipeParameters parameters = + new PipeParameters( + new HashMap() { + { + put( + PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + Boolean.FALSE.toString()); + } + }); + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileProgressOrder").toFile(); + + try { + source.validate(new PipeParameterValidator(parameters)); + final TsFileResource earlierProgressIndex = createTsFileResource(tempDir, "300-1-0-0.tsfile"); + earlierProgressIndex.updateProgressIndex(new SimpleProgressIndex(0, 1)); + final TsFileResource laterProgressIndex = createTsFileResource(tempDir, "100-1-0-0.tsfile"); + laterProgressIndex.updateProgressIndex(new SimpleProgressIndex(0, 2)); + + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldExtractDeletion", false); + setPrivateField(source, "startIndex", MinimumProgressIndex.INSTANCE); + + final List resources = + new ArrayList<>(Arrays.asList(laterProgressIndex, earlierProgressIndex)); + sortExtractedResources(source, resources); + + Assert.assertFalse( + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime")); + Assert.assertEquals(Arrays.asList(earlierProgressIndex, laterProgressIndex), resources); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + public void testFlushTimeOrderReportsProgressAfterAllHistoricalResources() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final ProgressIndex expectedProgressIndex = new SimpleProgressIndex(0, 10); + + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "pipeName", "pipe"); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 1); + setPrivateField(source, "pipeTaskMeta", new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + setPrivateField(source, "pendingQueue", new ArrayDeque()); + setPrivateField(source, "maxHistoricalProgressIndex", expectedProgressIndex); + setPrivateField(source, "shouldReportMaxHistoricalProgressIndex", true); + + final Event event = source.supply(); + + Assert.assertTrue(event instanceof ProgressReportEvent); + Assert.assertEquals(expectedProgressIndex, ((ProgressReportEvent) event).getProgressIndex()); + Assert.assertFalse((Boolean) getPrivateField(source, "shouldReportMaxHistoricalProgressIndex")); + } + + @Test + @SuppressWarnings("unchecked") + public void testFlushTimeOrderProgressOnlyCoversSelectedResources() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileSelectedProgress").toFile(); + + try { + final TsFileResource selectedResource = createTsFileResource(tempDir, "100-1-0-0.tsfile"); + selectedResource.updateProgressIndex(new SimpleProgressIndex(0, 1)); + final TsFileResource filteredResource = createTsFileResource(tempDir, "200-1-0-0.tsfile"); + filteredResource.updateProgressIndex(new SimpleProgressIndex(0, 100)); + + ((Map>) + getPrivateField(source, "filteredTsFileResources2TableNames")) + .put(selectedResource, Set.of()); + + final List resources = + new ArrayList<>(Arrays.asList(filteredResource, selectedResource)); + prepareResourcesForHistoricalTsFileFlushTimeOrder(source, resources); + + Assert.assertEquals(Arrays.asList(selectedResource), resources); + Assert.assertEquals( + new SimpleProgressIndex(0, 1), getPrivateField(source, "maxHistoricalProgressIndex")); + Assert.assertTrue( + (Boolean) getPrivateField(source, "shouldReportMaxHistoricalProgressIndex")); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testReplicateIndexShouldBeStableBeforeResourceConsumed() throws Exception { final TestablePipeHistoricalDataRegionTsFileAndDeletionSource source = @@ -302,6 +448,28 @@ private static ProgressIndex hybridProgressIndex( return result; } + private static void sortExtractedResources( + final PipeHistoricalDataRegionTsFileAndDeletionSource source, + final List resources) + throws ReflectiveOperationException { + final Method method = + PipeHistoricalDataRegionTsFileAndDeletionSource.class.getDeclaredMethod( + "sortExtractedResources", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + + private static void prepareResourcesForHistoricalTsFileFlushTimeOrder( + final PipeHistoricalDataRegionTsFileAndDeletionSource source, + final List resources) + throws ReflectiveOperationException { + final Method method = + PipeHistoricalDataRegionTsFileAndDeletionSource.class.getDeclaredMethod( + "prepareResourcesForHistoricalTsFileFlushTimeOrder", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + private static void setPrivateField( final PipeHistoricalDataRegionTsFileAndDeletionSource source, final String fieldName, diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java index e755ef3c6019e..a44a34d14fee6 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java @@ -101,6 +101,11 @@ public class PipeSourceConstant { public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE = "path"; public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_ALL_VALUE = "all"; public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_DEFAULT_VALUE = ""; + public static final String EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY = + "extractor.history.tsfile.order-by-flush-time"; + public static final String SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY = + "source.history.tsfile.order-by-flush-time"; + public static final boolean EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE = true; public static final String EXTRACTOR_MODS_ENABLE_KEY = "extractor.mods.enable"; public static final String SOURCE_MODS_ENABLE_KEY = "source.mods.enable"; public static final boolean EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE = false; From 528c372786e0aaccc1f4dc00d23dbf68a6a78d69 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 3 Jul 2026 10:44:11 +0800 Subject: [PATCH 2/4] Pipe: order historical TsFiles by query priority --- .../dataregion/IoTDBDataRegionSource.java | 8 +- ...icalDataRegionTsFileAndDeletionSource.java | 82 ++++++------ ...DataRegionTsFileAndDeletionSourceTest.java | 123 +++++++++++++----- .../config/constant/PipeSourceConstant.java | 10 +- 4 files changed, 143 insertions(+), 80 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java index 66944f776d0f8..7e8b5a1f71f25 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java @@ -75,7 +75,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY; -import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STREAMING_DEFAULT_VALUE; @@ -108,7 +108,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY; -import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STREAMING_KEY; @@ -189,12 +189,12 @@ public void validate(final PipeParameterValidator validator) throws Exception { .validateAttributeValueRange( SOURCE_REALTIME_ENABLE_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString()) .validateAttributeValueRange( - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString()) .validateAttributeValueRange( - SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, true, Boolean.TRUE.toString(), Boolean.FALSE.toString()) 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 b1faf4efa7f6e..ad3b360aec28f 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 @@ -59,6 +59,7 @@ import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.MergeReaderPriority; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; @@ -99,8 +100,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_TIME_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY; -import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE; -import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STRICT_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STRICT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_DEFAULT_VALUE; @@ -113,7 +114,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY; -import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STRICT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_KEY; @@ -153,8 +154,8 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource private boolean sloppyTimeRange; // true to disable time range filter after extraction private boolean sloppyPattern; // true to disable pattern filter after extraction - private boolean shouldOrderHistoricalTsFileByFlushTime = - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE; + private boolean shouldOrderHistoricalTsFileByQueryPriority = + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE; private Pair listeningOptionPair; private boolean shouldExtractInsertion; @@ -193,12 +194,12 @@ public void validate(final PipeParameterValidator validator) { throw new PipeParameterNotValidException(e.getMessage()); } - shouldOrderHistoricalTsFileByFlushTime = + shouldOrderHistoricalTsFileByQueryPriority = parameters.getBooleanOrDefault( Arrays.asList( - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, - SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY), - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE); + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, + SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY), + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE); if (parameters.hasAnyAttributes(EXTRACTOR_MODE_STRICT_KEY, SOURCE_MODE_STRICT_KEY)) { final boolean isStrictMode = @@ -326,12 +327,12 @@ public void customize( throws IllegalPathException { shouldExtractInsertion = listeningOptionPair.getLeft(); shouldExtractDeletion = listeningOptionPair.getRight(); - shouldOrderHistoricalTsFileByFlushTime = + shouldOrderHistoricalTsFileByQueryPriority = parameters.getBooleanOrDefault( Arrays.asList( - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, - SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY), - EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE); + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, + SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY), + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE); final PipeRuntimeEnvironment environment = configuration.getRuntimeEnvironment(); @@ -542,8 +543,8 @@ public synchronized void start() { .ifPresent(manager -> extractDeletions(manager, originalResourceList)); } - if (shouldUseHistoricalTsFileFlushTimeOrder()) { - prepareResourcesForHistoricalTsFileFlushTimeOrder(originalResourceList); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + prepareResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); } // Sort tsFileResource and deletionResource @@ -569,19 +570,19 @@ public synchronized void start() { } } - private boolean shouldUseHistoricalTsFileFlushTimeOrder() { + private boolean shouldUseHistoricalTsFileQueryPriorityOrder() { // Deletion resources only carry progressIndex. Keep the old progressIndex order when deletions // are extracted together with TsFiles so insertion/deletion ordering semantics are unchanged. - return shouldOrderHistoricalTsFileByFlushTime + return shouldOrderHistoricalTsFileByQueryPriority && shouldExtractInsertion && !shouldExtractDeletion; } - private void prepareResourcesForHistoricalTsFileFlushTimeOrder( + private void prepareResourcesForHistoricalTsFileQueryPriorityOrder( final List resourceList) { - // Flush-time order is intentionally not compatible with progressIndex order, so report progress - // only after all selected historical TsFiles are supplied. This prefers possible retransmission - // over losing overwrite semantics. + // Query-priority order is intentionally not compatible with progressIndex order, so report + // progress only after all selected historical TsFiles are supplied. This prefers possible + // retransmission over losing overwrite semantics. resourceList.removeIf( resource -> resource instanceof TsFileResource @@ -601,14 +602,14 @@ private void updateMaxHistoricalProgressIndex(final List res } private void sortExtractedResources(final List resourceList) { - if (shouldUseHistoricalTsFileFlushTimeOrder()) { - // Send TsFiles in source-side file creation order. For duplicated points, older files are - // loaded first on the receiver and newer files are loaded later to preserve overwrite - // semantics. + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + // Send TsFiles from lower query/compaction priority to higher priority. For duplicated + // points, covered files are loaded first on the receiver and covering files are loaded later + // to preserve overwrite semantics. resourceList.sort( (o1, o2) -> o1 instanceof TsFileResource && o2 instanceof TsFileResource - ? compareTsFileResourcesByFlushTime((TsFileResource) o1, (TsFileResource) o2) + ? compareTsFileResourcesByQueryPriority((TsFileResource) o1, (TsFileResource) o2) : comparePersistentResourcesByProgressIndex(o1, o2)); return; } @@ -625,18 +626,17 @@ private int comparePersistentResourcesByProgressIndex( return resource1.getProgressIndex().topologicalCompareTo(resource2.getProgressIndex()); } - private int compareTsFileResourcesByFlushTime( + private int compareTsFileResourcesByQueryPriority( final TsFileResource resource1, final TsFileResource resource2) { - if (resource1.isSeq() != resource2.isSeq()) { - return resource1.isSeq() ? -1 : 1; - } - - int result = Long.compare(resource1.getTsFileID().timestamp, resource2.getTsFileID().timestamp); - if (result != 0) { - return result; - } - - result = Long.compare(resource1.getVersion(), resource2.getVersion()); + int result = + new MergeReaderPriority( + resource1.getTsFileID().timestamp, resource1.getVersion(), 0, resource1.isSeq()) + .compareTo( + new MergeReaderPriority( + resource2.getTsFileID().timestamp, + resource2.getVersion(), + 0, + resource2.isSeq())); if (result != 0) { return result; } @@ -660,7 +660,7 @@ private void flushTsFilesForExtraction(DataRegion dataRegion) { // serious waiting for locks. Therefore, the flush operation is always performed for the // consensus pipe, and the lastFlushed timestamp is not updated here. if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX) - || shouldUseHistoricalTsFileFlushTimeOrder()) { + || shouldUseHistoricalTsFileQueryPriorityOrder()) { dataRegion.syncCloseAllWorkingTsFileProcessors(); } else { dataRegion.asyncCloseAllWorkingTsFileProcessors(); @@ -1002,7 +1002,7 @@ public synchronized Event supply() { if (consumeSkippedHistoricalTsFileEventIfNecessary(tsFileResource)) { clearReplicateIndexForResource(tsFileResource); pendingQueue.poll(); - if (shouldUseHistoricalTsFileFlushTimeOrder()) { + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { continue; } return supplyProgressReportEvent(tsFileResource.getMaxProgressIndex()); @@ -1097,7 +1097,7 @@ protected Event supplyProgressReportEvent(final ProgressIndex progressIndex) { protected Event supplyTsFileEvent(final TsFileResource resource) { if (!filteredTsFileResources2TableNames.containsKey(resource)) { clearReplicateIndexForResource(resource); - return shouldUseHistoricalTsFileFlushTimeOrder() + return shouldUseHistoricalTsFileQueryPriorityOrder() ? null : supplyProgressReportEvent(resource.getMaxProgressIndex()); } @@ -1127,7 +1127,7 @@ protected Event supplyTsFileEvent(final TsFileResource resource) { historicalDataExtractionStartTime, historicalDataExtractionEndTime); - if (shouldUseHistoricalTsFileFlushTimeOrder()) { + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { event.skipReportOnCommitAndGeneratedEvents(); } 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 53cf442c5239f..804df8d5722ce 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 @@ -182,35 +182,40 @@ public void testSupplyRetriesSameTsFileAfterEventCreationFailure() throws Except } @Test - public void testHistoricalTsFileFlushTimeOrderDefaultsToTrue() throws Exception { + public void testHistoricalTsFileQueryPriorityOrderDefaultsToTrue() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); source.validate(new PipeParameterValidator(new PipeParameters(new HashMap<>()))); - Assert.assertTrue((Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime")); + Assert.assertTrue( + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); } @Test - public void testHistoricalTsFileFlushTimeOrderSortsOlderFlushTimeFirst() throws Exception { + public void testHistoricalTsFileQueryPriorityOrderMatchesQueryCoverage() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileOrder").toFile(); try { - final TsFileResource seqOlderFlushTimeNewerVersion = - createTsFileResource(tempDir, "50-3-0-0.tsfile"); - seqOlderFlushTimeNewerVersion.setSeq(true); - final TsFileResource seqSameFlushTimeOlderVersion = - createTsFileResource(tempDir, "100-1-0-0.tsfile"); - seqSameFlushTimeOlderVersion.setSeq(true); - final TsFileResource seqSameFlushTimeNewerVersion = + final TsFileResource seqLowerVersionNewerFileTimestamp = + createTsFileResource(tempDir, "300-1-0-0.tsfile"); + seqLowerVersionNewerFileTimestamp.setSeq(true); + final TsFileResource seqSameVersionOlderFileTimestamp = createTsFileResource(tempDir, "100-2-0-0.tsfile"); - seqSameFlushTimeNewerVersion.setSeq(true); - final TsFileResource unseqOldestFlushTime = createTsFileResource(tempDir, "1-1-0-0.tsfile"); - unseqOldestFlushTime.setSeq(false); + seqSameVersionOlderFileTimestamp.setSeq(true); + final TsFileResource seqSameVersionNewerFileTimestamp = + createTsFileResource(tempDir, "200-2-0-0.tsfile"); + seqSameVersionNewerFileTimestamp.setSeq(true); + final TsFileResource seqHigherVersionOlderFileTimestamp = + createTsFileResource(tempDir, "50-3-0-0.tsfile"); + seqHigherVersionOlderFileTimestamp.setSeq(true); + final TsFileResource unseqLowerVersionOldestFileTimestamp = + createTsFileResource(tempDir, "1-1-0-0.tsfile"); + unseqLowerVersionOldestFileTimestamp.setSeq(false); - setPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime", true); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); setPrivateField(source, "shouldExtractInsertion", true); setPrivateField(source, "shouldExtractDeletion", false); setPrivateField(source, "startIndex", MinimumProgressIndex.INSTANCE); @@ -218,18 +223,20 @@ public void testHistoricalTsFileFlushTimeOrderSortsOlderFlushTimeFirst() throws final List resources = new ArrayList<>( Arrays.asList( - unseqOldestFlushTime, - seqSameFlushTimeNewerVersion, - seqSameFlushTimeOlderVersion, - seqOlderFlushTimeNewerVersion)); + unseqLowerVersionOldestFileTimestamp, + seqHigherVersionOlderFileTimestamp, + seqSameVersionNewerFileTimestamp, + seqSameVersionOlderFileTimestamp, + seqLowerVersionNewerFileTimestamp)); sortExtractedResources(source, resources); Assert.assertEquals( Arrays.asList( - seqOlderFlushTimeNewerVersion, - seqSameFlushTimeOlderVersion, - seqSameFlushTimeNewerVersion, - unseqOldestFlushTime), + seqLowerVersionNewerFileTimestamp, + seqSameVersionOlderFileTimestamp, + seqSameVersionNewerFileTimestamp, + seqHigherVersionOlderFileTimestamp, + unseqLowerVersionOldestFileTimestamp), resources); } finally { FileUtils.deleteFileOrDirectory(tempDir); @@ -237,7 +244,7 @@ public void testHistoricalTsFileFlushTimeOrderSortsOlderFlushTimeFirst() throws } @Test - public void testHistoricalTsFileFlushTimeOrderCanBeDisabled() throws Exception { + public void testHistoricalTsFileQueryPriorityOrderCanBeDisabled() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); final PipeParameters parameters = @@ -245,7 +252,7 @@ public void testHistoricalTsFileFlushTimeOrderCanBeDisabled() throws Exception { new HashMap() { { put( - PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY, + PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, Boolean.FALSE.toString()); } }); @@ -267,7 +274,7 @@ public void testHistoricalTsFileFlushTimeOrderCanBeDisabled() throws Exception { sortExtractedResources(source, resources); Assert.assertFalse( - (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByFlushTime")); + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); Assert.assertEquals(Arrays.asList(earlierProgressIndex, laterProgressIndex), resources); } finally { FileUtils.deleteFileOrDirectory(tempDir); @@ -275,7 +282,63 @@ public void testHistoricalTsFileFlushTimeOrderCanBeDisabled() throws Exception { } @Test - public void testFlushTimeOrderReportsProgressAfterAllHistoricalResources() throws Exception { + public void testHistoricalTsFileQueryPriorityOrderCanBeDisabledByExtractorKey() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final PipeParameters parameters = + new PipeParameters( + new HashMap() { + { + put( + PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY, + Boolean.FALSE.toString()); + } + }); + + source.validate(new PipeParameterValidator(parameters)); + + Assert.assertFalse( + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); + } + + @Test + public void testHistoricalTsFileQueryPriorityOrderFallsBackWhenDeletionExtracted() + throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final File tempDir = + Files.createTempDirectory("pipeHistoricalTsFileDeletionProgressOrder").toFile(); + + try { + final TsFileResource higherQueryPriorityEarlierProgress = + createTsFileResource(tempDir, "300-3-0-0.tsfile"); + higherQueryPriorityEarlierProgress.setSeq(true); + higherQueryPriorityEarlierProgress.updateProgressIndex(new SimpleProgressIndex(0, 1)); + final TsFileResource lowerQueryPriorityLaterProgress = + createTsFileResource(tempDir, "100-1-0-0.tsfile"); + lowerQueryPriorityLaterProgress.setSeq(true); + lowerQueryPriorityLaterProgress.updateProgressIndex(new SimpleProgressIndex(0, 2)); + + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldExtractDeletion", true); + setPrivateField(source, "startIndex", MinimumProgressIndex.INSTANCE); + + final List resources = + new ArrayList<>( + Arrays.asList(lowerQueryPriorityLaterProgress, higherQueryPriorityEarlierProgress)); + sortExtractedResources(source, resources); + + Assert.assertEquals( + Arrays.asList(higherQueryPriorityEarlierProgress, lowerQueryPriorityLaterProgress), + resources); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + public void testQueryPriorityOrderReportsProgressAfterAllHistoricalResources() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); final ProgressIndex expectedProgressIndex = new SimpleProgressIndex(0, 10); @@ -298,7 +361,7 @@ public void testFlushTimeOrderReportsProgressAfterAllHistoricalResources() throw @Test @SuppressWarnings("unchecked") - public void testFlushTimeOrderProgressOnlyCoversSelectedResources() throws Exception { + public void testQueryPriorityOrderProgressOnlyCoversSelectedResources() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileSelectedProgress").toFile(); @@ -315,7 +378,7 @@ public void testFlushTimeOrderProgressOnlyCoversSelectedResources() throws Excep final List resources = new ArrayList<>(Arrays.asList(filteredResource, selectedResource)); - prepareResourcesForHistoricalTsFileFlushTimeOrder(source, resources); + prepareResourcesForHistoricalTsFileQueryPriorityOrder(source, resources); Assert.assertEquals(Arrays.asList(selectedResource), resources); Assert.assertEquals( @@ -459,13 +522,13 @@ private static void sortExtractedResources( method.invoke(source, resources); } - private static void prepareResourcesForHistoricalTsFileFlushTimeOrder( + private static void prepareResourcesForHistoricalTsFileQueryPriorityOrder( final PipeHistoricalDataRegionTsFileAndDeletionSource source, final List resources) throws ReflectiveOperationException { final Method method = PipeHistoricalDataRegionTsFileAndDeletionSource.class.getDeclaredMethod( - "prepareResourcesForHistoricalTsFileFlushTimeOrder", List.class); + "prepareResourcesForHistoricalTsFileQueryPriorityOrder", List.class); method.setAccessible(true); method.invoke(source, resources); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java index a44a34d14fee6..523fa9b0698f5 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java @@ -101,11 +101,11 @@ public class PipeSourceConstant { public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE = "path"; public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_ALL_VALUE = "all"; public static final String EXTRACTOR_HISTORY_LOOSE_RANGE_DEFAULT_VALUE = ""; - public static final String EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY = - "extractor.history.tsfile.order-by-flush-time"; - public static final String SOURCE_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_KEY = - "source.history.tsfile.order-by-flush-time"; - public static final boolean EXTRACTOR_HISTORY_TSFILE_ORDER_BY_FLUSH_TIME_DEFAULT_VALUE = true; + public static final String EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY = + "extractor.history.tsfile.order-by-query-priority"; + public static final String SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY = + "source.history.tsfile.order-by-query-priority"; + public static final boolean EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE = true; public static final String EXTRACTOR_MODS_ENABLE_KEY = "extractor.mods.enable"; public static final String SOURCE_MODS_ENABLE_KEY = "source.mods.enable"; public static final boolean EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE = false; From 6eea572961e94dc71a052f19fe13ec0b027cd10e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 7 Jul 2026 11:57:21 +0800 Subject: [PATCH 3/4] Pipe: report safe query-order historical progress --- ...icalDataRegionTsFileAndDeletionSource.java | 87 +++++++++++++++++-- ...DataRegionTsFileAndDeletionSourceTest.java | 84 ++++++++++++++++++ 2 files changed, 163 insertions(+), 8 deletions(-) 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 ad3b360aec28f..cebd497fa333f 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 @@ -81,6 +81,7 @@ 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.Objects; @@ -177,7 +178,10 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource new HashMap<>(); private final Map pendingResource2ReplicateIndexForIoTV2 = new HashMap<>(); + private final Set historicalProgressReportResources = new HashSet<>(); private ProgressIndex maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + private ProgressIndex maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE; + private ProgressIndex pendingHistoricalProgressIndexToReport; private boolean shouldReportMaxHistoricalProgressIndex = false; private int extractedHistoricalTsFileCount = 0; private int extractedHistoricalDeletionCount = 0; @@ -519,7 +523,10 @@ public synchronized void start() { extractedHistoricalTsFileCount = 0; extractedHistoricalDeletionCount = 0; maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE; + pendingHistoricalProgressIndexToReport = null; shouldReportMaxHistoricalProgressIndex = false; + historicalProgressReportResources.clear(); final DataRegion dataRegion = StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId)); @@ -543,15 +550,17 @@ public synchronized void start() { .ifPresent(manager -> extractDeletions(manager, originalResourceList)); } - if (shouldUseHistoricalTsFileQueryPriorityOrder()) { - prepareResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); - } - // Sort tsFileResource and deletionResource long startTime = System.currentTimeMillis(); LOGGER.info( DataNodePipeMessages.PIPE_START_TO_SORT_ALL_EXTRACTED_RESOURCES, pipeName, dataRegionId); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + prepareResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); + } sortExtractedResources(originalResourceList); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); + } pendingQueue = new ArrayDeque<>(originalResourceList); PipeTerminateEvent.initializeHistoricalTransferSummary( pipeName, @@ -580,9 +589,8 @@ private boolean shouldUseHistoricalTsFileQueryPriorityOrder() { private void prepareResourcesForHistoricalTsFileQueryPriorityOrder( final List resourceList) { - // Query-priority order is intentionally not compatible with progressIndex order, so report - // progress only after all selected historical TsFiles are supplied. This prefers possible - // retransmission over losing overwrite semantics. + // Query-priority order is intentionally not compatible with progressIndex order, so only + // selected historical TsFiles should participate in query-order progress reports. resourceList.removeIf( resource -> resource instanceof TsFileResource @@ -591,6 +599,40 @@ private void prepareResourcesForHistoricalTsFileQueryPriorityOrder( shouldReportMaxHistoricalProgressIndex = !resourceList.isEmpty(); } + private void prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder( + final List resourceList) { + historicalProgressReportResources.clear(); + final List remainingMinimalProgressIndexes = new ArrayList<>(); + for (int i = resourceList.size() - 1; i >= 0; --i) { + final PersistentResource resource = resourceList.get(i); + final ProgressIndex progressIndex = resource.getProgressIndex(); + if (Objects.isNull(progressIndex)) { + continue; + } + + // A resource can report progress only if that progress will not cover any later resource. + // Otherwise a restart may skip untransferred data. + if (remainingMinimalProgressIndexes.stream().noneMatch(progressIndex::isEqualOrAfter)) { + historicalProgressReportResources.add(resource); + } + updateRemainingMinimalProgressIndexes(remainingMinimalProgressIndexes, progressIndex); + } + } + + private void updateRemainingMinimalProgressIndexes( + final List remainingMinimalProgressIndexes, + final ProgressIndex progressIndex) { + if (remainingMinimalProgressIndexes.stream().anyMatch(progressIndex::isEqualOrAfter)) { + return; + } + + // Keep only suffix minimal progress indexes. They are sufficient to test whether a new + // progress index covers any remaining resource. + remainingMinimalProgressIndexes.removeIf( + minimalProgressIndex -> minimalProgressIndex.isEqualOrAfter(progressIndex)); + remainingMinimalProgressIndexes.add(progressIndex); + } + private void updateMaxHistoricalProgressIndex(final List resourceList) { for (final PersistentResource resource : resourceList) { final ProgressIndex progressIndex = resource.getProgressIndex(); @@ -983,6 +1025,12 @@ public synchronized Event supply() { start(); } + if (Objects.nonNull(pendingHistoricalProgressIndexToReport)) { + final ProgressIndex progressIndex = pendingHistoricalProgressIndexToReport; + pendingHistoricalProgressIndexToReport = null; + return supplyHistoricalProgressReportEvent(progressIndex); + } + if (Objects.isNull(pendingQueue)) { return null; } @@ -992,7 +1040,10 @@ public synchronized Event supply() { if (resource == null) { if (shouldReportMaxHistoricalProgressIndex) { shouldReportMaxHistoricalProgressIndex = false; - return supplyProgressReportEvent(maxHistoricalProgressIndex); + if (!maxSuppliedHistoricalProgressReportIndex.isEqualOrAfter( + maxHistoricalProgressIndex)) { + return supplyHistoricalProgressReportEvent(maxHistoricalProgressIndex); + } } return supplyTerminateEvent(); } @@ -1003,6 +1054,9 @@ public synchronized Event supply() { clearReplicateIndexForResource(tsFileResource); pendingQueue.poll(); if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + if (shouldReportHistoricalProgressAfterResource(tsFileResource)) { + return supplyHistoricalProgressReportEvent(tsFileResource.getMaxProgressIndex()); + } continue; } return supplyProgressReportEvent(tsFileResource.getMaxProgressIndex()); @@ -1010,6 +1064,9 @@ public synchronized Event supply() { final Event event = supplyTsFileEvent(tsFileResource); pendingQueue.poll(); + if (Objects.nonNull(event) && shouldReportHistoricalProgressAfterResource(tsFileResource)) { + pendingHistoricalProgressIndexToReport = tsFileResource.getMaxProgressIndex(); + } return event; } @@ -1019,6 +1076,11 @@ public synchronized Event supply() { } } + private boolean shouldReportHistoricalProgressAfterResource(final PersistentResource resource) { + return shouldUseHistoricalTsFileQueryPriorityOrder() + && historicalProgressReportResources.remove(resource); + } + private Event supplyTerminateEvent() { final PipeTerminateEvent.HistoricalTransferSummary historicalTransferSummary = PipeTerminateEvent.snapshotHistoricalTransferSummary(pipeName, creationTime, dataRegionId); @@ -1094,6 +1156,13 @@ protected Event supplyProgressReportEvent(final ProgressIndex progressIndex) { return isReferenceCountIncreased ? progressReportEvent : null; } + private Event supplyHistoricalProgressReportEvent(final ProgressIndex progressIndex) { + maxSuppliedHistoricalProgressReportIndex = + maxSuppliedHistoricalProgressReportIndex.updateToMinimumEqualOrIsAfterProgressIndex( + progressIndex); + return supplyProgressReportEvent(progressIndex); + } + protected Event supplyTsFileEvent(final TsFileResource resource) { if (!filteredTsFileResources2TableNames.containsKey(resource)) { clearReplicateIndexForResource(resource); @@ -1293,5 +1362,7 @@ public synchronized void close() { pendingQueue = null; } pendingResource2ReplicateIndexForIoTV2.clear(); + historicalProgressReportResources.clear(); + pendingHistoricalProgressIndexToReport = null; } } 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 804df8d5722ce..3a69e6d0d96f5 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 @@ -390,6 +390,79 @@ public void testQueryPriorityOrderProgressOnlyCoversSelectedResources() throws E } } + @Test + @SuppressWarnings("unchecked") + public void testQueryPriorityOrderPreparesIncrementalSafeProgressReports() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final File tempDir = + Files.createTempDirectory("pipeHistoricalTsFileIncrementalProgress").toFile(); + + try { + final TsFileResource progress1 = createTsFileResource(tempDir, "1.tsfile"); + progress1.updateProgressIndex(new SimpleProgressIndex(0, 1)); + final TsFileResource progress2 = createTsFileResource(tempDir, "2.tsfile"); + progress2.updateProgressIndex(new SimpleProgressIndex(0, 2)); + final TsFileResource progress4 = createTsFileResource(tempDir, "4.tsfile"); + progress4.updateProgressIndex(new SimpleProgressIndex(0, 4)); + final TsFileResource progress3 = createTsFileResource(tempDir, "3.tsfile"); + progress3.updateProgressIndex(new SimpleProgressIndex(0, 3)); + final TsFileResource progress5 = createTsFileResource(tempDir, "5.tsfile"); + progress5.updateProgressIndex(new SimpleProgressIndex(0, 5)); + + final List resources = + new ArrayList<>(Arrays.asList(progress1, progress2, progress4, progress3, progress5)); + prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder(source, resources); + + Assert.assertEquals( + new HashSet<>(Arrays.asList(progress1, progress2, progress3, progress5)), + getPrivateField(source, "historicalProgressReportResources")); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testQueryPriorityOrderSuppliesProgressAfterSafeTsFileEvent() throws Exception { + final TestablePipeHistoricalDataRegionTsFileAndDeletionSource source = + new TestablePipeHistoricalDataRegionTsFileAndDeletionSource(); + final Event expectedEvent = new Event() {}; + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileSafeProgress").toFile(); + + try { + final TsFileResource firstResource = createTsFileResource(tempDir, "first.tsfile"); + firstResource.updateProgressIndex(new SimpleProgressIndex(0, 1)); + final TsFileResource secondResource = createTsFileResource(tempDir, "second.tsfile"); + secondResource.updateProgressIndex(new SimpleProgressIndex(0, 2)); + + source.setSuppliedEvent(expectedEvent); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "pipeName", "pipe"); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "pipeTaskMeta", new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldExtractDeletion", false); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque(Arrays.asList(firstResource, secondResource))); + ((Set) getPrivateField(source, "historicalProgressReportResources")) + .add(firstResource); + + Assert.assertSame(expectedEvent, source.supply()); + final Event progressEvent = source.supply(); + Assert.assertTrue(progressEvent instanceof ProgressReportEvent); + Assert.assertEquals( + firstResource.getMaxProgressIndex(), + ((ProgressReportEvent) progressEvent).getProgressIndex()); + Assert.assertSame(expectedEvent, source.supply()); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testReplicateIndexShouldBeStableBeforeResourceConsumed() throws Exception { final TestablePipeHistoricalDataRegionTsFileAndDeletionSource source = @@ -533,6 +606,17 @@ private static void prepareResourcesForHistoricalTsFileQueryPriorityOrder( method.invoke(source, resources); } + private static void prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder( + final PipeHistoricalDataRegionTsFileAndDeletionSource source, + final List resources) + throws ReflectiveOperationException { + final Method method = + PipeHistoricalDataRegionTsFileAndDeletionSource.class.getDeclaredMethod( + "prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + private static void setPrivateField( final PipeHistoricalDataRegionTsFileAndDeletionSource source, final String fieldName, From dd886fa82826559218d878f7436d932378881b77 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:08:45 +0800 Subject: [PATCH 4/4] Pipe: report time-partition historical progress --- ...icalDataRegionTsFileAndDeletionSource.java | 187 ++++++++--- ...DataRegionTsFileAndDeletionSourceTest.java | 178 ++++++++++- .../TsFileResourceProgressIndexTest.java | 42 +++ .../consensus/index/ProgressIndexType.java | 8 +- .../impl/TimePartitionProgressIndex.java | 292 ++++++++++++++++++ .../commons/pipe/task/PipeMetaDeSerTest.java | 11 + 6 files changed, 664 insertions(+), 54 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimePartitionProgressIndex.java 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 cebd497fa333f..527acb5a8ec4b 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent; @@ -460,41 +461,72 @@ public void customize( * * @return recoverProgressIndex dedicated in local DataNodeId or origin for fallback. */ - private ProgressIndex tryToExtractLocalProgressIndexForIoTV2(ProgressIndex origin) { - // There are only 2 cases: - // 1. origin is RecoverProgressIndex + private ProgressIndex tryToExtractLocalProgressIndexForIoTV2(final ProgressIndex origin) { + return tryToExtractLocalProgressIndexForIoTV2(origin, true); + } + + private ProgressIndex tryToExtractLocalProgressIndexForIoTV2( + final ProgressIndex origin, final boolean shouldWarnUnexpectedType) { + if (Objects.isNull(origin)) { + return MinimumProgressIndex.INSTANCE; + } + + if (origin instanceof StateProgressIndex) { + final StateProgressIndex stateProgressIndex = (StateProgressIndex) origin; + return new StateProgressIndex( + stateProgressIndex.getVersion(), + stateProgressIndex.getState(), + tryToExtractLocalProgressIndexForIoTV2( + stateProgressIndex.getInnerProgressIndex(), shouldWarnUnexpectedType)); + } + if (origin instanceof RecoverProgressIndex) { - RecoverProgressIndex toBeTransformed = (RecoverProgressIndex) origin; - return extractRecoverProgressIndex(toBeTransformed); + return extractRecoverProgressIndex((RecoverProgressIndex) origin); } - // 2. origin is HybridProgressIndex - else if (origin instanceof HybridProgressIndex) { - HybridProgressIndex toBeTransformed = (HybridProgressIndex) origin; - // if hybridProgressIndex contains recoverProgressIndex, which is what we expected. - if (toBeTransformed - .getType2Index() - .containsKey(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType())) { - // 2.1. transform recoverProgressIndex - RecoverProgressIndex specificToBeTransformed = - (RecoverProgressIndex) - toBeTransformed - .getType2Index() - .get(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType()); - return extractRecoverProgressIndex(specificToBeTransformed); + + if (origin instanceof TimePartitionProgressIndex) { + return new TimePartitionProgressIndex( + ((TimePartitionProgressIndex) origin) + .getTimePartitionId2ProgressIndex().entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + entry -> + tryToExtractLocalProgressIndexForIoTV2(entry.getValue(), false)))); + } + + if (origin instanceof HybridProgressIndex) { + final Map type2Index = ((HybridProgressIndex) origin).getType2Index(); + ProgressIndex result = null; + if (type2Index.containsKey(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType())) { + result = + extractRecoverProgressIndex( + (RecoverProgressIndex) + type2Index.get(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType())); } - // if hybridProgressIndex doesn't contain recoverProgressIndex, which is not what we expected, - // fallback. - return origin; - } else { - // fallback + if (type2Index.containsKey(ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX.getType())) { + final ProgressIndex timePartitionProgressIndex = + tryToExtractLocalProgressIndexForIoTV2( + type2Index.get(ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX.getType()), false); + result = + Objects.isNull(result) + ? timePartitionProgressIndex + : result.updateToMinimumEqualOrIsAfterProgressIndex(timePartitionProgressIndex); + } + if (Objects.nonNull(result)) { + return result; + } + } + + if (shouldWarnUnexpectedType) { LOGGER.warn( DataNodePipeMessages.PIPE_UNEXPECTED_PROGRESSINDEX_TYPE_FALLBACK_TO_ORIGIN, pipeName, dataRegionId, origin.getType(), origin); - return origin; } + return origin; } private ProgressIndex extractRecoverProgressIndex(RecoverProgressIndex toBeTransformed) { @@ -602,16 +634,26 @@ private void prepareResourcesForHistoricalTsFileQueryPriorityOrder( private void prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder( final List resourceList) { historicalProgressReportResources.clear(); - final List remainingMinimalProgressIndexes = new ArrayList<>(); + final Map> timePartitionId2RemainingMinimalProgressIndexes = + new HashMap<>(); for (int i = resourceList.size() - 1; i >= 0; --i) { final PersistentResource resource = resourceList.get(i); + if (!(resource instanceof TsFileResource)) { + continue; + } + final ProgressIndex progressIndex = resource.getProgressIndex(); if (Objects.isNull(progressIndex)) { continue; } - // A resource can report progress only if that progress will not cover any later resource. - // Otherwise a restart may skip untransferred data. + final List remainingMinimalProgressIndexes = + timePartitionId2RemainingMinimalProgressIndexes.computeIfAbsent( + ((TsFileResource) resource).getTimePartition(), ignored -> new ArrayList<>()); + // A query-priority report is persisted as a time-partition-scoped progress index. Recovery + // only uses it to cover TsFiles from the same partition, so it does not rely on any global + // ordering guarantee between partitions. A numerically larger progress in partition A cannot + // skip an untransferred resource in partition B. if (remainingMinimalProgressIndexes.stream().noneMatch(progressIndex::isEqualOrAfter)) { historicalProgressReportResources.add(resource); } @@ -864,13 +906,11 @@ private boolean shouldExtractTsFileResource( } private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) { - if (startIndex instanceof TimeWindowStateProgressIndex) { + final ProgressIndex innerStartIndex = getInnerProgressIndex(startIndex); + if (innerStartIndex instanceof TimeWindowStateProgressIndex) { // The resource is closed thus the TsFileResource#getFileEndTime() is safe to use - return ((TimeWindowStateProgressIndex) startIndex).getMinTime() <= resource.getFileEndTime(); - } - - if (startIndex instanceof StateProgressIndex) { - startIndex = ((StateProgressIndex) startIndex).getInnerProgressIndex(); + return ((TimeWindowStateProgressIndex) innerStartIndex).getMinTime() + <= resource.getFileEndTime(); } if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) { @@ -886,18 +926,64 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) { private boolean isProgressIndexNotCoveredByStartIndex( PersistentResource resource, ProgressIndex progressIndex) { - if (!startIndex.isEqualOrAfter(progressIndex)) { - LOGGER.info( - DataNodePipeMessages - .PIPE_RESOURCE_MEETS_MAYTSFILECONTAINUNPROCESSEDDATA_CONDITION_EXTRACT, - pipeName, - dataRegionId, - resource, - startIndex, - progressIndex); - return true; + final ProgressIndex innerStartIndex = getInnerProgressIndex(startIndex); + if (innerStartIndex.isEqualOrAfter(progressIndex) + || isProgressIndexCoveredByTimePartitionProgressIndex( + resource, progressIndex, innerStartIndex)) { + return false; } - return false; + + LOGGER.info( + DataNodePipeMessages.PIPE_RESOURCE_MEETS_MAYTSFILECONTAINUNPROCESSEDDATA_CONDITION_EXTRACT, + pipeName, + dataRegionId, + resource, + innerStartIndex, + progressIndex); + return true; + } + + private ProgressIndex getInnerProgressIndex(final ProgressIndex progressIndex) { + return progressIndex instanceof StateProgressIndex + ? ((StateProgressIndex) progressIndex).getInnerProgressIndex() + : Objects.isNull(progressIndex) ? MinimumProgressIndex.INSTANCE : progressIndex; + } + + private boolean isProgressIndexCoveredByTimePartitionProgressIndex( + final PersistentResource resource, + final ProgressIndex progressIndex, + final ProgressIndex startIndex) { + if (!(resource instanceof TsFileResource)) { + return false; + } + + final TimePartitionProgressIndex timePartitionProgressIndex = + getTimePartitionProgressIndex(startIndex); + // Keep this check strictly partition-local, matching the reporting side. This is what makes + // query-priority historical transfer restart-safe even when different partitions are sent in an + // order that conflicts with the global ProgressIndex order. + return Objects.nonNull(timePartitionProgressIndex) + && timePartitionProgressIndex.isProgressIndexEqualOrAfter( + ((TsFileResource) resource).getTimePartition(), progressIndex); + } + + private TimePartitionProgressIndex getTimePartitionProgressIndex( + final ProgressIndex progressIndex) { + final ProgressIndex innerProgressIndex = getInnerProgressIndex(progressIndex); + if (innerProgressIndex instanceof TimePartitionProgressIndex) { + return (TimePartitionProgressIndex) innerProgressIndex; + } + + if (innerProgressIndex instanceof HybridProgressIndex) { + final ProgressIndex timePartitionProgressIndex = + ((HybridProgressIndex) innerProgressIndex) + .getType2Index() + .get(ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX.getType()); + if (timePartitionProgressIndex instanceof TimePartitionProgressIndex) { + return (TimePartitionProgressIndex) timePartitionProgressIndex; + } + } + return null; } private static class HistoricalTsFileExtractionStatistics { @@ -1055,7 +1141,8 @@ public synchronized Event supply() { pendingQueue.poll(); if (shouldUseHistoricalTsFileQueryPriorityOrder()) { if (shouldReportHistoricalProgressAfterResource(tsFileResource)) { - return supplyHistoricalProgressReportEvent(tsFileResource.getMaxProgressIndex()); + return supplyHistoricalProgressReportEvent( + getHistoricalProgressIndexAfterResource(tsFileResource)); } continue; } @@ -1065,7 +1152,8 @@ public synchronized Event supply() { final Event event = supplyTsFileEvent(tsFileResource); pendingQueue.poll(); if (Objects.nonNull(event) && shouldReportHistoricalProgressAfterResource(tsFileResource)) { - pendingHistoricalProgressIndexToReport = tsFileResource.getMaxProgressIndex(); + pendingHistoricalProgressIndexToReport = + getHistoricalProgressIndexAfterResource(tsFileResource); } return event; } @@ -1081,6 +1169,11 @@ private boolean shouldReportHistoricalProgressAfterResource(final PersistentReso && historicalProgressReportResources.remove(resource); } + private ProgressIndex getHistoricalProgressIndexAfterResource(final TsFileResource resource) { + return new TimePartitionProgressIndex( + resource.getTimePartition(), resource.getMaxProgressIndex()); + } + private Event supplyTerminateEvent() { final PipeTerminateEvent.HistoricalTransferSummary historicalTransferSummary = PipeTerminateEvent.snapshotHistoricalTransferSummary(pipeName, creationTime, dataRegionId); 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 3a69e6d0d96f5..9d5c895ce83ee 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 @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; @@ -422,6 +423,96 @@ public void testQueryPriorityOrderPreparesIncrementalSafeProgressReports() throw } } + @Test + @SuppressWarnings("unchecked") + public void testQueryPriorityOrderPreparesSafeProgressReportsByTimePartition() throws Exception { + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final File tempDir = + Files.createTempDirectory("pipeHistoricalTsFilePartitionProgress").toFile(); + + try { + final TsFileResource partition0Progress100 = + createTsFileResource(tempDir, 0L, "100-1-0-0.tsfile"); + partition0Progress100.updateProgressIndex(new SimpleProgressIndex(0, 100)); + final TsFileResource partition1Progress20 = + createTsFileResource(tempDir, 1L, "20-1-0-0.tsfile"); + partition1Progress20.updateProgressIndex(new SimpleProgressIndex(0, 20)); + final List resources = + new ArrayList<>(Arrays.asList(partition0Progress100, partition1Progress20)); + prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder(source, resources); + + Assert.assertEquals( + new HashSet<>(Arrays.asList(partition0Progress100, partition1Progress20)), + getPrivateField(source, "historicalProgressReportResources")); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testQueryPriorityOrderSuppliesPartitionProgressForOutOfGlobalOrderTsFiles() + throws Exception { + final TestablePipeHistoricalDataRegionTsFileAndDeletionSource source = + new TestablePipeHistoricalDataRegionTsFileAndDeletionSource(); + final Event expectedEvent = new Event() {}; + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileCrossPartition").toFile(); + + try { + final TsFileResource partition0Progress100 = + createTsFileResource(tempDir, 0L, "100-1-0-0.tsfile"); + partition0Progress100.updateProgressIndex(new SimpleProgressIndex(0, 100)); + final TsFileResource partition1Progress20 = + createTsFileResource(tempDir, 1L, "20-1-0-0.tsfile"); + partition1Progress20.updateProgressIndex(new SimpleProgressIndex(0, 20)); + + source.setSuppliedEvent(expectedEvent); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "pipeName", "pipe"); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "pipeTaskMeta", new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldExtractDeletion", false); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque( + Arrays.asList(partition0Progress100, partition1Progress20))); + ((Set) getPrivateField(source, "historicalProgressReportResources")) + .add(partition0Progress100); + ((Set) getPrivateField(source, "historicalProgressReportResources")) + .add(partition1Progress20); + + Assert.assertSame(expectedEvent, source.supply()); + final Event partition0ProgressEvent = source.supply(); + Assert.assertTrue(partition0ProgressEvent instanceof ProgressReportEvent); + final ProgressIndex partition0ProgressIndex = + ((ProgressReportEvent) partition0ProgressEvent).getProgressIndex(); + Assert.assertEquals( + new TimePartitionProgressIndex( + partition0Progress100.getTimePartition(), + partition0Progress100.getMaxProgressIndex()), + partition0ProgressIndex); + Assert.assertFalse( + ((TimePartitionProgressIndex) partition0ProgressIndex) + .isProgressIndexEqualOrAfter( + partition1Progress20.getTimePartition(), + partition1Progress20.getMaxProgressIndex())); + + Assert.assertSame(expectedEvent, source.supply()); + final Event partition1ProgressEvent = source.supply(); + Assert.assertTrue(partition1ProgressEvent instanceof ProgressReportEvent); + Assert.assertEquals( + new TimePartitionProgressIndex( + partition1Progress20.getTimePartition(), partition1Progress20.getMaxProgressIndex()), + ((ProgressReportEvent) partition1ProgressEvent).getProgressIndex()); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test @SuppressWarnings("unchecked") public void testQueryPriorityOrderSuppliesProgressAfterSafeTsFileEvent() throws Exception { @@ -455,7 +546,8 @@ public void testQueryPriorityOrderSuppliesProgressAfterSafeTsFileEvent() throws final Event progressEvent = source.supply(); Assert.assertTrue(progressEvent instanceof ProgressReportEvent); Assert.assertEquals( - firstResource.getMaxProgressIndex(), + new TimePartitionProgressIndex( + firstResource.getTimePartition(), firstResource.getMaxProgressIndex()), ((ProgressReportEvent) progressEvent).getProgressIndex()); Assert.assertSame(expectedEvent, source.supply()); } finally { @@ -534,6 +626,44 @@ public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws } } + @Test + public void testMayTsFileContainUnprocessedDataUsesTimePartitionProgressCoverage() + throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalPartitionCoverage").toFile(); + + try { + final ProgressIndex startIndex = + new TimePartitionProgressIndex(0L, new SimpleProgressIndex(0, 100)); + assertMayTsFileContainUnprocessedData( + tempDir, + 0L, + "partition-covered.tsfile", + startIndex, + new SimpleProgressIndex(0, 50), + false); + assertMayTsFileContainUnprocessedData( + tempDir, + 1L, + "partition-uncovered.tsfile", + startIndex, + new SimpleProgressIndex(0, 50), + true); + + final ProgressIndex hybridStartIndex = + hybridProgressIndex( + startIndex, new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 1))); + assertMayTsFileContainUnprocessedData( + tempDir, + 0L, + "hybrid-partition-covered.tsfile", + hybridStartIndex, + new SimpleProgressIndex(0, 80), + false); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + private static TsFileResource createTsFileResource(final File tempDir, final String fileName) throws IOException { final File file = new File(tempDir, fileName); @@ -541,6 +671,14 @@ private static TsFileResource createTsFileResource(final File tempDir, final Str return new TsFileResource(file); } + private static TsFileResource createTsFileResource( + final File tempDir, final long timePartitionId, final String fileName) throws IOException { + final File regionDir = new File(tempDir, "1"); + final File partitionDir = new File(regionDir, String.valueOf(timePartitionId)); + Assert.assertTrue(partitionDir.exists() || partitionDir.mkdirs()); + return createTsFileResource(partitionDir, fileName); + } + private static TsFileResource createClosedTsFileResource( final File tempDir, final String fileName, final ProgressIndex progressIndex) throws IOException { @@ -550,6 +688,18 @@ private static TsFileResource createClosedTsFileResource( return resource; } + private static TsFileResource createClosedTsFileResource( + final File tempDir, + final long timePartitionId, + final String fileName, + final ProgressIndex progressIndex) + throws IOException { + final TsFileResource resource = createTsFileResource(tempDir, timePartitionId, fileName); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + resource.updateProgressIndex(progressIndex); + return resource; + } + private static void assertMayTsFileContainUnprocessedData( final File tempDir, final String fileName, @@ -557,8 +707,27 @@ private static void assertMayTsFileContainUnprocessedData( final ProgressIndex resourceProgressIndex, final boolean expected) throws Exception { - Assert.assertEquals(!expected, startIndex.isEqualOrAfter(resourceProgressIndex)); + assertMayTsFileContainUnprocessedData( + startIndex, createClosedTsFileResource(tempDir, fileName, resourceProgressIndex), expected); + } + private static void assertMayTsFileContainUnprocessedData( + final File tempDir, + final long timePartitionId, + final String fileName, + final ProgressIndex startIndex, + final ProgressIndex resourceProgressIndex, + final boolean expected) + throws Exception { + assertMayTsFileContainUnprocessedData( + startIndex, + createClosedTsFileResource(tempDir, timePartitionId, fileName, resourceProgressIndex), + expected); + } + + private static void assertMayTsFileContainUnprocessedData( + final ProgressIndex startIndex, final TsFileResource resource, final boolean expected) + throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = new PipeHistoricalDataRegionTsFileAndDeletionSource(); setPrivateField(source, "pipeName", "pipe"); @@ -569,10 +738,7 @@ private static void assertMayTsFileContainUnprocessedData( PipeHistoricalDataRegionTsFileAndDeletionSource.class.getDeclaredMethod( "mayTsFileContainUnprocessedData", TsFileResource.class); method.setAccessible(true); - Assert.assertEquals( - expected, - method.invoke( - source, createClosedTsFileResource(tempDir, fileName, resourceProgressIndex))); + Assert.assertEquals(expected, method.invoke(source, resource)); } private static ProgressIndex hybridProgressIndex( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java index f9bf991afee56..774f656099931 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; 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.generator.TsFileNameGenerator; @@ -248,6 +249,47 @@ public void testHybridProgressIndex() { hybridProgressIndex.isAfter(new RecoverProgressIndex(1, new SimpleProgressIndex(2, 21)))); } + @Test + public void testTimePartitionProgressIndex() { + final TimePartitionProgressIndex partition0Progress100 = + new TimePartitionProgressIndex(0L, new SimpleProgressIndex(0, 100)); + + Assert.assertTrue( + partition0Progress100.isProgressIndexEqualOrAfter(0L, new SimpleProgressIndex(0, 50))); + Assert.assertFalse( + partition0Progress100.isProgressIndexEqualOrAfter(1L, new SimpleProgressIndex(0, 50))); + + final TimePartitionProgressIndex partition0Progress120 = + (TimePartitionProgressIndex) + partition0Progress100.updateToMinimumEqualOrIsAfterProgressIndex( + new TimePartitionProgressIndex(0L, new SimpleProgressIndex(0, 120))); + Assert.assertTrue( + partition0Progress120.isProgressIndexEqualOrAfter(0L, new SimpleProgressIndex(0, 120))); + Assert.assertFalse( + partition0Progress120.isProgressIndexEqualOrAfter(0L, new SimpleProgressIndex(0, 121))); + + final TimePartitionProgressIndex partition0And1Progress = + (TimePartitionProgressIndex) + partition0Progress120.updateToMinimumEqualOrIsAfterProgressIndex( + new TimePartitionProgressIndex(1L, new SimpleProgressIndex(0, 20))); + Assert.assertTrue( + partition0And1Progress.isProgressIndexEqualOrAfter(0L, new SimpleProgressIndex(0, 120))); + Assert.assertTrue( + partition0And1Progress.isProgressIndexEqualOrAfter(1L, new SimpleProgressIndex(0, 20))); + + final ByteBuffer byteBuffer = ByteBuffer.allocate(1024); + partition0And1Progress.serialize(byteBuffer); + byteBuffer.flip(); + Assert.assertEquals(partition0And1Progress, ProgressIndexType.deserializeFrom(byteBuffer)); + + final ProgressIndex hybridProgressIndex = + partition0And1Progress.updateToMinimumEqualOrIsAfterProgressIndex( + new IoTProgressIndex(1, 100L)); + Assert.assertTrue(hybridProgressIndex instanceof HybridProgressIndex); + Assert.assertTrue(hybridProgressIndex.isEqualOrAfter(partition0And1Progress)); + Assert.assertTrue(hybridProgressIndex.isEqualOrAfter(new IoTProgressIndex(1, 100L))); + } + @Test public void testProgressIndexMinimumProgressIndexTopologicalSort() { List progressIndexList = new ArrayList<>(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java index 58548e18c4b61..07f1b85205f12 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndexType.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; import org.apache.tsfile.utils.ReadWriteIOUtils; @@ -43,7 +44,8 @@ public enum ProgressIndexType { HYBRID_PROGRESS_INDEX((short) 5), META_PROGRESS_INDEX((short) 6), TIME_WINDOW_STATE_PROGRESS_INDEX((short) 7), - STATE_PROGRESS_INDEX((short) 8); + STATE_PROGRESS_INDEX((short) 8), + TIME_PARTITION_PROGRESS_INDEX((short) 9); private final short type; @@ -82,6 +84,8 @@ public static ProgressIndex deserializeFrom(ByteBuffer byteBuffer) { return TimeWindowStateProgressIndex.deserializeFrom(byteBuffer); case 8: return StateProgressIndex.deserializeFrom(byteBuffer); + case 9: + return TimePartitionProgressIndex.deserializeFrom(byteBuffer); default: throw new UnsupportedOperationException( String.format("Unsupported progress index type %s.", indexType)); @@ -107,6 +111,8 @@ public static ProgressIndex deserializeFrom(InputStream stream) throws IOExcepti return TimeWindowStateProgressIndex.deserializeFrom(stream); case 8: return StateProgressIndex.deserializeFrom(stream); + case 9: + return TimePartitionProgressIndex.deserializeFrom(stream); default: throw new UnsupportedOperationException( String.format("Unsupported progress index type %s.", indexType)); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimePartitionProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimePartitionProgressIndex.java new file mode 100644 index 0000000000000..fa1bb84fff6ec --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimePartitionProgressIndex.java @@ -0,0 +1,292 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.consensus.index.impl; + +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.ProgressIndexType; + +import com.google.common.collect.ImmutableMap; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import javax.annotation.Nonnull; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; + +public class TimePartitionProgressIndex extends ProgressIndex { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(TimePartitionProgressIndex.class) + + RamUsageEstimator.shallowSizeOfInstance(HashMap.class) + + ProgressIndex.LOCK_SIZE; + private static final long ENTRY_SIZE = + RamUsageEstimator.HASHTABLE_RAM_BYTES_PER_ENTRY + + RamUsageEstimator.alignObjectSize(Long.BYTES); + + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + + private final Map timePartitionId2ProgressIndex; + + private TimePartitionProgressIndex() { + this(new HashMap<>()); + } + + public TimePartitionProgressIndex(final Map timePartitionId2ProgressIndex) { + this.timePartitionId2ProgressIndex = new HashMap<>(); + timePartitionId2ProgressIndex.forEach( + (timePartitionId, progressIndex) -> { + if (Objects.nonNull(progressIndex) && !(progressIndex instanceof MinimumProgressIndex)) { + this.timePartitionId2ProgressIndex.put(timePartitionId, progressIndex); + } + }); + } + + public TimePartitionProgressIndex(final long timePartitionId, final ProgressIndex progressIndex) { + this(Collections.singletonMap(timePartitionId, progressIndex)); + } + + public Map getTimePartitionId2ProgressIndex() { + lock.readLock().lock(); + try { + return ImmutableMap.copyOf(timePartitionId2ProgressIndex); + } finally { + lock.readLock().unlock(); + } + } + + public boolean isProgressIndexEqualOrAfter( + final long timePartitionId, final ProgressIndex progressIndex) { + lock.readLock().lock(); + try { + final ProgressIndex timePartitionProgressIndex = + timePartitionId2ProgressIndex.get(timePartitionId); + return Objects.nonNull(timePartitionProgressIndex) + && timePartitionProgressIndex.isEqualOrAfter(progressIndex); + } finally { + lock.readLock().unlock(); + } + } + + @Override + public void serialize(final ByteBuffer byteBuffer) { + lock.readLock().lock(); + try { + ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX.serialize(byteBuffer); + + ReadWriteIOUtils.write(timePartitionId2ProgressIndex.size(), byteBuffer); + for (final Map.Entry entry : timePartitionId2ProgressIndex.entrySet()) { + ReadWriteIOUtils.write(entry.getKey(), byteBuffer); + entry.getValue().serialize(byteBuffer); + } + } finally { + lock.readLock().unlock(); + } + } + + @Override + public void serialize(final OutputStream stream) throws IOException { + lock.readLock().lock(); + try { + ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX.serialize(stream); + + ReadWriteIOUtils.write(timePartitionId2ProgressIndex.size(), stream); + for (final Map.Entry entry : timePartitionId2ProgressIndex.entrySet()) { + ReadWriteIOUtils.write(entry.getKey(), stream); + entry.getValue().serialize(stream); + } + } finally { + lock.readLock().unlock(); + } + } + + @Override + public boolean isAfter(@Nonnull final ProgressIndex progressIndex) { + lock.readLock().lock(); + try { + if (progressIndex instanceof MinimumProgressIndex) { + return !timePartitionId2ProgressIndex.isEmpty(); + } + + if (progressIndex instanceof HybridProgressIndex) { + return ((HybridProgressIndex) progressIndex).isGivenProgressIndexAfterSelf(this); + } + + if (!(progressIndex instanceof TimePartitionProgressIndex)) { + return false; + } + + final TimePartitionProgressIndex thatTimePartitionProgressIndex = + (TimePartitionProgressIndex) progressIndex; + boolean hasStrictlyAfterTimePartition = + timePartitionId2ProgressIndex.size() + > thatTimePartitionProgressIndex.timePartitionId2ProgressIndex.size(); + for (final Map.Entry entry : + thatTimePartitionProgressIndex.timePartitionId2ProgressIndex.entrySet()) { + final ProgressIndex thisProgressIndex = timePartitionId2ProgressIndex.get(entry.getKey()); + if (Objects.isNull(thisProgressIndex) + || !thisProgressIndex.isEqualOrAfter(entry.getValue())) { + return false; + } + if (thisProgressIndex.isAfter(entry.getValue())) { + hasStrictlyAfterTimePartition = true; + } + } + return hasStrictlyAfterTimePartition; + } finally { + lock.readLock().unlock(); + } + } + + @Override + public boolean equals(final ProgressIndex progressIndex) { + lock.readLock().lock(); + try { + if (!(progressIndex instanceof TimePartitionProgressIndex)) { + return false; + } + + return timePartitionId2ProgressIndex.equals( + ((TimePartitionProgressIndex) progressIndex).timePartitionId2ProgressIndex); + } finally { + lock.readLock().unlock(); + } + } + + @Override + public boolean equals(final Object obj) { + if (obj == null) { + return false; + } + if (this == obj) { + return true; + } + if (!(obj instanceof TimePartitionProgressIndex)) { + return false; + } + return this.equals((TimePartitionProgressIndex) obj); + } + + @Override + public int hashCode() { + return Objects.hash(timePartitionId2ProgressIndex); + } + + @Override + public ProgressIndex updateToMinimumEqualOrIsAfterProgressIndex( + final ProgressIndex progressIndex) { + lock.writeLock().lock(); + try { + if (progressIndex == null || progressIndex instanceof MinimumProgressIndex) { + return this; + } + + if (!(progressIndex instanceof TimePartitionProgressIndex)) { + return ProgressIndex.blendProgressIndex(this, progressIndex); + } + + final Map updatedTimePartitionId2ProgressIndex = + new HashMap<>(timePartitionId2ProgressIndex); + ((TimePartitionProgressIndex) progressIndex) + .timePartitionId2ProgressIndex.forEach( + (thatK, thatV) -> + updatedTimePartitionId2ProgressIndex.compute( + thatK, + (thisK, thisV) -> + Objects.isNull(thisV) + ? thatV + : thisV.updateToMinimumEqualOrIsAfterProgressIndex(thatV))); + return new TimePartitionProgressIndex(updatedTimePartitionId2ProgressIndex); + } finally { + lock.writeLock().unlock(); + } + } + + @Override + public ProgressIndexType getType() { + return ProgressIndexType.TIME_PARTITION_PROGRESS_INDEX; + } + + @Override + public TotalOrderSumTuple getTotalOrderSumTuple() { + lock.readLock().lock(); + try { + final ArrayList tupleList = + timePartitionId2ProgressIndex.values().stream() + .map(ProgressIndex::getTotalOrderSumTuple) + .collect(Collectors.toCollection(ArrayList::new)); + tupleList.add(new TotalOrderSumTuple((long) timePartitionId2ProgressIndex.size())); + return ProgressIndex.TotalOrderSumTuple.sum(tupleList); + } finally { + lock.readLock().unlock(); + } + } + + public static TimePartitionProgressIndex deserializeFrom(final ByteBuffer byteBuffer) { + final TimePartitionProgressIndex timePartitionProgressIndex = new TimePartitionProgressIndex(); + + final int size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; ++i) { + final long timePartitionId = ReadWriteIOUtils.readLong(byteBuffer); + final ProgressIndex progressIndex = ProgressIndexType.deserializeFrom(byteBuffer); + timePartitionProgressIndex.timePartitionId2ProgressIndex.put(timePartitionId, progressIndex); + } + return timePartitionProgressIndex; + } + + public static TimePartitionProgressIndex deserializeFrom(final InputStream stream) + throws IOException { + final TimePartitionProgressIndex timePartitionProgressIndex = new TimePartitionProgressIndex(); + + final int size = ReadWriteIOUtils.readInt(stream); + for (int i = 0; i < size; ++i) { + final long timePartitionId = ReadWriteIOUtils.readLong(stream); + final ProgressIndex progressIndex = ProgressIndexType.deserializeFrom(stream); + timePartitionProgressIndex.timePartitionId2ProgressIndex.put(timePartitionId, progressIndex); + } + return timePartitionProgressIndex; + } + + @Override + public String toString() { + return "TimePartitionProgressIndex{" + + "timePartitionId2ProgressIndex=" + + timePartitionId2ProgressIndex + + '}'; + } + + @Override + public long ramBytesUsed() { + return INSTANCE_SIZE + + timePartitionId2ProgressIndex.size() * ENTRY_SIZE + + timePartitionId2ProgressIndex.values().stream() + .map(ProgressIndex::ramBytesUsed) + .reduce(0L, Long::sum); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java index 1e29c96e0905e..1ca10ea709c63 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException; @@ -110,6 +111,16 @@ public void test() throws IOException { new PipeTaskMeta( new TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap), 789)); + put( + 789, + new PipeTaskMeta( + new TimePartitionProgressIndex( + Map.of( + 0L, + new SimpleProgressIndex(0, 1), + 1L, + new SimpleProgressIndex(0, 2))), + 789)); put(Integer.MIN_VALUE, new PipeTaskMeta(new MetaProgressIndex(987), 0)); } });