Caideyipi commented on code in PR #18088:
URL: https://github.com/apache/iotdb/pull/18088#discussion_r3533498001
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java:
##########
@@ -881,27 +987,36 @@ public synchronized Event supply() {
return null;
}
- final PersistentResource resource = pendingQueue.peek();
- if (resource == null) {
- return supplyTerminateEvent();
- }
+ while (true) {
+ final PersistentResource resource = pendingQueue.peek();
+ if (resource == null) {
+ if (shouldReportMaxHistoricalProgressIndex) {
+ shouldReportMaxHistoricalProgressIndex = false;
+ return supplyProgressReportEvent(maxHistoricalProgressIndex);
+ }
+ return supplyTerminateEvent();
+ }
+
+ if (resource instanceof TsFileResource) {
+ final TsFileResource tsFileResource = (TsFileResource) resource;
+ if (consumeSkippedHistoricalTsFileEventIfNecessary(tsFileResource)) {
+ clearReplicateIndexForResource(tsFileResource);
+ pendingQueue.poll();
+ if (shouldUseHistoricalTsFileQueryPriorityOrder()) {
+ continue;
+ }
+ return
supplyProgressReportEvent(tsFileResource.getMaxProgressIndex());
+ }
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
