FrankChen021 commented on code in PR #20589:
URL: https://github.com/apache/druid/pull/20589#discussion_r4270848743
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java:
##########
@@ -557,19 +562,29 @@ private long setNextCursor(
if (query.isLimited()
&& ordering.size() >= query.getOrderBys().size()
&& query.getOrderBys().equals(ordering.subList(0,
query.getOrderBys().size()))) {
- cursorPushDownLimit = getQueryOffsetPlusLimit();
+ cursorPushDownLimit = queryOffsetPlusLimit;
} else {
cursorPushDownLimit = NO_LIMIT;
}
return rowsFlushed;
}
+ /**
+ * Whether the limit in {@link #sharedRunningCountForLimit} has been
reached, or would be reached if
+ * {@code pendingRows} were added.
+ */
+ private boolean isSharedLimitReached(final long pendingRows)
+ {
+ return sharedRunningCountForLimit != null
+ && sharedRunningCountForLimit.get() + pendingRows >=
queryOffsetPlusLimit;
Review Comment:
[P2] Shared limit admission is not atomic
**Finding:** Each processor reads the shared count here, but the count is
incremented only after its frame is written. Concurrent processors can
therefore all observe remaining capacity, fill and flush rows, and push the
count beyond offset+limit by roughly one buffered frame per processor; the
downstream limit may discard those rows, but the scan work and shuffle overhead
remain.
**Suggestion:** Atomically reserve the remaining row budget or otherwise
coordinate row admission with accounting, and cover multi-processor execution
with a regression test.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -303,11 +316,15 @@ private ListenableFuture<ReadableInput> 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<ReadableInput> loadNextSegment()
{
+ if (canSkipRemaining.getAsBoolean()) {
Review Comment:
[P2] Segment load can race the skip predicate
**Finding:** The `canSkipRemaining` check occurs before entering the
synchronized poll-and-acquire block, while the shared count is updated
concurrently by processors. A thread can observe false, wait for the monitor,
and then start acquiring another segment after another processor reaches
offset+limit, handing work to a processor that immediately exits and wasting
segment I/O and cache resources.
**Suggestion:** Coordinate or recheck the skip decision with the
synchronized poll-and-acquire operation, and add a concurrent transition test.
--
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]
