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 f47544ab64f6..2a3923800a3f 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 @@ -92,6 +92,7 @@ private PipeRawTabletInsertionEvent( this.treeModelDatabaseName = treeModelDatabaseName; this.sourceEvent = sourceEvent; this.needToReport = needToReport; + inheritSourceEventReportSkippingIfNecessary(); // Allocate empty memory block, will be resized later. this.allocatedMemoryBlock = @@ -342,6 +343,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 98c660ea3513..ba01491b16af 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 @@ -93,6 +93,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent protected volatile ProgressIndex overridingProgressIndex; private Set tableNames; + // False when generated tablet events should wait for an external progress report. + private volatile boolean shouldReportGeneratedEventsOnCommit = true; public PipeTsFileInsertionEvent(final TsFileResource resource, final boolean isLoaded) { // The modFile must be copied before the event is assigned to the listening pipes @@ -388,6 +390,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)) { PipeTsFileEpochProgressIndexKeeper.getInstance() @@ -404,18 +423,19 @@ public PipeTsFileInsertionEvent shallowCopySelfAndBindPipeTaskMetaForProgressRep final long startTime, final long endTime) { return new PipeTsFileInsertionEvent( - resource, - tsFile, - isWithMod, - isLoaded, - isGeneratedByHistoricalExtractor, - pipeName, - creationTime, - pipeTaskMeta, - pattern, - startTime, - endTime, - isTsFileSealed); + resource, + tsFile, + isWithMod, + isLoaded, + isGeneratedByHistoricalExtractor, + pipeName, + creationTime, + pipeTaskMeta, + pattern, + startTime, + endTime, + isTsFileSealed) + .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 6df5e10d0a25..4c09a5864768 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 @@ -57,6 +57,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY; 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_START_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_PATTERN_FORMAT_IOTDB_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_FORMAT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE; @@ -76,6 +77,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY; 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_START_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_PATTERN_FORMAT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_MODE_KEY; @@ -138,6 +140,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_QUERY_PRIORITY_KEY, + true, + Boolean.TRUE.toString(), + Boolean.FALSE.toString()) + .validateAttributeValueRange( + SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_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/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index 00fd08dc2b7b..3406c60f4714 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java @@ -21,7 +21,11 @@ import org.apache.iotdb.commons.consensus.DataRegionId; 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.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; @@ -38,6 +42,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.db.utils.DateTimeUtils; @@ -58,6 +63,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -79,6 +85,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_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_MODS_ENABLE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_ENABLE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_START_TIME_KEY; @@ -88,6 +96,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_QUERY_PRIORITY_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_START_TIME_KEY; @@ -113,6 +122,8 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR private boolean sloppyTimeRange; // true to disable time range filter after extraction private boolean sloppyPattern; // true to disable pattern filter after extraction + private boolean shouldOrderHistoricalTsFileByQueryPriority = + EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_DEFAULT_VALUE; private Pair listeningOptionPair; private boolean shouldExtractInsertion; @@ -127,6 +138,11 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR private Queue pendingQueue; private final Set filteredTsFileResources = new HashSet<>(); + private final Set historicalProgressReportResources = new HashSet<>(); + private ProgressIndex maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + private ProgressIndex maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE; + private ProgressIndex pendingHistoricalProgressIndexToReport; + private boolean shouldReportMaxHistoricalProgressIndex = false; @Override public void validate(final PipeParameterValidator validator) { @@ -140,6 +156,13 @@ public void validate(final PipeParameterValidator validator) { throw new PipeParameterNotValidException(e.getMessage()); } + shouldOrderHistoricalTsFileByQueryPriority = + parameters.getBooleanOrDefault( + Arrays.asList( + 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 String extractorHistoryLooseRangeValue = parameters .getStringOrDefault( @@ -254,6 +277,12 @@ public void customize( final PipeParameters parameters, final PipeExtractorRuntimeConfiguration configuration) throws IllegalPathException { shouldExtractInsertion = listeningOptionPair.getLeft(); + shouldOrderHistoricalTsFileByQueryPriority = + parameters.getBooleanOrDefault( + Arrays.asList( + 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); // Do nothing if only extract deletion if (!shouldExtractInsertion) { return; @@ -324,6 +353,11 @@ public synchronized void start() { return; } hasBeenStarted = true; + maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; + maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE; + pendingHistoricalProgressIndexToReport = null; + shouldReportMaxHistoricalProgressIndex = false; + historicalProgressReportResources.clear(); final DataRegion dataRegion = StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId)); @@ -342,7 +376,8 @@ public synchronized void start() { // 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) + || shouldUseHistoricalTsFileQueryPriorityOrder()) { dataRegion.syncCloseAllWorkingTsFileProcessors(); } else { dataRegion.asyncCloseAllWorkingTsFileProcessors(); @@ -396,11 +431,13 @@ public synchronized void start() { } }); - originalResourceList.sort( - (o1, o2) -> - startIndex instanceof TimeWindowStateProgressIndex - ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime()) - : o1.getMaxProgressIndex().topologicalCompareTo(o2.getMaxProgressIndex())); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + prepareResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); + } + sortExtractedResources(originalResourceList); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder(originalResourceList); + } pendingQueue = new ArrayDeque<>(originalResourceList); PipeTerminateEvent.initializeHistoricalTransferSummary( pipeName, creationTime, dataRegionId, filteredTsFileResources.size(), 0); @@ -425,6 +462,109 @@ public synchronized void start() { } } + private boolean shouldUseHistoricalTsFileQueryPriorityOrder() { + return shouldOrderHistoricalTsFileByQueryPriority && shouldExtractInsertion; + } + + private void prepareResourcesForHistoricalTsFileQueryPriorityOrder( + final List resourceList) { + // 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 -> !filteredTsFileResources.contains(resource)); + updateMaxHistoricalProgressIndex(resourceList); + shouldReportMaxHistoricalProgressIndex = !resourceList.isEmpty(); + } + + private void prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder( + final List resourceList) { + historicalProgressReportResources.clear(); + final Map> timePartitionId2RemainingMinimalProgressIndexes = + new HashMap<>(); + for (int i = resourceList.size() - 1; i >= 0; --i) { + final TsFileResource resource = resourceList.get(i); + final ProgressIndex progressIndex = resource.getMaxProgressIndex(); + if (Objects.isNull(progressIndex)) { + continue; + } + + final List remainingMinimalProgressIndexes = + timePartitionId2RemainingMinimalProgressIndexes.computeIfAbsent( + 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. + 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 TsFileResource resource : resourceList) { + final ProgressIndex progressIndex = resource.getMaxProgressIndex(); + if (Objects.nonNull(progressIndex)) { + maxHistoricalProgressIndex = + maxHistoricalProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex); + } + } + } + + private void sortExtractedResources(final List resourceList) { + 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(this::compareTsFileResourcesByQueryPriority); + return; + } + + resourceList.sort( + (o1, o2) -> + getInnerProgressIndex(startIndex) instanceof TimeWindowStateProgressIndex + ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime()) + : o1.getMaxProgressIndex().topologicalCompareTo(o2.getMaxProgressIndex())); + } + + private int compareTsFileResourcesByQueryPriority( + final TsFileResource resource1, final TsFileResource resource2) { + 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; + } + + result = + Long.compare( + resource1.getTsFileID().compactionVersion, resource2.getTsFileID().compactionVersion); + if (result != 0) { + return result; + } + + return resource1.getTsFilePath().compareTo(resource2.getTsFilePath()); + } + private boolean shouldExtractTsFileResource(final TsFileResource resource) { if (!isHistoricalSourceEnabled) { return false; @@ -466,26 +606,67 @@ private boolean shouldExtractTsFileResource(final TsFileResource resource) { } 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(); + return ((TimeWindowStateProgressIndex) innerStartIndex).getMinTime() + <= resource.getFileEndTime(); } - if (startIndex instanceof StateProgressIndex) { - startIndex = ((StateProgressIndex) startIndex).getInnerProgressIndex(); + final ProgressIndex resourceProgressIndex = resource.getMaxProgressIndex(); + if (innerStartIndex.isEqualOrAfter(resourceProgressIndex) + || isProgressIndexCoveredByTimePartitionProgressIndex( + resource, resourceProgressIndex, innerStartIndex)) { + return false; } - if (!startIndex.isEqualOrAfter(resource.getMaxProgressIndex())) { - LOGGER.info( - "Pipe {}@{}: file {} meets mayTsFileContainUnprocessedData condition, extractor progressIndex: {}, resource ProgressIndex: {}", - pipeName, - dataRegionId, - resource.getTsFilePath(), - startIndex, - resource.getMaxProgressIndex()); - return true; + LOGGER.info( + "Pipe {}@{}: file {} meets mayTsFileContainUnprocessedData condition, extractor progressIndex: {}, resource ProgressIndex: {}", + pipeName, + dataRegionId, + resource.getTsFilePath(), + innerStartIndex, + resourceProgressIndex); + 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 TsFileResource resource, + final ProgressIndex progressIndex, + final ProgressIndex startIndex) { + 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( + 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 false; + return null; } private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) { @@ -516,16 +697,6 @@ private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource reso deviceID -> pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID())); } - private boolean isTsFileResourceOverlappedWithTimeRange(final TsFileResource resource) { - return !(resource.getFileEndTime() < historicalDataExtractionStartTime - || historicalDataExtractionEndTime < resource.getFileStartTime()); - } - - private boolean isTsFileResourceCoveredByTimeRange(final TsFileResource resource) { - return historicalDataExtractionStartTime <= resource.getFileStartTime() - && historicalDataExtractionEndTime >= resource.getFileEndTime(); - } - private boolean isTsFileResourceCoveredByPattern(final TsFileResource resource) { final Set deviceSet; try { @@ -544,64 +715,128 @@ private boolean isTsFileResourceCoveredByPattern(final TsFileResource resource) deviceID -> pipePattern.coversDevice(((PlainDeviceID) deviceID).toStringID())); } + private boolean isTsFileResourceOverlappedWithTimeRange(final TsFileResource resource) { + return !(resource.getFileEndTime() < historicalDataExtractionStartTime + || historicalDataExtractionEndTime < resource.getFileStartTime()); + } + + private boolean isTsFileResourceCoveredByTimeRange(final TsFileResource resource) { + return historicalDataExtractionStartTime <= resource.getFileStartTime() + && historicalDataExtractionEndTime >= resource.getFileEndTime(); + } + @Override public synchronized Event supply() { if (!hasBeenStarted && StorageEngine.getInstance().isReadyForNonReadWriteFunctions()) { start(); } + if (Objects.nonNull(pendingHistoricalProgressIndexToReport)) { + final ProgressIndex progressIndex = pendingHistoricalProgressIndexToReport; + pendingHistoricalProgressIndexToReport = null; + return supplyHistoricalProgressReportEvent(progressIndex); + } + if (Objects.isNull(pendingQueue)) { return null; } - final TsFileResource resource = pendingQueue.poll(); + while (true) { + final TsFileResource resource = pendingQueue.peek(); + if (resource == null) { + if (shouldReportMaxHistoricalProgressIndex) { + shouldReportMaxHistoricalProgressIndex = false; + if (!maxSuppliedHistoricalProgressReportIndex.isEqualOrAfter( + maxHistoricalProgressIndex)) { + return supplyHistoricalProgressReportEvent(maxHistoricalProgressIndex); + } + } + return supplyTerminateEvent(); + } - if (resource == null) { - final PipeTerminateEvent.HistoricalTransferSummary historicalTransferSummary = - PipeTerminateEvent.snapshotHistoricalTransferSummary( - pipeName, creationTime, dataRegionId); - if (Objects.nonNull(historicalTransferSummary)) { - LOGGER.info( - "Pipe {}@{}: historical source has supplied all events, emitting terminate event. {}", - pipeName, - dataRegionId, - historicalTransferSummary.toReportMessage()); + pendingQueue.poll(); + if (!filteredTsFileResources.contains(resource)) { + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + if (shouldReportHistoricalProgressAfterResource(resource)) { + return supplyHistoricalProgressReportEvent( + getHistoricalProgressIndexAfterResource(resource)); + } + continue; + } + return supplyProgressReportEvent(resource.getMaxProgressIndex()); } - final PipeTerminateEvent terminateEvent = - new PipeTerminateEvent( - pipeName, - creationTime, - pipeTaskMeta, - dataRegionId, - shouldTerminatePipeOnAllHistoricalEventsConsumed); - if (!terminateEvent.increaseReferenceCount( - PipeHistoricalDataRegionTsFileSource.class.getName())) { - LOGGER.warn( - "Pipe {}@{}: failed to increase reference count for terminate event, will resend it", - pipeName, - dataRegionId); - return null; + final Event event = supplyTsFileEvent(resource); + if (Objects.nonNull(event) && shouldReportHistoricalProgressAfterResource(resource)) { + pendingHistoricalProgressIndexToReport = getHistoricalProgressIndexAfterResource(resource); } - isTerminateSignalSent = true; - return terminateEvent; + return event; } + } - if (!filteredTsFileResources.contains(resource)) { - final ProgressReportEvent progressReportEvent = - new ProgressReportEvent(pipeName, creationTime, pipeTaskMeta); - progressReportEvent.bindProgressIndex(resource.getMaxProgressIndex()); - final boolean isReferenceCountIncreased = - progressReportEvent.increaseReferenceCount( - PipeHistoricalDataRegionTsFileSource.class.getName()); - if (!isReferenceCountIncreased) { - LOGGER.warn( - "The reference count of the event {} cannot be increased, skipping it.", - progressReportEvent); - } - return isReferenceCountIncreased ? progressReportEvent : null; + private boolean shouldReportHistoricalProgressAfterResource(final TsFileResource resource) { + return shouldUseHistoricalTsFileQueryPriorityOrder() + && 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); + if (Objects.nonNull(historicalTransferSummary)) { + LOGGER.info( + "Pipe {}@{}: historical source has supplied all events, emitting terminate event. {}", + pipeName, + dataRegionId, + historicalTransferSummary.toReportMessage()); } + final PipeTerminateEvent terminateEvent = + new PipeTerminateEvent( + pipeName, + creationTime, + pipeTaskMeta, + dataRegionId, + shouldTerminatePipeOnAllHistoricalEventsConsumed); + if (!terminateEvent.increaseReferenceCount( + PipeHistoricalDataRegionTsFileSource.class.getName())) { + LOGGER.warn( + "Pipe {}@{}: failed to increase reference count for terminate event, will resend it", + pipeName, + dataRegionId); + return null; + } + isTerminateSignalSent = true; + return terminateEvent; + } + + private Event supplyProgressReportEvent(final ProgressIndex progressIndex) { + final ProgressReportEvent progressReportEvent = + new ProgressReportEvent(pipeName, creationTime, pipeTaskMeta); + progressReportEvent.bindProgressIndex(progressIndex); + final boolean isReferenceCountIncreased = + progressReportEvent.increaseReferenceCount( + PipeHistoricalDataRegionTsFileSource.class.getName()); + if (!isReferenceCountIncreased) { + LOGGER.warn( + "The reference count of the event {} cannot be increased, skipping it.", + progressReportEvent); + } + return isReferenceCountIncreased ? progressReportEvent : null; + } + + private Event supplyHistoricalProgressReportEvent(final ProgressIndex progressIndex) { + maxSuppliedHistoricalProgressReportIndex = + maxSuppliedHistoricalProgressReportIndex.updateToMinimumEqualOrIsAfterProgressIndex( + progressIndex); + return supplyProgressReportEvent(progressIndex); + } + + private Event supplyTsFileEvent(final TsFileResource resource) { filteredTsFileResources.remove(resource); final PipeTsFileInsertionEvent event = new PipeTsFileInsertionEvent( @@ -616,6 +851,10 @@ public synchronized Event supply() { pipePattern, historicalDataExtractionStartTime, historicalDataExtractionEndTime); + if (shouldUseHistoricalTsFileQueryPriorityOrder()) { + event.skipReportOnCommitAndGeneratedEvents(); + } + if (sloppyPattern || isDbNameCoveredByPattern || isTsFileResourceCoveredByPattern(resource)) { event.skipParsingPattern(); } @@ -682,5 +921,7 @@ public synchronized void close() { pendingQueue.clear(); pendingQueue = null; } + historicalProgressReportResources.clear(); + pendingHistoricalProgressIndexToReport = null; } } 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 b46a86e9944a..53f9a6b3a901 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 @@ -28,6 +28,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; import org.apache.iotdb.db.pipe.event.common.tablet.TabletInsertionDataContainer; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; @@ -42,6 +43,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; @@ -264,6 +266,63 @@ private void createTablet() { PipeTabletUtils.compactBitMaps(bitMapsForInsertTabletNode, 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() { TabletInsertionDataContainer container1 = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java index a00fc2c19dbf..ad43f6623ae6 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java @@ -22,12 +22,20 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.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.datastructure.pattern.PrefixPipePattern; +import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; +import org.apache.iotdb.pipe.api.event.Event; import com.google.common.collect.ImmutableMap; import org.apache.tsfile.file.metadata.PlainDeviceID; @@ -35,9 +43,17 @@ import org.junit.Test; import java.io.File; +import java.io.IOException; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.file.Files; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Set; public class PipeHistoricalDataRegionTsFileSourceTest { @@ -70,6 +86,263 @@ public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws } } + @Test + public void testHistoricalTsFileQueryPriorityOrderDefaultsToTrue() { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + + source.validate(new PipeParameterValidator(new PipeParameters(new HashMap<>()))); + + Assert.assertTrue( + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); + } + + @Test + public void testHistoricalTsFileQueryPriorityOrderMatchesQueryCoverage() throws Exception { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + final File tempDir = Files.createTempDirectory("pipeHistoricalTsFileOrder").toFile(); + + try { + final TsFileResource seqLowerVersionNewerFileTimestamp = + createTsFileResource(tempDir, "300-1-0-0.tsfile"); + seqLowerVersionNewerFileTimestamp.setSeq(true); + final TsFileResource seqSameVersionOlderFileTimestamp = + createTsFileResource(tempDir, "100-2-0-0.tsfile"); + 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, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "startIndex", MinimumProgressIndex.INSTANCE); + + final List resources = + new ArrayList<>( + Arrays.asList( + unseqLowerVersionOldestFileTimestamp, + seqHigherVersionOlderFileTimestamp, + seqSameVersionNewerFileTimestamp, + seqSameVersionOlderFileTimestamp, + seqLowerVersionNewerFileTimestamp)); + sortExtractedResources(source, resources); + + Assert.assertEquals( + Arrays.asList( + seqLowerVersionNewerFileTimestamp, + seqSameVersionOlderFileTimestamp, + seqSameVersionNewerFileTimestamp, + seqHigherVersionOlderFileTimestamp, + unseqLowerVersionOldestFileTimestamp), + resources); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + public void testHistoricalTsFileQueryPriorityOrderCanBeDisabled() throws Exception { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + final PipeParameters parameters = + new PipeParameters( + new HashMap() { + { + put( + PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_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, "startIndex", MinimumProgressIndex.INSTANCE); + + final List resources = + new ArrayList<>(Arrays.asList(laterProgressIndex, earlierProgressIndex)); + sortExtractedResources(source, resources); + + Assert.assertFalse( + (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); + Assert.assertEquals(Arrays.asList(earlierProgressIndex, laterProgressIndex), resources); + } finally { + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @Test + public void testHistoricalTsFileQueryPriorityOrderCanBeDisabledByExtractorKey() { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + 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 + @SuppressWarnings("unchecked") + public void testQueryPriorityOrderProgressOnlyCoversSelectedResources() throws Exception { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + 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)); + + ((Set) getPrivateField(source, "filteredTsFileResources")) + .add(selectedResource); + + final List resources = + new ArrayList<>(Arrays.asList(filteredResource, selectedResource)); + prepareResourcesForHistoricalTsFileQueryPriorityOrder(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 testQueryPriorityOrderPreparesIncrementalSafeProgressReports() throws Exception { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + 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 + public void testQueryPriorityOrderPreparesSafeProgressReportsByTimePartition() throws Exception { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + 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 + public void testQueryPriorityOrderReportsProgressAfterAllHistoricalResources() { + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + 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 + 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); + } + } + @Test public void testTsFileResourceCoveredByPattern() throws Exception { final File tempDir = Files.createTempDirectory("pipeHistoricalPatternCoverage").toFile(); @@ -109,8 +382,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 PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); setPrivateField(source, "pipeName", "pipe"); setPrivateField(source, "dataRegionId", 1); @@ -120,10 +412,22 @@ private static void assertMayTsFileContainUnprocessedData( PipeHistoricalDataRegionTsFileSource.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 TsFileResource createTsFileResource(final File tempDir, final String fileName) + throws IOException { + final File file = new File(tempDir, fileName); + Assert.assertTrue(file.createNewFile()); + 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( @@ -138,6 +442,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 TsFileResource createClosedTsFileResourceWithDevices( final File tempDir, final String fileName, final String... devices) throws Exception { final TsFileResource resource = @@ -159,11 +475,57 @@ private static ProgressIndex hybridProgressIndex( return result; } - private static void setPrivateField( - final PipeHistoricalDataRegionTsFileSource source, final String fieldName, final Object value) + private static void sortExtractedResources( + final PipeHistoricalDataRegionTsFileSource source, final List resources) + throws ReflectiveOperationException { + final Method method = + PipeHistoricalDataRegionTsFileSource.class.getDeclaredMethod( + "sortExtractedResources", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + + private static void prepareResourcesForHistoricalTsFileQueryPriorityOrder( + final PipeHistoricalDataRegionTsFileSource source, final List resources) throws ReflectiveOperationException { - final Field field = PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName); - field.setAccessible(true); - field.set(source, value); + final Method method = + PipeHistoricalDataRegionTsFileSource.class.getDeclaredMethod( + "prepareResourcesForHistoricalTsFileQueryPriorityOrder", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + + private static void prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder( + final PipeHistoricalDataRegionTsFileSource source, final List resources) + throws ReflectiveOperationException { + final Method method = + PipeHistoricalDataRegionTsFileSource.class.getDeclaredMethod( + "prepareProgressReportResourcesForHistoricalTsFileQueryPriorityOrder", List.class); + method.setAccessible(true); + method.invoke(source, resources); + } + + private static Object getPrivateField( + final PipeHistoricalDataRegionTsFileSource source, final String fieldName) { + try { + final Field field = PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName); + field.setAccessible(true); + return field.get(source); + } catch (final ReflectiveOperationException e) { + throw new AssertionError(e); + } + } + + private static void setPrivateField( + final PipeHistoricalDataRegionTsFileSource source, + final String fieldName, + final Object value) { + try { + final Field field = PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(source, value); + } catch (final ReflectiveOperationException e) { + throw new AssertionError(e); + } } } 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 87b25883fb17..b9b80b8413b5 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; @@ -245,6 +246,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 58548e18c4b6..07f1b85205f1 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 000000000000..fa1bb84fff6e --- /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/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 4a7ca7534e84..5f94ee2ee939 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 @@ -89,6 +89,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_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; 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 fa0ac2c1a37e..2c0207340dc3 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 @@ -19,12 +19,14 @@ package org.apache.iotdb.commons.pipe.task; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex; 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 +112,17 @@ public void test() throws IOException { new PipeTaskMeta( new TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap), 789)); + put( + 789, + new PipeTaskMeta( + new TimePartitionProgressIndex( + new HashMap() { + { + put(0L, new SimpleProgressIndex(0, 1)); + put(1L, new SimpleProgressIndex(0, 2)); + } + }), + 789)); put(Integer.MIN_VALUE, new PipeTaskMeta(new MetaProgressIndex(987), 0)); } });