[MSE] Build ORDER BY on SortOperator instead of SortedMailboxReceiveOperator

[gortiz]

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:

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:

/// 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:

/// 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:

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 [MSE] Merge sender-sorted streams for ordered windows #19396, Add streaming k-way merge to SortedMailboxReceiveOperator #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 [MSE] Merge sender-sorted streams for ordered windows #19396 and Add streaming k-way merge to SortedMailboxReceiveOperator #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.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 88.46154% with 15 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.58%. Comparing base (80dbb7d) to head (f5501aa).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...inot/query/runtime/operator/LimitSortOperator.java	71.87%	6 Missing and 3 partials ⚠️
...e/rel/rules/PinotWindowExchangeNodeInsertRule.java	72.72%	0 Missing and 3 partials ⚠️
...el/rules/PinotAggregateExchangeNodeInsertRule.java	0.00%	2 Missing ⚠️
...che/pinot/query/runtime/operator/SortOperator.java	97.56%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19412      +/-   ##
============================================
+ Coverage     67.56%   67.58%   +0.02%     
+ Complexity     1430     1424       -6     
============================================
  Files          3486     3489       +3     
  Lines        224173   224236      +63     
  Branches      35381    35391      +10     
============================================
+ Hits         151462   151553      +91     
+ Misses        60684    60646      -38     
- Partials      12027    12037      +10     

Flag	Coverage Δ
integration	?
integration1	?
integration2	?
java-25	67.58% <88.46%> (+0.02%)	⬆️
lane-a	?
lane-b	?
temurin	67.58% <88.46%> (+0.02%)	⬆️
unittests	67.58% <88.46%> (+0.02%)	⬆️
unittests1	57.70% <88.46%> (+0.03%)	⬆️
unittests2	39.32% <27.69%> (-0.01%)	⬇️

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.