Fix resource stats double-counting and exception handling in single-thread fast path

Two bugs in getNextBlockSingleThread():

1. Resource stats double-counting:
   The method accumulated _totalWorkerThreadCpuTimeNs on the calling thread,
   but InstanceResponseOperator.getBaseBlock() already captures the same
   thread's CPU time as mainThreadCpuTimeNs. The calSystemActivitiesCpuTimeNs
   formula (wallClock - mainThread - workerThread/N) then subtracted the same
   work twice, producing a negative value clamped to 0. This broke
   testResourceUsageStats in the CPU/memory query killing integration tests.
   Fix: remove the ThreadResourceSnapshot tracking; the main thread's snapshot
   already accounts for all work done in the single-thread path.

2. Exception handling for operator failures:
   wrapOperatorException() throws (not returns) a new QueryException for most
   RuntimeExceptions. Calling it as an argument to createExceptionResultsBlock-
   AndAttachExecutionStats caused the thrown exception to escape the catch block
   uncaught, propagating to callers instead of being wrapped in an error block.
   This broke JsonExtractScalarTransformFunctionTest.mvWithNullsWithoutDefault.
   Fix: wrap in try/throw/catch to capture whatever wrapOperatorException
   throws or returns, then convert to an error block — mirroring the multi-
   thread path's onProcessSegmentsException handler.

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,7 +29,6 @@
 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;
@@ -80,49 +79,54 @@ protected BaseResultsBlock getNextBlock() {
   /// 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() {
-    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));
+    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();
         }
-        Operator operator = _operators.get(i);
-        T resultsBlock;
+        resultsBlock = (T) operator.nextBlock();
+      } catch (RuntimeException e) {
+        // wrapOperatorException either returns the exception (for EarlyTerminationException) or throws a new
+        // QueryException. Either way we catch it here and convert to an error block, mirroring the multi-thread path
+        // where onProcessSegmentsException handles all operator exceptions.
         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);
+          throw wrapOperatorException(operator, e);
+        } catch (Exception wrapped) {
+          return createExceptionResultsBlockAndAttachExecutionStats(wrapped, "processing segments");
         }
-        if (_resultsBlockMerger.isQuerySatisfied(mergedBlock)) {
-          break;
+      } finally {
+        if (operator instanceof AcquireReleaseColumnsSegmentOperator) {
+          ((AcquireReleaseColumnsSegmentOperator) operator).release();
         }
       }
-    } finally {
-      _totalWorkerThreadCpuTimeNs.addAndGet(resourceSnapshot.getCpuTimeNs());
-      _totalWorkerThreadMemAllocatedBytes.addAndGet(resourceSnapshot.getAllocatedBytes());
+      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;
+      }
     }
     return checkTerminateExceptionAndAttachExecutionStats(mergedBlock);
   }