gortiz opened a new pull request, #19364:
URL: https://github.com/apache/pinot/pull/19364
## Problem
MSE stage stats are aggregated by summing across the workers of a stage, so
they cannot show how unevenly the work was spread. `clockTimeMs` is
`executionTimeMs / parallelism`, which assumes a uniform distribution across
workers.
That assumption fails exactly when it matters most: a stage with
`parallelism: 10` whose rows all landed on a single worker reports a tenth of
the clock time it actually took, and nothing in the output hints that the
average is meaningless.
## What this adds
**`activeWorkers`** on `LEAF`, `MAILBOX_RECEIVE` and `MAILBOX_SEND`. Each
operator reports how many of the stage's workers it was active on, applying its
own notion of activity:
| operator | active on a worker when |
|---|---|
| `LEAF` | it had at least one segment assigned to it |
| `MAILBOX_RECEIVE` | it received at least one row |
| `MAILBOX_SEND` | it sent at least one row |
Deliberately not a single stage-level number. Comparing `activeWorkers`
against `parallelism` on a node detects bias; comparing it *across* the nodes
of a stage shows where the stage narrowed. A leaf active on ten workers under a
send active on one means the work was spread but the output was not — which
reads very differently from ten idle workers, and a single number cannot tell
them apart.
**`maxEmittedRows`** and **`minNonzeroEmittedRows`** on `MAILBOX_SEND`,
giving the spread of rows over the workers that sent any. Workers that sent
nothing are excluded from the minimum, so one idle worker cannot pin it to
zero. Since a send is active exactly on the workers that sent rows,
`minNonzeroEmittedRows * activeWorkers <= emittedRows <= maxEmittedRows *
activeWorkers` holds.
Example, a `GROUP BY` over two servers where the shuffle split 3/2:
```
MAILBOX_SEND parallelism:2 activeWorkers:2 emittedRows:5 max:3
minNonzero:2
AGGREGATE
MAILBOX_RECEIVE parallelism:2 activeWorkers:2 emittedRows:10
MAILBOX_SEND parallelism:2 activeWorkers:2 emittedRows:10 max:5
minNonzero:5
LEAF parallelism:2 activeWorkers:2 emittedRows:10
```
And the same query with a filter matching nothing, where only the leaf is
active — the segments were read, and the stage produced nothing:
```
MAILBOX_SEND parallelism:2
AGGREGATE
MAILBOX_RECEIVE parallelism:2
MAILBOX_SEND parallelism:2
LEAF parallelism:2 activeWorkers:2 numSegmentsQueried:3
numSegmentsPrunedByServer:3
```
## Implementation notes
- The stats are derived in `copyStatMaps()` rather than accumulated in
`registerExecution()`, because they are per-worker totals and not per-block
quantities — merging `MAX_EMITTED_ROWS` once per block would report the largest
block instead of the busiest worker, and `ACTIVE_WORKERS` would count blocks.
- They are computed on the returned copy rather than on the operator's own
stat map, so the several `calculateStats()` calls made per opchain cannot count
the same worker more than once. This is covered by a test.
- `ServerQueryRequest` gains `hasSegmentsToQuery()`, resolving the two
segment representations it holds (flat in `getSegmentsToQuery()` for a plain
table, grouped per referenced table in `getTableSegmentsContexts()` for a
logical table, with the other one `null`) so callers needing only the answer do
not have to know which applies. `getSegmentsToQuery()` is now annotated
`@Nullable`, which it always was in practice.
## Compatibility
The new keys are appended at the end of their enums, since `StatMap`
identifies keys by their ordinal on the wire.
A reader that does not know a key cannot skip it, because the encoding is
not self-delimiting, so in a mixed-version cluster it loses the whole stat map
rather than just the new stat. On the default mailbox path that failure is
contained — `BlockingMultiStreamConsumer.mergeStats` drops the stats and the
query completes normally. `SendStatsPredicate`'s `SAFE` mode avoids it entirely
by not sending stats unless every instance reports the same version. This is a
pre-existing property of `StatMap` shared by every previous key addition;
making the reader skip unknown keys needs a format change and is worth its own
PR.
## Testing
- Unit tests per operator for the positive and negative activity cases,
cross-worker merge, merge through the serialized form, and repeated stat
collection.
- End-to-end tests over the two-server runtime asserting the invariants
above on real merged stats, including a query whose filter matches nothing so
that an active-but-not-emitting stage is exercised.
## Also
Corrects `SendStatsPredicate`'s class javadoc, which still described `SAFE`
as the default mode after the default became `ALWAYS`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
