perf: skip concurrency overhead in BaseSingleBlockCombineOperator when numTasks=1

When a query runs with a single execution task (e.g. one segment or
maxExecutionThreads=1), BaseSingleBlockCombineOperator still incurred the
full multi-thread overhead: ExecutorService.submit(), Phaser registration/
deregistration, BlockingQueue.poll() with timeout, AtomicInteger, and
AtomicReference.

This adds a single-thread fast path in getNextBlock(): when _numTasks==1
and _resultsBlockMerger is non-null (i.e. the subclass uses the default
merge strategy), segments are processed sequentially on the calling thread
with none of that synchronization overhead. CPU time and memory are still
tracked via ThreadResourceSnapshot.

Subclasses that override mergeResults() with custom logic (e.g.
SequentialSortedGroupByCombineOperator, which passes null for
_resultsBlockMerger) are unaffected and continue using the standard path.

Fixes #14617

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 2 people

pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java

@@ -29,6 +29,7 @@
 import org.apache.pinot.core.operator.blocks.results.ExceptionResultsBlock;
 import org.apache.pinot.core.operator.combine.merger.ResultsBlockMerger;
 import org.apache.pinot.core.query.request.context.QueryContext;
+import org.apache.pinot.spi.accounting.ThreadResourceSnapshot;
 import org.apache.pinot.spi.exception.QueryErrorCode;
 import org.apache.pinot.spi.exception.QueryErrorMessage;
 import org.apache.pinot.spi.exception.QueryException;
@@ -57,8 +58,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();
+    }
     try {
       startProcess();
       return checkTerminateExceptionAndAttachExecutionStats(mergeResults());
@@ -69,6 +76,49 @@ 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.
+  @SuppressWarnings("unchecked")
+  private BaseResultsBlock getNextBlockSingleThread() {
+    ThreadResourceSnapshot resourceSnapshot = new ThreadResourceSnapshot();
+    T mergedBlock = null;
+    try {
+      for (int i = 0; i < _numOperators; i++) {
+        Operator operator = _operators.get(i);
+        T resultsBlock;
+        try {
+          if (operator instanceof AcquireReleaseColumnsSegmentOperator) {
+            ((AcquireReleaseColumnsSegmentOperator) operator).acquire();
+          }
+          resultsBlock = (T) operator.nextBlock();
+        } catch (RuntimeException e) {
+          return createExceptionResultsBlockAndAttachExecutionStats(wrapOperatorException(operator, e),
+              "processing segments");
+        } finally {
+          if (operator instanceof AcquireReleaseColumnsSegmentOperator) {
+            ((AcquireReleaseColumnsSegmentOperator) operator).release();
+          }
+        }
+        if (resultsBlock.getErrorMessages() != null) {
+          // Propagate segment-level error immediately
+          return attachExecutionStats(resultsBlock);
+        }
+        if (mergedBlock == null) {
+          mergedBlock = resultsBlock;
+        } else {
+          _resultsBlockMerger.mergeResultsBlocks(mergedBlock, resultsBlock);
+        }
+        if (_resultsBlockMerger.isQuerySatisfied(mergedBlock)) {
+          break;
+        }
+      }
+    } finally {
+      _totalWorkerThreadCpuTimeNs.addAndGet(resourceSnapshot.getCpuTimeNs());
+      _totalWorkerThreadMemAllocatedBytes.addAndGet(resourceSnapshot.getAllocatedBytes());
+    }
+    return checkTerminateExceptionAndAttachExecutionStats(mergedBlock);
+  }
+
   @Override
   protected void processSegments() {
     int operatorId;