Account a mailbox send's current block before it reports its stats

[gortiz]

Problem

A mailbox send operator serializes its stats into the end-of-stream block it sends, so it collects them from inside the getNextBlock() call it is still running. nextBlock() only registers a block's usage once getNextBlock() has returned:

MultiStageOperator.nextBlock()
  stopwatch.start()
  └─ MailboxSendOperator.getNextBlock()
       ├─ _input.nextBlock()                     ← input runs to completion, registers its time
       └─ sendEos() → calculateStats()            ← stats collected HERE
  registerExecution(stopwatch.elapsed())          ← the send's own time, recorded too late

Since the send's call contains the call it made to its input, it ends up reporting less than its own children, and selfExecutionTimeMs = parent − Σ children goes negative:

MAILBOX_SEND  (no executionTimeMs)   selfExecutionTimeMs: -310   selfClockTimeMs: -155
  AGGREGATE     executionTimeMs: 310
    MAILBOX_RECEIVE  executionTimeMs: 310
      MAILBOX_SEND   (no executionTimeMs)  selfExecutionTimeMs: -24
        LEAF           executionTimeMs: 24

It is most visible when a stage handles no data block at all — the end-of-stream block is then the only block, so the operator reports nothing — but the value is understated by its final block in every case. memoryUsedBytes and gcTimeMs are captured at the same point and are understated the same way, so selfAllocatedMB and selfGcTimeMs can go negative too.

Measured directly: a send whose input takes ~300ms and then reports EOS serializes executionTimeMs: 0 while the call actually took 291ms.

Fix

MultiStageOperator keeps what the running getNextBlock() call has spent in a BlockExecution object, non-null exactly while such a call is running, and registers only what has not been accounted yet when the call returns. registerExecutionSoFar() lets an operator account the call from the inside; MailboxSendOperator calls it just before collecting its stats.

Each registration contributes only what accrued since the previous one, so the totals an operator ends up with are unchanged — OpChainSchedulerService.onOpChainFinished and the CPU_EXECUTION_TIME_MS meter see exactly what they saw before. Only the serialized snapshot moves earlier.

Same query after the fix:

MAILBOX_SEND  executionTimeMs: 334   clockTimeMs: 167
  AGGREGATE     executionTimeMs: 334
    MAILBOX_RECEIVE  executionTimeMs: 334   selfExecutionTimeMs: 334
      MAILBOX_SEND   executionTimeMs: 24    selfExecutionTimeMs: 2
        LEAF           executionTimeMs: 22    selfExecutionTimeMs: 22

The self time is now non-negative by construction: the stats are collected strictly after every input call has returned, so the send's elapsed time necessarily covers its children's.

What this does not fix

The send remains short by the time it spends serializing the stats and handing the block to the exchange, after the snapshot. Counting that would require the stats to leave the operator outside getNextBlock(), which is a protocol change.

Stats reported directly to the broker were already correct and are untouched: QueryRunner.effectiveSendStats() returns false in stream mode, so that path never goes through sendEos() and collects its stats after the opchain loop has finished.

Testing

• MailboxSendOperatorTest.shouldAccountCurrentBlockBeforeReportingStats — an input that takes time and then reports EOS with no data block, so the end-of-stream block is the only one. Asserts the stats handed to the exchange carry a non-zero time, and that the total does not exceed the wall time of the call, which is what a double count would look like.
• QueryRunnerTest.testSelfStatsAreNotNegative — asserts no selfExecutionTimeMs, selfClockTimeMs, selfAllocatedMB or selfGcTimeMs anywhere in the rendered stats tree is negative, over a real two-server query whose filter matches nothing.

Both were verified to fail without the fix (got 0 and selfExecutionTimeMs is -310 respectively).

Relationship to #19364

Independent — this touches only when stats are read, not what they contain. The negative self time was noticed while working on that PR.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 95.45455% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 67.16%. Comparing base (4824a5f) to head (87cdbc4).
⚠️ Report is 4 commits behind head on master.

Files with missing lines	Patch %	Lines
...not/query/runtime/operator/MultiStageOperator.java	95.23%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19365      +/-   ##
============================================
- Coverage     67.20%   67.16%   -0.04%     
- Complexity     1418     1424       +6     
============================================
  Files          3479     3479              
  Lines        223075   223092      +17     
  Branches      35135    35136       +1     
============================================
- Hits         149920   149847      -73     
- Misses        61184    61286     +102     
+ Partials      11971    11959      -12     

Flag	Coverage Δ
integration	100.00% <ø> (+100.00%)	⬆️
integration1	100.00% <ø> (?)
integration2	0.00% <ø> (ø)
java-25	67.16% <95.45%> (-0.04%)	⬇️
lane-a	100.00% <ø> (+100.00%)	⬆️
lane-b	0.00% <ø> (ø)
temurin	67.16% <95.45%> (-0.04%)	⬇️
unittests	67.16% <95.45%> (-0.05%)	⬇️
unittests1	57.70% <95.45%> (+0.01%)	⬆️
unittests2	38.95% <77.27%> (-0.05%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[xiangfu0]

Documentation follow-up: pinot-contrib/pinot-docs#1008 (merged).