This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 064fd76e3f3 Pipe: order historical TsFiles by query priority (#18088)
(#18165)
064fd76e3f3 is described below
commit 064fd76e3f38d8d65a516d864b35e5c58a27e52a
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 24 17:19:23 2026 +0800
Pipe: order historical TsFiles by query priority (#18088) (#18165)
* Pipe: order historical TsFiles by flush time
* Pipe: order historical TsFiles by query priority
* Pipe: report safe query-order historical progress
* Pipe: report time-partition historical progress
(cherry picked from commit 4c1f833adf1748d6f39bb8d88c3e446f680f1706)
---
.../common/tablet/PipeRawTabletInsertionEvent.java | 13 +
.../common/tsfile/PipeTsFileInsertionEvent.java | 20 ++
.../source/dataregion/IoTDBDataRegionSource.java | 12 +
.../PipeHistoricalDataRegionTsFileSource.java | 383 +++++++++++++++++----
.../pipe/event/PipeTabletInsertionEventTest.java | 59 ++++
.../PipeHistoricalDataRegionTsFileSourceTest.java | 380 +++++++++++++++++++-
.../TsFileResourceProgressIndexTest.java | 42 +++
.../commons/consensus/index/ProgressIndexType.java | 8 +-
.../index/impl/TimePartitionProgressIndex.java | 301 ++++++++++++++++
.../pipe/config/constant/PipeSourceConstant.java | 5 +
.../iotdb/commons/pipe/task/PipeMetaDeSerTest.java | 13 +
11 files changed, 1148 insertions(+), 88 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 f47544ab64f..2a3923800a3 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 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent
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 class PipeRawTabletInsertionEvent extends
EnrichedEvent
});
}
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 ae3708fde58..2692000689d 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
@@ -106,6 +106,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
protected volatile ProgressIndex overridingProgressIndex;
private Set<String> tableNames;
+ // False when generated tablet events should wait for an external progress
report.
+ private volatile boolean shouldReportGeneratedEventsOnCommit = true;
private String tsFileParser;
public PipeTsFileInsertionEvent(final TsFileResource resource, final boolean
isLoaded) {
@@ -432,6 +434,23 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
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()
@@ -469,6 +488,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
startTime,
endTime,
isTsFileSealed);
+
copiedEvent.setShouldReportGeneratedEventsOnCommit(shouldReportGeneratedEventsOnCommit);
copiedEvent.setTsFileParser(tsFileParser);
return copiedEvent;
}
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 5c82af0d459..7dcd9d972f4 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.E
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;
@@ -79,6 +80,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.S
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;
@@ -155,6 +157,16 @@ public class IoTDBDataRegionSource extends IoTDBSource {
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 25ace9387b1..d0dd6549f07 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 @@ package
org.apache.iotdb.db.pipe.source.dataregion.historical;
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.pipe.source.dataregion.DataRegionListeningFilter;
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.ArrayDeque;
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.E
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;
@@ -89,6 +97,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.S
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;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_TSFILE_PARSER_KEY;
@@ -115,6 +124,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<Boolean, Boolean> listeningOptionPair;
private boolean shouldExtractInsertion;
@@ -130,6 +141,11 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
private Queue<TsFileResource> pendingQueue;
private final Set<TsFileResource> filteredTsFileResources = new HashSet<>();
+ private final Set<TsFileResource> 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) {
@@ -143,6 +159,13 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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 class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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;
@@ -326,6 +355,11 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
return;
}
hasBeenStarted = true;
+ maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE;
+ maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE;
+ pendingHistoricalProgressIndexToReport = null;
+ shouldReportMaxHistoricalProgressIndex = false;
+ historicalProgressReportResources.clear();
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new
DataRegionId(dataRegionId));
@@ -344,7 +378,8 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
// 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();
@@ -398,11 +433,13 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
}
});
- originalResourceList.sort(
- (o1, o2) ->
- Objects.nonNull(getTimeWindowStateProgressIndex(startIndex))
- ? 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);
@@ -427,6 +464,109 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
}
}
+ private boolean shouldUseHistoricalTsFileQueryPriorityOrder() {
+ return shouldOrderHistoricalTsFileByQueryPriority &&
shouldExtractInsertion;
+ }
+
+ private void prepareResourcesForHistoricalTsFileQueryPriorityOrder(
+ final List<TsFileResource> 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<TsFileResource> resourceList) {
+ historicalProgressReportResources.clear();
+ final Map<Long, List<ProgressIndex>>
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<ProgressIndex> 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<ProgressIndex> 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<TsFileResource>
resourceList) {
+ for (final TsFileResource resource : resourceList) {
+ final ProgressIndex progressIndex = resource.getMaxProgressIndex();
+ if (Objects.nonNull(progressIndex)) {
+ maxHistoricalProgressIndex =
+
maxHistoricalProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex);
+ }
+ }
+ }
+
+ private void sortExtractedResources(final List<TsFileResource> 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;
@@ -468,28 +608,67 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
}
private boolean mayTsFileContainUnprocessedData(final TsFileResource
resource) {
- final TimeWindowStateProgressIndex timeWindowStateProgressIndex =
- getTimeWindowStateProgressIndex(startIndex);
- if (Objects.nonNull(timeWindowStateProgressIndex)) {
+ final ProgressIndex innerStartIndex = getInnerProgressIndex(startIndex);
+ if (innerStartIndex instanceof TimeWindowStateProgressIndex) {
// The resource is closed thus the TsFileResource#getFileEndTime() is
safe to use
- return timeWindowStateProgressIndex.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 TimeWindowStateProgressIndex getTimeWindowStateProgressIndex(
@@ -527,16 +706,6 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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<IDeviceID> deviceSet;
try {
@@ -555,64 +724,128 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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(
@@ -627,6 +860,10 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
pipePattern,
historicalDataExtractionStartTime,
historicalDataExtractionEndTime);
+ if (shouldUseHistoricalTsFileQueryPriorityOrder()) {
+ event.skipReportOnCommitAndGeneratedEvents();
+ }
+
event.setTsFileParser(tsFileParser);
if (sloppyPattern || isDbNameCoveredByPattern ||
isTsFileResourceCoveredByPattern(resource)) {
event.skipParsingPattern();
@@ -694,5 +931,7 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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 b46a86e9944..53f9a6b3a90 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.PipeInsertNodeTabletInsertio
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.apache.tsfile.write.schema.MeasurementSchema;
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 @@ public class PipeTabletInsertionEventTest {
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 003d8f09c54..61e55dc5cac 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,16 +22,21 @@ package
org.apache.iotdb.db.pipe.source.dataregion.historical;
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.config.constant.SystemConstant;
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;
@@ -39,11 +44,18 @@ import org.junit.Assert;
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.Map;
+import java.util.Set;
public class PipeHistoricalDataRegionTsFileSourceTest {
@@ -111,6 +123,263 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
}
}
+ @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<TsFileResource> 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<String, String>() {
+ {
+ 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<TsFileResource> 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<String, String>() {
+ {
+ 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<TsFileResource>) getPrivateField(source,
"filteredTsFileResources"))
+ .add(selectedResource);
+
+ final List<TsFileResource> 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<TsFileResource> 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<TsFileResource> 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<TsFileResource>());
+ 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();
@@ -150,8 +419,27 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
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);
@@ -161,10 +449,22 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
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(
@@ -179,6 +479,18 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
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 =
@@ -200,19 +512,57 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
return result;
}
- private static void setPrivateField(
- final PipeHistoricalDataRegionTsFileSource source, final String
fieldName, final Object value)
+ private static void sortExtractedResources(
+ final PipeHistoricalDataRegionTsFileSource source, final
List<TsFileResource> resources)
throws ReflectiveOperationException {
- final Field field =
PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName);
- field.setAccessible(true);
- field.set(source, value);
+ final Method method =
+ PipeHistoricalDataRegionTsFileSource.class.getDeclaredMethod(
+ "sortExtractedResources", List.class);
+ method.setAccessible(true);
+ method.invoke(source, resources);
}
- private static Object getPrivateField(
- final PipeHistoricalDataRegionTsFileSource source, final String
fieldName)
+ private static void prepareResourcesForHistoricalTsFileQueryPriorityOrder(
+ final PipeHistoricalDataRegionTsFileSource source, final
List<TsFileResource> resources)
throws ReflectiveOperationException {
- final Field field =
PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName);
- field.setAccessible(true);
- return field.get(source);
+ 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<TsFileResource> 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 a0a61e3bbd3..0ad03dc8888 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.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.db.storageengine.dataregion.tsfile.TsFileResource;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
@@ -254,6 +255,47 @@ public class TsFileResourceProgressIndexTest {
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<ProgressIndex> 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 58548e18c4b..07f1b85205f 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.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.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 enum ProgressIndexType {
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 enum ProgressIndexType {
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 00000000000..05ce7c86718
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimePartitionProgressIndex.java
@@ -0,0 +1,301 @@
+/*
+ * 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.Optional;
+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<Long, ProgressIndex> timePartitionId2ProgressIndex;
+
+ private TimePartitionProgressIndex() {
+ this(new HashMap<>());
+ }
+
+ public TimePartitionProgressIndex(final Map<Long, ProgressIndex>
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<Long, ProgressIndex> 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<Long, ProgressIndex> 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<Long, ProgressIndex> 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<Long, ProgressIndex> 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<Long, ProgressIndex> 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 <T extends ProgressIndex> Optional<T> getProgressIndexByType(
+ final Class<T> progressIndexClass) {
+ return progressIndexClass.isInstance(this)
+ ? Optional.of(progressIndexClass.cast(this))
+ : Optional.empty();
+ }
+
+ @Override
+ public TotalOrderSumTuple getTotalOrderSumTuple() {
+ lock.readLock().lock();
+ try {
+ final ArrayList<TotalOrderSumTuple> 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 fe53e228350..346a274563b 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 1e2ecc1bfd3..0992939a5e4 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;
@@ -118,6 +120,17 @@ public class PipeMetaDeSerTest {
new PipeTaskMeta(
new
TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap),
789));
+ put(
+ 789,
+ new PipeTaskMeta(
+ new TimePartitionProgressIndex(
+ new HashMap<Long, ProgressIndex>() {
+ {
+ put(0L, new SimpleProgressIndex(0, 1));
+ put(1L, new SimpleProgressIndex(0, 2));
+ }
+ }),
+ 789));
put(Integer.MIN_VALUE, new PipeTaskMeta(new
MetaProgressIndex(987), 0));
}
});
