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.66% +0.01%
Complexity 318 318
============================================
Files 5296 5296
Lines 371663 371760 +97
Branches 48088 48101 +13
============================================
+ Hits 154819 154908 +89
- Misses 216844 216852 +8 ☔ 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.
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.
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.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,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/datanodemvn spotless:apply -pl iotdb-core/node-commonsmvn "-Ddevelocity.off=true" -o install -pl iotdb-core/node-commons "-DskipTests=true" "-Dcheckstyle.skip=true" "-Dspotless.check.skip=true"git diff --checkAttempted targeted datanode UT execution with:
mvn -U "-Ddevelocity.off=true" test -pl iotdb-core/datanode -Dtest=PipeHistoricalDataRegionTsFileAndDeletionSourceTest -DfailIfNoTests=falseIt failed before test execution because the configured repositories do not currently contain
org.apache.iotdb:iotdb-thrift:jar:2.0.7-20260702.105822-1.