perf: skip concurrency overhead in BaseSingleBlockCombineOperator when numTasks=1

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,57 @@ 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.
+  @SuppressWarnings("unchecked")
+  private BaseResultsBlock getNextBlockSingleThread() {
+    ThreadResourceSnapshot resourceSnapshot = new ThreadResourceSnapshot();
+    T mergedBlock = null;
+    long endTimeMs = _queryContext.getEndTimeMs();
+    try {
+      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) {
+          return attachExecutionStats(getTimeoutResultsBlock(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;