xiangfu0 commented on code in PR #19867:
URL: https://github.com/apache/pinot/pull/19867#discussion_r4278322771
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java:
##########
@@ -254,6 +266,49 @@ protected void stopProcess() {
_phaser.awaitAdvance(_phaser.arriveAndDeregister());
}
+ /// Waits for the next results block in the queue until the end time.
+ ///
+ /// Returns `null` when the end time is reached without a results block.
Throws the [TerminationException] when the
+ /// query is terminated while waiting.
+ @Nullable
+ protected BaseResultsBlock pollResultsBlock(BlockingQueue<BaseResultsBlock>
blockingQueue, long endTimeMs)
+ throws InterruptedException {
+ while (true) {
+ throwIfTerminated();
+ long waitTimeMs = endTimeMs - System.currentTimeMillis();
+ BaseResultsBlock resultsBlock =
+ blockingQueue.poll(Math.min(waitTimeMs,
TERMINATION_CHECK_INTERVAL_MS), TimeUnit.MILLISECONDS);
+ if (resultsBlock != null || waitTimeMs <= TERMINATION_CHECK_INTERVAL_MS)
{
Review Comment:
Nice and tight. One small behavioral note: the old single
`poll(fullTimeout)` is now sliced into 100 ms waits, so the main thread wakes
up to ~10x more often while genuinely waiting for a slow block. This isn't a
per-record hot path so it's fine, just worth being aware of.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java:
##########
@@ -146,7 +145,7 @@ protected BaseResultsBlock mergeResults()
if (waitTimeMs <= 0) {
return getTimeoutResultsBlock(numBlocksMerged);
}
- BaseResultsBlock blockToMerge = _blockingQueue.poll(waitTimeMs,
TimeUnit.MILLISECONDS);
+ BaseResultsBlock blockToMerge = pollResultsBlock(_blockingQueue,
endTimeMs);
Review Comment:
Minor: the `waitTimeMs <= 0` guard just above is now also covered inside
`pollResultsBlock` (a non-positive slice polls once and returns). Harmless
belt-and-suspenders — you could drop it, but keeping it makes the timeout path
obvious, so either way is fine.
--
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]
