This is an automated email from the ASF dual-hosted git repository.
gortiz 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 c5c95497516 Account a mailbox send's current block before it reports
its stats (#19365)
c5c95497516 is described below
commit c5c95497516600aa1b21d9a3c6af973cf25910e4
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Wed Aug 26 14:00:30 2026 +0200
Account a mailbox send's current block before it reports its stats (#19365)
---
.../runtime/operator/MailboxSendOperator.java | 4 ++
.../query/runtime/operator/MultiStageOperator.java | 70 ++++++++++++++++++++--
.../runtime/operator/MailboxSendOperatorTest.java | 40 +++++++++++++
.../query/runtime/queries/QueryRunnerTest.java | 40 +++++++++++++
4 files changed, 150 insertions(+), 4 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
index a4aca1dc3ff..3e748deaa97 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
@@ -247,6 +247,10 @@ public class MailboxSendOperator extends
MultiStageOperator {
MultiStageQueryStats stats = null;
List<DataBuffer> serializedStats;
if (_context.isSendStats()) {
+ // The stats are serialized into the block this method is about to send,
so what this operator has spent in
+ // the getNextBlock() call it is running has to be accounted before they
are collected. Otherwise this
+ // operator reports less than the input whose call it contains, and the
stage renders a negative self time.
+ registerExecutionSoFar();
stats = calculateStats();
try {
serializedStats = stats.serialize();
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
index 65aa6da8b2b..479f0fdeed5 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
@@ -61,6 +61,13 @@ public abstract class MultiStageOperator implements
Operator<MseBlock>, AutoClos
_operatorId = Joiner.on("_").join(getClass().getSimpleName(),
_context.getStageId(), _context.getServer());
}
+ /// The [#getNextBlock()] call currently running, or null when none is. See
[#registerExecutionSoFar()].
+ ///
+ /// A single thread runs an opchain at a time and an operator never
re-enters its own [#nextBlock()], so this is
+ /// only ever read and written from inside the call it describes.
+ @Nullable
+ private BlockExecution _blockExecution;
+
/// Returns the logger for the operator.
///
/// This method is used to generic multi-stage operator messages using the
name of the specific operator.
@@ -72,6 +79,57 @@ public abstract class MultiStageOperator implements
Operator<MseBlock>, AutoClos
public abstract void registerExecution(long time, int numRows, long
memoryUsedBytes, long gcTimeMs);
+ /// Accounts everything this operator has spent so far in the
[#getNextBlock()] call currently running.
+ ///
+ /// [#nextBlock()] normally registers a block's usage only once
[#getNextBlock()] has returned, which is too late
+ /// for an operator that has to report its own stats from inside that call:
[MailboxSendOperator] serializes them
+ /// into the end-of-stream block it is about to send. Without this, that
operator reports less time, memory and GC
+ /// than the inputs whose calls it contains, and the stats tree renders a
negative self time for the stage.
+ ///
+ /// Whatever is left when the call returns is registered as usual, so the
totals an operator ends up with are the
+ /// same either way. No rows are attributed here; they are counted from the
block the call returns.
+ ///
+ /// Does nothing when called outside a [#nextBlock()] call.
+ protected void registerExecutionSoFar() {
+ BlockExecution blockExecution = _blockExecution;
+ if (blockExecution != null) {
+ blockExecution.registerUnaccounted(0);
+ }
+ }
+
+ /// What a single [#getNextBlock()] call has spent, and how much of that has
already been handed to
+ /// [#registerExecution].
+ ///
+ /// The meters are created by [#nextBlock()] and passed in rather than
created here, so that each of them keeps
+ /// measuring from exactly the point it always did.
+ private final class BlockExecution {
+ private final Stopwatch _stopwatch = Stopwatch.createStarted();
+ private final ThreadResourceSnapshot _resourceSnapshot;
+ private final long _preGcTimeMs;
+ private long _accountedTimeMs;
+ private long _accountedMemoryBytes;
+ private long _accountedGcTimeMs;
+
+ private BlockExecution(ThreadResourceSnapshot resourceSnapshot, long
preGcTimeMs) {
+ _resourceSnapshot = resourceSnapshot;
+ _preGcTimeMs = preGcTimeMs;
+ }
+
+ /// Hands whatever this call has spent and not yet registered to
[MultiStageOperator#registerExecution],
+ /// attributing `numRows` rows to it. Each invocation registers only what
accrued since the previous one, which
+ /// is what lets the call report from the inside and still end up with
exact totals.
+ private void registerUnaccounted(int numRows) {
+ long timeMs = _stopwatch.elapsed(TimeUnit.MILLISECONDS);
+ long memoryBytes = _resourceSnapshot.getAllocatedBytes();
+ long gcTimeMs = getGcTimeMillis() - _preGcTimeMs;
+ registerExecution(timeMs - _accountedTimeMs, numRows, memoryBytes -
_accountedMemoryBytes,
+ gcTimeMs - _accountedGcTimeMs);
+ _accountedTimeMs = timeMs;
+ _accountedMemoryBytes = memoryBytes;
+ _accountedGcTimeMs = gcTimeMs;
+ }
+ }
+
/// By default, it uses the active deadline, which is the one that should be
used for most operators, but if the
/// operator does not actively process data (ie both mailbox operators), it
should override this method to use the
/// passive deadline instead.
@@ -104,18 +162,22 @@ public abstract class MultiStageOperator implements
Operator<MseBlock>, AutoClos
long preBlockGcTime = getGcTimeMillis();
try (InvocationScope ignored =
Tracing.getTracer().createScope(getClass())) {
MseBlock nextBlock;
- Stopwatch executeStopwatch = Stopwatch.createStarted();
+ BlockExecution blockExecution = new BlockExecution(resourceSnapshot,
preBlockGcTime);
+ _blockExecution = blockExecution;
try {
checkTermination();
nextBlock = getNextBlock();
} catch (Exception e) {
logger().warn("Operator {}: Exception while processing next block",
_operatorId, e);
nextBlock = ErrorMseBlock.fromException(e);
+ } finally {
+ // Cleared even when getNextBlock() throws, so a later
registerExecutionSoFar() cannot read a finished call.
+ _blockExecution = null;
}
int numRows = nextBlock instanceof MseBlock.Data ? ((MseBlock.Data)
nextBlock).getNumRows() : 0;
- long memoryUsedBytes = resourceSnapshot.getAllocatedBytes();
- long gcTimeMs = getGcTimeMillis() - preBlockGcTime;
- registerExecution(executeStopwatch.elapsed(TimeUnit.MILLISECONDS),
numRows, memoryUsedBytes, gcTimeMs);
+ // Only what registerExecutionSoFar() left unaccounted, so the totals
are the same whether or not the operator
+ // reported from inside the call.
+ blockExecution.registerUnaccounted(numRows);
if (logger().isDebugEnabled()) {
logger().debug("Operator {}. Block {} ready to send", _operatorId,
nextBlock);
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
index e0936b6763f..88d07ad2694 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
@@ -171,6 +171,46 @@ public class MailboxSendOperatorTest {
verify(_input).earlyTerminate();
}
+ /// The stats a send operator reports travel inside the end-of-stream block
it sends, so they are collected from
+ /// inside its own getNextBlock() call. Everything that call has spent,
including the input call it contains, must
+ /// already be accounted by then: otherwise this operator reports less time
than its own input, and the stats tree
+ /// renders a negative self time for the stage.
+ @Test
+ public void shouldAccountCurrentBlockBeforeReportingStats()
+ throws Exception {
+ // Given: an input that takes a measurable time and then reports EOS
without ever producing a data block, so the
+ // end-of-stream block is the only block this operator ever handles.
+ when(_input.nextBlock()).thenAnswer(invocation -> {
+ Thread.sleep(50);
+ return SuccessMseBlock.INSTANCE;
+ });
+ long[] reportedAtSendTime = {-1};
+ MailboxSendOperator[] operatorRef = new MailboxSendOperator[1];
+ doAnswer(invocation -> {
+ reportedAtSendTime[0] =
+
operatorRef[0].copyStatMaps().getLong(MailboxSendOperator.StatKey.EXECUTION_TIME_MS);
+ return null;
+ }).when(_exchange).send(any(MseBlock.Eos.class), anyList());
+
+ // When:
+ MailboxSendOperator operator = getOperator();
+ operatorRef[0] = operator;
+ long startNs = System.nanoTime();
+ operator.nextBlock();
+ long wallTimeMs = (System.nanoTime() - startNs) / 1_000_000;
+
+ // Then: the time the input spent is already part of what this operator
reports when it hands its stats over.
+ assertTrue(reportedAtSendTime[0] > 0,
+ "expected the current block to be accounted before the stats are
collected, got " + reportedAtSendTime[0]);
+ long total =
operator.copyStatMaps().getLong(MailboxSendOperator.StatKey.EXECUTION_TIME_MS);
+ assertTrue(total >= reportedAtSendTime[0],
+ "total " + total + " must not be below what was already reported " +
reportedAtSendTime[0]);
+ // What was accounted early must not be counted a second time when the
call returns. An operator can never have
+ // spent more than the call it was made from took, so double counting
shows up as exceeding the wall time.
+ assertTrue(total <= wallTimeMs,
+ "total " + total + " exceeds the " + wallTimeMs + "ms the call
actually took, so it was counted twice");
+ }
+
private MailboxSendOperator getOperator() {
WorkerMetadata workerMetadata = new WorkerMetadata(0, Map.of(), Map.of());
StageMetadata stageMetadata = new StageMetadata(SENDER_STAGE_ID,
List.of(workerMetadata), Map.of());
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
index 00121f59d66..b0d0eb3c404 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.query.runtime.queries;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -29,7 +31,9 @@ import org.apache.pinot.common.response.broker.ResultTable;
import org.apache.pinot.query.QueryEnvironmentTestBase;
import org.apache.pinot.query.QueryServerEnclosure;
import org.apache.pinot.query.mailbox.MailboxService;
+import org.apache.pinot.query.planner.physical.DispatchablePlanFragment;
import org.apache.pinot.query.routing.QueryServerInstance;
+import org.apache.pinot.query.runtime.MultiStageStatsTreeBuilder;
import org.apache.pinot.query.service.dispatch.QueryDispatcher;
import org.apache.pinot.query.testutils.MockInstanceDataManagerFactory;
import org.apache.pinot.query.testutils.QueryTestUtils;
@@ -163,6 +167,42 @@ public class QueryRunnerTest extends QueryRunnerTestBase {
_mailboxService.shutdown();
}
+ /// The self stats of a node are the node's own value minus its children's.
A mailbox send reports its stats from
+ /// inside the getNextBlock() call whose time it is still spending, so
unless that call is accounted first it
+ /// reports less than the input whose call it contains and the subtraction
goes negative. A query whose filter
+ /// matches nothing makes the end-of-stream block the only block a stage
handles, which is when the whole of the
+ /// send's time would be missing.
+ @Test
+ public void testSelfStatsAreNotNegative() {
+ @Language("sql")
+ String sql = "SELECT col1, COUNT(*) FROM a WHERE col1 = 'no-such-value'
GROUP BY col1";
+ QueryDispatcher.QueryResult queryResult = queryRunner(sql, true);
+ Map<Integer, DispatchablePlanFragment> planNodes =
planQuery(sql).getQueryPlan().getQueryStageMap();
+ ObjectNode statsTree =
+ new MultiStageStatsTreeBuilder(planNodes,
queryResult.getQueryStats()).jsonStatsByStage(1);
+
+ int checked = assertSelfStatsAreNotNegative(statsTree);
+ Assert.assertTrue(checked > 0, "expected some self stats to check, got: "
+ statsTree);
+ }
+
+ /// Asserts that no self stat in the tree is negative, and returns how many
were checked.
+ private static int assertSelfStatsAreNotNegative(JsonNode node) {
+ int checked = 0;
+ for (String statName : List.of("selfExecutionTimeMs", "selfClockTimeMs",
"selfAllocatedMB", "selfGcTimeMs")) {
+ JsonNode stat = node.get(statName);
+ if (stat != null) {
+ Assert.assertTrue(stat.asLong() >= 0,
+ statName + " is " + stat.asLong() + ", which means this node
reported less than its children, for node "
+ + node);
+ checked++;
+ }
+ }
+ for (JsonNode child : node.path("children")) {
+ checked += assertSelfStatsAreNotNegative(child);
+ }
+ return checked;
+ }
+
/// Test compares with expected row count only.
@Test(dataProvider = "testDataWithSqlToFinalRowCount")
public void testSqlWithFinalRowCountChecker(String sql, int expectedRows) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
