FrankChen021 commented on code in PR #20805:
URL: https://github.com/apache/druid/pull/20805#discussion_r4435621088
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -442,7 +460,7 @@ protected ReturnOrAwait<Unit> runWithInputChannel(
// Cursor has some more data in it.
populateFrameWriterAndFlushIfNeededWithExceptionHandling();
- if (cursor.isDone()) {
+ if (isCursorDone()) {
Review Comment:
[P2] Exit input-channel scans when the row limit is satisfied
**Finding:** For RowCountScanLimitTracker, STOP flushes the frame and makes
the shared tracker satisfied, but cursorStopped makes isCursorDone() true and
this branch waits for input-channel readability. With an open, empty channel,
the processor will not re-enter runIncrementally to observe the satisfied
tracker until another frame or close arrives, delaying stage termination and
upstream cleanup.
**Suggestion:** After a stopped cursor flushes, terminate when the shared
tracker is satisfied; reserve the await path for per-cursor stops that still
need later input.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -331,47 +352,123 @@ private ListenableFuture<NumberedReadableInput>
nextChannelInput()
@Nullable
private ListenableFuture<NumberedReadableInput> nextSegmentInput()
{
- // Pick a loadahead future, preferring ones that are already loaded.
- ListenableFuture<NumberedReadableInput> selectedLoadaheadFuture = null;
synchronized (this) {
- for (ListenableFuture<NumberedReadableInput> f : pendingNextInputs) {
- if (selectedLoadaheadFuture == null || f.isDone()) {
- selectedLoadaheadFuture = f;
- if (f.isDone()) {
- break;
- }
- }
- }
-
- if (selectedLoadaheadFuture != null) {
- pendingNextInputs.remove(selectedLoadaheadFuture);
+ while (!pendingNextInputs.isEmpty()) {
+ final ListenableFuture<NumberedReadableInput> selectedFuture =
selectPendingInput();
+ final LoadableSegment selectedSegment =
pendingNextInputs.remove(selectedFuture);
Review Comment:
[P2] Release pruned load-ahead futures
**Finding:** When a load-ahead future becomes prunable, this removes it from
pendingNextInputs but leaves ownership in loadedSegments if it completed or
loadingSegments if it is still in flight. Neither path is canceled or drained
here, so repeated pruning can retain segment references until close().
**Suggestion:** Add a discard path that cancels or closes the associated
acquisition and drains any completed SegmentReferenceHolder before continuing.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -331,47 +352,123 @@ private ListenableFuture<NumberedReadableInput>
nextChannelInput()
@Nullable
private ListenableFuture<NumberedReadableInput> nextSegmentInput()
{
- // Pick a loadahead future, preferring ones that are already loaded.
- ListenableFuture<NumberedReadableInput> selectedLoadaheadFuture = null;
synchronized (this) {
- for (ListenableFuture<NumberedReadableInput> f : pendingNextInputs) {
- if (selectedLoadaheadFuture == null || f.isDone()) {
- selectedLoadaheadFuture = f;
- if (f.isDone()) {
- break;
- }
- }
- }
-
- if (selectedLoadaheadFuture != null) {
- pendingNextInputs.remove(selectedLoadaheadFuture);
+ while (!pendingNextInputs.isEmpty()) {
+ final ListenableFuture<NumberedReadableInput> selectedFuture =
selectPendingInput();
+ final LoadableSegment selectedSegment =
pendingNextInputs.remove(selectedFuture);
Review Comment:
Reviewed 20 of 20 changed files at head
`68cf410149cdba39f3cde1544eba742fe86163e8`. The P2 remains valid: removing a
pruned future from `pendingNextInputs` does not cancel its
`AcquireSegmentAction` or drain a completed `SegmentReferenceHolder`, so
references can remain retained until `queue.close()` while other input work
continues. Please add a discard path that releases the selected acquisition
before continuing.
<!-- mergelens:review -->
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
