Pipe: order historical TsFiles by query priority#18088
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #18088 +/- ##
============================================
+ Coverage 41.65% 41.89% +0.24%
Complexity 318 318
============================================
Files 5296 5298 +2
Lines 371663 374390 +2727
Branches 48088 48345 +257
============================================
+ Hits 154819 156859 +2040
- Misses 216844 217531 +687 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Caideyipi
changed the title
| while (true) { | ||
| final PersistentResource resource = pendingQueue.peek(); | ||
| if (resource == null) { | ||
| if (shouldReportMaxHistoricalProgressIndex) { | ||
| shouldReportMaxHistoricalProgressIndex = false; | ||
| return supplyProgressReportEvent(maxHistoricalProgressIndex); | ||
| } | ||
| return supplyTerminateEvent(); | ||
| } | ||
|
|
||
| if (resource instanceof TsFileResource) { | ||
| final TsFileResource tsFileResource = (TsFileResource) resource; | ||
| if (consumeSkippedHistoricalTsFileEventIfNecessary(tsFileResource)) { | ||
| clearReplicateIndexForResource(tsFileResource); | ||
| pendingQueue.poll(); | ||
| if (shouldUseHistoricalTsFileQueryPriorityOrder()) { | ||
| continue; | ||
| } | ||
| return supplyProgressReportEvent(tsFileResource.getMaxProgressIndex()); | ||
| } |
There was a problem hiding this comment.
The historical task will only report progress once, which means if the pipe is restarted, all files will be retransmitted?
The cost is just too high. Maybe we can find some resources that can still be reported after processing.
For example, if we use a simple integer to represent the progressIndex of each TsFile, and we have 5 TsFiles ordered by the query order:
[1, 2, 4, 3, 5]
Then we can report progress on TsFile 1, 2, 3, 5, because for these files, we know that there is no file that is behind them and has a smaller index than them.
There was a problem hiding this comment.
Done, thanks for the suggestion.
I changed the query-priority path to precompute which TsFiles can safely report their own progress: a TsFile reports progress only when that progress index will not cover any later pending resource. For the [1, 2, 4, 3, 5] case, this reports 1, 2, 3, and 5 incrementally, while still keeping the final max-progress report as a fallback after all historical TsFiles are supplied.
Added unit coverage for the incremental-safe-progress selection and for supplying the delayed ProgressReportEvent after a safe TsFile event.
|
2 participants
Description
This PR adds a historical Pipe source option:
extractor.history.tsfile.order-by-query-prioritysource.history.tsfile.order-by-query-prioritytrueWhen enabled for historical TsFile-only extraction, selected TsFiles are transferred from lower query/compaction priority to higher priority. This matches the real point coverage semantics used by query and compaction: unsequence files cover sequence files, and within the same seq/unseq category, higher file version wins before higher file-name timestamp.
The receiver loads incoming TsFiles into the unsequence list with monotonically increasing versions, so covered files must be loaded first and covering files later. This preserves duplicated-timestamp overwrite semantics after transfer.
This PR also makes reordered historical progress reporting restart-safe by introducing a time-partition-scoped progress index. Instead of reporting a globally larger
ProgressIndextoo early, the source can reportTimePartitionProgressIndex(timePartitionId, progressIndex)after a TsFile only when that progress cannot cover any remaining resource in the same time partition. Recovery uses this partition progress only for resources from the sametimePartitionId, so an out-of-global-order progress in one partition cannot skip an untransferred TsFile in another partition. After all selected reordered historical TsFiles are supplied, the source still emits the global max historical progress to simplify future recovery.Semantic Changes
For historical TsFile extraction where insertions are captured and deletions are not captured:
MergeReaderPriorityin ascending order, then by compaction version and file path as deterministic tie-breakers.skipReportOnCommitAndGeneratedEvents(), so neither the TsFile event nor generated tablet events report the original per-file progress directly.TsFileResource.getTimePartition().TimePartitionProgressIndex, and restart filtering first checks normal global progress coverage, then checks partition-local coverage.ProgressReportEventwith the max progressIndex of the selected resources.When historical deletions are captured together with insertions, the source keeps the previous progressIndex ordering. Deletion resources only carry progressIndex ordering information, so this avoids changing insertion/deletion ordering semantics.
The option can be set to
falseto keep the previous progressIndex-based ordering.Tests / Coverage
Added or updated unit coverage for:
true,source.*andextractor.*disable keys,TimePartitionProgressIndexfor out-of-global-order cross-partition TsFiles,TimePartitionProgressIndexmerge, coverage, hybrid blending, and serialization,I did not add an integration test because ordinary end-to-end SQL write scenarios do not deterministically create the exact file-version/file-timestamp/seq-unseq priority conflicts needed for this regression. The deterministic behavior that matters here is the historical source's resource ordering and progress reporting contract, which is covered by unit tests with explicit TsFileResource setup.
Local verification:
mvn spotless:apply -pl iotdb-core/node-commons,iotdb-core/datanodegit diff --checkmvn -pl iotdb-core/node-commons -Dtest=PipeMetaDeSerTest testmvn -pl iotdb-core/node-commons -DskipTests installmvn -pl iotdb-core/datanode "-Dtest=TsFileResourceProgressIndexTest,PipeHistoricalDataRegionTsFileAndDeletionSourceTest" testmvn -pl iotdb-core/datanode "-Dtest=PipeHistoricalDataRegionTsFileAndDeletionSourceTest" -DforkCount=1 test