jt2594838 commented on code in PR #18088:
URL: https://github.com/apache/iotdb/pull/18088#discussion_r3526181047
##########
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:
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.
--
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]
