Jackie-Jiang commented on code in PR #18146:
URL: https://github.com/apache/pinot/pull/18146#discussion_r3717274386
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java:
##########
@@ -69,6 +75,62 @@ protected BaseResultsBlock getNextBlock() {
}
}
+ /// Processes all segments sequentially on the calling thread when only one
task is needed.
+ /// Avoids all concurrency overhead: no ExecutorService submission, no
Phaser, no BlockingQueue, no atomics.
+ /// Respects the query deadline: if the timeout is exceeded before a segment
operator is invoked, a timeout
+ /// results block is returned immediately rather than blocking indefinitely
on a stalled operator.
+ ///
+ /// Note: we do NOT separately accumulate _totalWorkerThreadCpuTimeNs /
_totalWorkerThreadMemAllocatedBytes here.
+ /// This method runs on the calling (main) thread, which
InstanceResponseOperator already measures via its own
+ /// ThreadResourceSnapshot as mainThreadCpuTimeNs. Double-counting the same
thread's CPU time would cause
+ /// calSystemActivitiesCpuTimeNs to produce a negative value (clamped to 0),
breaking resource usage stats.
+ @SuppressWarnings("unchecked")
+ private BaseResultsBlock getNextBlockSingleThread() {
+ T mergedBlock = null;
+ long endTimeMs = _queryContext.getEndTimeMs();
+ for (int i = 0; i < _numOperators; i++) {
+ // Check timeout before invoking each segment operator so we respect the
query deadline
+ // even if a segment operator blocks for a long time (mirrors
mergeResults() timeout logic).
+ if (System.currentTimeMillis() >= endTimeMs) {
Review Comment:
We probably don't need to check timeout here, as it is embedded in
`checkTermination()` for the thread. Please make sure you have a test to cover
this
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java:
##########
@@ -57,8 +57,14 @@ protected
BaseSingleBlockCombineOperator(ResultsBlockMerger<T> resultsBlockMerge
/// @inheritDoc
///
/// Handles exceptions here so that execution stats can be attached.
+ /// When only a single task is needed and the subclass uses the default
ResultsBlockMerger (not null), segments are
+ /// processed directly on the calling thread to avoid the overhead of thread
submission, Phaser synchronization,
+ /// BlockingQueue polling, and atomic operations.
@Override
protected BaseResultsBlock getNextBlock() {
+ if (_numTasks == 1 && _resultsBlockMerger != null) {
+ return getNextBlockSingleThread();
Review Comment:
Even without `_resultsBlockMerger`, we can still benefit from single-thread
merge. I'd suggest only checking `_numTasks == 1`, and make an abstract method
`@Nullable BaseResultsBlock combineSequentially()`. This way we can extend it
for other cases such as group-by and min-max based select order-by operator
--
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]
