gortiz opened a new pull request, #19412:
URL: https://github.com/apache/pinot/pull/19412
## Two operators solve the same problem, and one of them is worse
The multi-stage engine has two ways to produce sorted output.
| | `SortedMailboxReceiveOperator` | `SortOperator` |
|---|---|---|
| Where it can be used | only directly over mailboxes | anywhere in a stage |
| Knows `fetch` / `offset` | **no** | yes |
| Rows held | **the whole input, always** | `fetch + offset`, or the whole
input when there is no limit |
| Requires sorted senders | no | no |
| Exploits sorted senders | no | no |
| Output | one block | one block |
`SortedMailboxReceiveOperator` sits in an awkward middle: it neither
requires its senders to be sorted nor takes any advantage when they are, and it
cannot see the `fetch`/`offset` that `SortOperator` has been given.
## What master builds
`PinotSortExchangeNodeInsertRule` re-parents the `Sort` on top of the
exchange it creates, so **an `ORDER BY` always gets both operators**:
```
SortOperator <- has fetch/offset, but skips sorting:
_priorityQueue = null
SortedMailboxReceiveOperator <- does the sort, knows nothing about the
limit
```
`SortOperator` recognises its input by type and stands down:
```java
if (collations.isEmpty() || input instanceof SortedMailboxReceiveOperator) {
_priorityQueue = null; // degenerate to a limit/offset
trim
```
So the operator that knows the bound does nothing but trim, and the operator
that does the work is blind to the bound. Window functions and `WITHIN GROUP`
aggregates are worse still: `PinotWindowExchangeNodeInsertRule` and
`PinotAggregateExchangeNodeInsertRule` insert `SortedMailboxReceiveOperator`
alone, with no `SortOperator` above it at all. `WindowAggregateOperator` states
the contract it depends on and does no ordering of its own:
```java
/// keys are already ordered based on the 'ORDER BY' keys. No ordering is
performed in this operator. The planner
/// should handle adding a 'SortExchange' to do the ordering prior to
pipelining the data to the upstream operators
/// wherever ordering is required.
```
## How it got this way
The split is not a considered design. #10408 (Mar 2023) moved sorting *into*
the receive operator, adding a
`PriorityQueue`, `_collationKeys` and `_isSortOnReceiver` to what was then a
single `MailboxReceiveOperator`. Its own
commit body says what it was waiting for:
```
- MailboxSendOperator will be modified later to add sort support.
```
That forced one loop to serve two incompatible disciplines — *return the
first block that arrives* for an unsorted
receive, and *drain every mailbox before emitting anything* for a sorted
one. #10570 split the class along exactly
that seam a month later; its message says the non-sorted half restores
"behavior identical to prior to #10408". The
sorted half was born with the TODO it still carries today:
```java
/// TODO: Once sorting on the `MailboxSendOperator` is available, modify
this to use a k-way merge instead of
/// resorting via the PriorityQueue.
```
Sender-side sorting arrived later, through `PinotSortExchangeCopyRule`. The
receive operator never caught up, and
`SortOperator` — which can be placed anywhere and already knows the bound —
was the better home for the work all
along.
## Why the pairing is worse than the alternative
**It fetches everything, then applies the limit.**
`SortedMailboxReceiveOperator` has no `numRowsToKeep`; it accumulates
unconditionally:
```java
private final List<Object[]> _rows = new ArrayList<>();
...
_rows.addAll(((MseBlock.Data) block).asRowHeap().getRows()); // every row,
every mailbox
...
_rows.sort(new SortUtils.SortComparator(_collations, false));
return new RowHeapDataBlock(_rows, _dataSchema); // one block,
the whole result
```
For `SELECT ... ORDER BY k LIMIT 10` across `k` senders that is `k × (limit
+ offset)` rows held to return `limit`. A plain `MailboxReceiveOperator`
feeding `SortOperator`'s bounded heap holds `limit + offset`. The
`earlyTerminate()` `SortOperator` sends afterwards saves nothing — the
buffering has already happened.
**It is slower even with no limit.** Its `ArrayList` + `sort` is fine, but
it emits the whole result as a single block, which is expensive to serialize
and forces the consumer to materialize all of it at once.
**It makes the work built on top of it harder than it needs to be.** #19121
adds a streaming k-way merge so a receive can finally exploit sorted senders —
the right feature, and the measured payoff is real. But because the merge has
to live inside `SortedMailboxReceiveOperator`, it also needs a per-stream
handle abstraction on `BlockingMultiStreamConsumer`, a read-mode latch, a
planner gate and a runtime order check; and it lands in the one operator that
cannot see `fetch`/`offset`, so it has nothing to bound what it buffers. The
same feature implemented over `SortOperator` has the limit in hand and one
fewer abstraction to introduce. The goal here is to make that PR smaller, not
to argue against it.
## This PR
Deprecates `SortedMailboxReceiveOperator` and builds `ORDER BY` as
`MailboxReceiveOperator` + `SortOperator`, so that sorting has one home instead
of two — and so that work layered on top of it, #19121 included, has one place
to go rather than the more awkward of the two.
**All three exchange rules stop setting `isSortOnReceiver`**, so nothing in
the planner produces `SortedMailboxReceiveOperator` any more:
| Rule | Before | After |
|---|---|---|
| `PinotSortExchangeNodeInsertRule` | receive sorts; the `Sort` it
re-parents only trims | the `Sort` does the work |
| `PinotWindowExchangeNodeInsertRule` | receive sorts; nothing above it | an
explicit `Sort` over the exchange |
| `PinotAggregateExchangeNodeInsertRule` | receive sorts; nothing above it |
an explicit `Sort` over the exchange |
The exchange stays a `PinotLogicalSortExchange` throughout, so
`PinotSortExchangeCopyRule` still matches and pushes the sender-side top-N down
— emitting a plain exchange would look tidier and would silently cost far more
than this saves.
The `Sort` inserted for a window or ordered aggregate carries no `fetch`, so
`RexExpressionUtils.getValueAsInt(null)` returns `-1`, `_numRowsToKeep` falls
back to `Integer.MAX_VALUE`, and every row is kept — the same semantics as the
unbounded list the receive operator used. An explicit `Integer.MAX_VALUE` fetch
would be wrong here, because `SortOperator` computes `fetch + offset` and would
overflow.
**`PinotWindowExchangeNodeInsertRule.matches()` had to change too**, and
this is a behavioural fix rather than cleanup. Its idempotence guard was
`!isExchange(window.getInput())`. Placing a `Sort` over the exchange makes the
window's input a `Sort`, so the rule stopped recognising its own output and
re-fired forever — `SELECT AVG(col3), AVG(col3) OVER(PARTITION BY col3) FROM a
GROUP BY col3 ORDER BY col3` hung the planner. The guard now also treats `Sort`
over an exchange as already-processed.
`SortOperator` then has to cover everything the receive operator used to, so
it becomes an abstract base with a factory that picks by *what bounds the
result*. Each implementation names itself in the explain plan:
| Implementation | Explain | Chosen when | Peak memory |
|---|---|---|---|
| `LimitSortOperator` | `SORT_LIMIT` | no collation | one input block |
| `TopNSortOperator` | `SORT_TOP_N` | `fetch`, or a finite response limit,
bounds the result | `fetch + offset` |
| `FullSortOperator` | `SORT_FULL` | nothing bounds it | the input |
All three emit their result as blocks of at most 10000 rows instead of one
block holding everything. `LimitSortOperator` additionally streams — input
blocks are forwarded as they arrive with `offset` skipped and at most `fetch`
rows emitted, then the input is early-terminated.
The `instanceof SortedMailboxReceiveOperator` coupling in `SortOperator`
goes away with it.
## Relationship to #19396 and #19121
Two open PRs add a k-way merge so a receive can exploit sorted senders:
#19396 (window exchanges, sender-side sorting plus a `sortedOnSender`
confirmation through the mailbox protocol) and #19121 (the
`streamingSortedMailboxReceive` option). They were written independently and
neither references the other. Both build on `SortedMailboxReceiveOperator`.
**Nothing here argues against the merge.** It is the right feature and both
PRs measure a real payoff. The argument is only about where it should live, and
the proposal is one of ordering:
1. **This PR first.** Collapse sorting to one implementation, in the
operator that knows the collation *and* the `fetch`/`offset`. Today the same
job is done in two places and neither is the one that can bound it; adding a
merge on top of that is building on the wrong base.
2. **Then decide** whether to take #19396, #19121, or both. The merge itself
belongs in a *receive* operator — it needs per-mailbox access that no operator
above the receive has — so the follow-up is a k-way receiver, whatever it ends
up being called. What changes is where the sender's order comes from: an
explicit `Sort` in the sending opchain rather than sorting inside
`MailboxSendOperator`, and no sort at all when the sending stage's output is
ordered by construction, as a sorted merge join's would be.
The one thing that must not serve as the proof is the table's sorted-column
property. It guarantees only that each segment is sorted internally, and says
nothing about the order of a stage's output stream.
One related observation for #19396: it sorts inside `MailboxSendOperator`.
Pinot already sorts in the sender fragment — `PinotSortExchangeCopyRule` has
been emitting a `LogicalSort` below the exchange for `ORDER BY` for years,
executed by `SortOperator`:
```
LogicalSort(sort0=[$0], dir0=[ASC], offset=[0], fetch=[10]) <- receiver
fragment
PinotLogicalSortExchange(...)
LogicalSort(sort0=[$0], dir0=[ASC], fetch=[10]) <- sender
fragment
PinotLogicalTableScan(table=[[default, a]])
```
Doing it in the send operator instead adds a third place that sorts, one
that cannot see a limit and that the planner cannot reason about — #19396's own
comment on `PinotSortExchangeNodeInsertRule` notes the sender would then be
"sorting rows that are sorted already". Keeping `MailboxSendOperator` agnostic
about ordering and putting a `Sort` in the sender opchain avoids that.
## Why the class is deprecated rather than deleted
No planner path reaches `SortedMailboxReceiveOperator` after this change,
but `MailboxReceiveNode.sort` is a proto field, so a broker running an older
build still sends `sort=true` and the server must still honour it. Deleting the
class needs a release of overlap.
The other direction is already safe: a new broker sends `sort=false`, and an
older server builds a plain receive plus its own `SortOperator`, whose
`instanceof SortedMailboxReceiveOperator` check correctly fails so it does sort.
## Effect
| Query | Before | After |
|---|---|---|
| `ORDER BY k LIMIT n` | `k × (limit + offset)` rows at the receiver |
`limit + offset` |
| `ORDER BY k`, no limit | whole input, one output block | whole input,
streamed in 10000-row blocks, sorted 1.6–4.5× faster |
| window / `WITHIN GROUP` | whole input, one output block | whole input,
streamed in 10000-row blocks, sorted 1.6–4.5× faster |
## Benchmark
`FullSortOperator` replaces an unbounded `PriorityQueue` with a single sort.
Both are `O(n log n)`, so the claim rests on constants and locality.
`BenchmarkMseSortImplementations` measures it with the real
`SortUtils.SortComparator` and `SelectionOperatorUtils.addToPriorityQueue`,
3-column rows, 3 forks × 10 iterations:
| input | rows | sort | heap | speedup |
|---|---|---|---|---|
| random | 10K | 0.695 ± 0.032 ms | 1.085 ± 0.043 ms | 1.6× |
| random | 1M | 302 ± 16 ms | 590 ± 29 ms | 2.0× |
| 64 sorted runs | 10K | 0.229 ± 0.016 ms | 0.907 ± 0.028 ms | 4.0× |
| 64 sorted runs | 1M | 82 ± 4 ms | 369 ± 16 ms | 4.5× |
The sorted-runs rows are the realistic ones: once the sender-side sort is
pushed down, a receive stage sees one sorted run per sender concatenated, and
TimSort detects those runs while a heap cannot.
The trade is allocation: the merge buffer costs 1.2× (1M rows) to 1.7× (10K
rows) what the heap allocates. Faster and hungrier, not strictly better.
## Testing
- `pinot-query-planner`: 1505 tests, 0 failures
- `pinot-query-runtime`: 4566 tests, 0 failures
- `SortOperatorTest` extended to pin implementation selection, streaming,
offset/fetch across block boundaries, block splitting, and the `requireSort`
stat per implementation.
- 200 expected-plan updates: 79 where the exchange's parent was already a
`LogicalSort`, and 121 window / ordered-aggregate plans that now show an
explicit `LogicalSort` over the exchange. The 121 were regenerated from actual
planner output rather than hand-edited.
## Follow-ups
- Recover a k-way merging receive operator, fed by senders whose order comes
from an explicit `Sort` in their opchain, or from a stage that is ordered by
construction — the work #19396 and #19121 are doing.
- Delete `SortedMailboxReceiveOperator` once no supported broker sets
`MailboxReceiveNode.sort`, and drop `isSortOnReceiver` from
`PinotLogicalSortExchange` with it.
- `PinotSortExchangeCopyRule` only pushes the sender-side sort down when
`limit + offset <= sortExchangeCopyThreshold` (10000 by default). Above that,
and with no `LIMIT`, senders ship unsorted and the receiver must full-sort.
--
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]
