FrankChen021 commented on code in PR #20589:
URL: https://github.com/apache/druid/pull/20589#discussion_r4336952832
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -343,11 +357,15 @@ private ListenableFuture<NumberedReadableInput>
nextSegmentInput()
/**
* Load the next segment from {@link #loadableSegments} and return a future
to its reference. Returns null
- * if {@link #loadableSegments} is empty.
+ * if {@link #loadableSegments} is empty, or if {@link #canSkipRemaining} is
true.
*/
@Nullable
private ListenableFuture<NumberedReadableInput> loadNextSegment()
{
+ if (canSkipRemaining.getAsBoolean()) {
Review Comment:
[P2] Queue stop races segment acquisition
**Finding:** The canSkipRemaining check occurs before the queue monitor,
while the segment is polled and acquired later. Another processor can reach the
shared limit between those operations, allowing a new partial or full segment
load to start after the queue should have stopped. Such loads remain held until
cleanup, so the hook does not reliably avoid the I/O and resource pressure it
targets.
**Suggestion:** Coordinate input admission with the shared limit using an
atomic claim or cancellation state, rechecking immediately before acquisition
and releasing any load that loses the race.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -485,8 +489,9 @@ private void populateFrameWriterAndFlushIfNeeded() throws
IOException
cursorOffset.increment();
partitionBoostVirtualColumn.setValue(partitionBoostVirtualColumn.getValue() +
1);
- // Flush if we reached cursorPushDownLimit.
- flush = cursorPushDownLimit >= 0 && cursorRowsRead >=
cursorPushDownLimit;
+ // Flush if we reached cursorPushDownLimit or the shared limit.
+ flush = (cursorPushDownLimit >= 0 && cursorRowsRead >=
cursorPushDownLimit)
+ || isSharedLimitReached(frameWriter.getNumRows());
Review Comment:
[P2] Shared limit admission is not atomic
**Finding:** The check reads the shared count, but the count is updated only
after the frame is written. Multiple processors can therefore observe the same
remaining budget, fill their frames, and then add all rows. With concurrent
segment processors and frames larger than the remaining limit, the scan stage
emits and shuffles substantially more than offset plus limit; the downstream
limit stage masks the final row-count error, but the intended performance
optimization is defeated.
**Suggestion:** Use an atomic reservation or serialized shared-limit
admission when accepting or committing rows, and add a multi-processor
regression test that exercises concurrent overshoot.
--
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]
