This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a2a27a1e514 Fix flaky SelectionCombineOperatorTest by waiting for
workers before reading stats (#18946)
a2a27a1e514 is described below
commit a2a27a1e5146fe6687f5be5de038743f2bdbbfbd
Author: Timothy Elgersma <[email protected]>
AuthorDate: Sat Jul 11 21:26:41 2026 -0400
Fix flaky SelectionCombineOperatorTest by waiting for workers before
reading stats (#18946)
* Fix flaky SelectionCombineOperatorTest by waiting for workers before
reading stats
In `BaseSingleBlockCombineOperator.getNextBlock()`,
`attachExecutionStats()` was called before `stopProcess()`, meaning worker
threads could still be mutating operator state (specifically `_numDocsScanned`
in `SelectionOnlyOperator`, a plain non-volatile `int`) when the main thread
iterated over operators to aggregate statistics. This caused
`numEntriesScannedPostFilter` (derived as `_numDocsScanned *
numColumnsProjected`) to diverge from `numDocsScanned` when a worker thread
updated [...]
The fix restructures `getNextBlock()` so that `stopProcess()` — which
already uses a `Phaser` to await all worker threads — runs in the `finally`
block before `attachExecutionStats()` is called. No behavior change on the
normal (non-early-termination) path.
Fix flakey test.
Reproduced the flake by adding `@Test(invocationCount = 5000)` to
`selectionOnly()` and running against the pre-fix code: failed twice with
`expected [30] but found [20]` and `expected [50] but found
[40]`(numEntriesScannedPostFilter ≠ numDocsScanned). The same 5000-iteration
run passes cleanly with the fix applied.
Safe to revert. The change only reorders when `stopProcess()` is called
relative to `attachExecutionStats()` within a single query execution; it has no
effect on query results or external behavior, only on the correctness of
execution statistics under early termination.
* wrap the whole body in another try catch to avoid throwing anything at
all.
---
.../combine/BaseSingleBlockCombineOperator.java | 24 ++++++++++++++++++++--
.../combine/CombineErrorOperatorsTest.java | 21 +++++++++++++++++++
2 files changed, 43 insertions(+), 2 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java
index 9713cc1a8b4..56348995a63 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseSingleBlockCombineOperator.java
@@ -59,14 +59,34 @@ public abstract class BaseSingleBlockCombineOperator<T
extends BaseResultsBlock>
/// Handles exceptions here so that execution stats can be attached.
@Override
protected BaseResultsBlock getNextBlock() {
+ try {
+ return mergeResultsAndAttachExecutionStats();
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while attaching execution stats (query:
{})", _queryContext, e);
+ QueryErrorMessage errMsg =
+ QueryErrorMessage.safeMsg(QueryErrorCode.INTERNAL, "Caught exception
while attaching execution stats");
+ return new ExceptionResultsBlock(errMsg);
+ }
+ }
+
+ /// Merges the results and waits for all worker threads to finish before
attaching execution statistics.
+ private BaseResultsBlock mergeResultsAndAttachExecutionStats() {
+ BaseResultsBlock mergedBlock = null;
+ Exception mergeException = null;
try {
startProcess();
- return checkTerminateExceptionAndAttachExecutionStats(mergeResults());
+ mergedBlock = mergeResults();
} catch (Exception e) {
- return createExceptionResultsBlockAndAttachExecutionStats(e, "merging
results blocks");
+ mergeException = e;
} finally {
+ // Wait for all worker threads to finish before reading execution stats.
This ensures that no worker thread is
+ // still mutating operator state (e.g. _numDocsScanned) when
attachExecutionStats() iterates over operators.
stopProcess();
}
+ if (mergeException != null) {
+ return
createExceptionResultsBlockAndAttachExecutionStats(mergeException, "merging
results blocks");
+ }
+ return checkTerminateExceptionAndAttachExecutionStats(mergedBlock);
}
@Override
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/combine/CombineErrorOperatorsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/combine/CombineErrorOperatorsTest.java
index 5d8eec88c98..4c51829cecd 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/operator/combine/CombineErrorOperatorsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/combine/CombineErrorOperatorsTest.java
@@ -128,6 +128,19 @@ public class CombineErrorOperatorsTest {
assertEquals(errorMsg.getErrCode(), QueryErrorCode.QUERY_EXECUTION);
}
+ @Test
+ public void testCombineExecutionStatisticsException() {
+ SelectionOnlyCombineOperator combineOperator =
+ new SelectionOnlyCombineOperator(List.of(new
ExecutionStatisticsExceptionOperator()), QUERY_CONTEXT,
+ _executorService);
+ BaseResultsBlock resultsBlock = combineOperator.nextBlock();
+ assertTrue(resultsBlock instanceof ExceptionResultsBlock);
+ List<QueryErrorMessage> errorMsgs = resultsBlock.getErrorMessages();
+ assertNotNull(errorMsgs);
+ assertEquals(errorMsgs.size(), 1);
+ assertEquals(errorMsgs.get(0).getErrCode(), QueryErrorCode.INTERNAL);
+ }
+
@Test(dataProvider = "getErrorCodes")
public void testCombineExceptionAndErrorOperator(QueryErrorCode
queryErrorCode) {
List<Operator> operators = new ArrayList<>(NUM_OPERATORS);
@@ -227,6 +240,14 @@ public class CombineErrorOperatorsTest {
}
}
+ /// Operator that successfully produces a block but fails while reporting
execution statistics.
+ private static class ExecutionStatisticsExceptionOperator extends
RegularOperator {
+ @Override
+ public ExecutionStatistics getExecutionStatistics() {
+ throw new RuntimeException("Failed to retrieve execution statistics");
+ }
+ }
+
private static class RegularGroupByOperator extends BaseOperator {
private static final String EXPLAIN_NAME = "REGULAR_GROUP_BY";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
