xiangfu0 opened a new pull request, #19396: URL: https://github.com/apache/pinot/pull/19396
Fixes #19395. `PinotLogicalSortExchange` has carried an `isSortOnSender` flag that nothing honored: `MailboxSendOperator` always streamed its input unchanged, and `SortedMailboxReceiveOperator` always buffered every row of every sender and sorted them all at once. The planner set the flag to `false` everywhere, with TODOs saying sender-side sorting was not implemented. This PR implements it and turns it on for the window exchanges, so a WINDOW with an `ORDER BY` inside `OVER()` no longer makes one node hold the whole exchanged input before emitting its first row. ### Sender side `MailboxSendOperator` now sorts when `MailboxSendNode.isSort()` is set and the node carries a collation, using the same `SortUtils.SortComparator` the receiver used, so direction and null ordering are unchanged. It reads its input in full, sorts it, and sends it in blocks of at most 10k rows rather than as one large block, so the receiving mailbox queue keeps providing backpressure and the receiver can start merging before the whole run has arrived. Every destination receives a subsequence of that order: each exchange (`HashExchange`, `RandomExchange`, `BroadcastExchange`, `SingletonExchange`) routes the rows of a block to its destinations without reordering them, and `BlockSplitter` splits a block sequentially. The schema and aggregation functions of the blocks it builds come from the input blocks rather than from the plan node, so what the receiving end reads is what this operator would have forwarded had it not sorted. An error from the input supersedes the rows read so far, as before, and the operator stops sending as soon as the exchange reports every receiver early terminated. ### Receiver side `SortedMailboxReceiveOperator` now has two paths, chosen by `MailboxReceiveNode.isSortedOnSender()`: - **Senders sorted**: a k-way merge with one row cursor per sender. It emits the smallest head row as soon as every sender that has not finished has one, in blocks of at most 10k rows, so rows flow downstream while the exchange is still running and only the rows the merge is ahead of are held. - **Senders did not sort**: the existing buffer-everything-and-sort, unchanged. This is what still runs for plain `ORDER BY` exchanges. The merge reads whichever mailbox has a block ready rather than blocking on the sender it needs next. Blocking on one sender would hold the other senders' rows in their own mailboxes — less memory here — but it deadlocks: a sender blocked on a full mailbox of some other receiver can be the very sender this operator is waiting for, and with two receivers merging two senders each can end up waiting on the sender the other one is blocking, with neither making progress until the query times out. Reading whatever is ready keeps every mailbox drained, cannot deadlock, and buys incremental output at the cost of buffering how far ahead the fastest senders run — never more than the buffer-everything path it replaces. `BlockingMultiStreamConsumer` gains two accessors for this: `getLastReadStream()`, to tell which sender the block just read belongs to, and `isStreamLive(stream)`, because the round-robin read consumes each stream's EOS without returning it and the merge must know that one sender is over while the others still produce. Cancellation, deadlines, backpressure and error propagation are untouched — the merge reads through the same `readMseBlockBlocking()` the unsorted receive uses. ### Planner `PinotWindowExchangeNodeInsertRule` now creates its two sort exchanges with `isSortOnSender=true`, which is what the stale TODOs in that rule described. Nothing else changes: `PinotSortExchangeNodeInsertRule` (plain `ORDER BY`) and `PinotAggregateExchangeNodeInsertRule` keep sorting on the receiver only, so the plan of every non-window query is byte-identical. Turning it on for `ORDER BY` is a larger decision — `PinotSortExchangeCopyRule` may already have pushed a `Sort` below the exchange, so the sender would be re-sorting sorted rows — and is left to a separate change; the TODO there now says so. For a window group with `PARTITION BY` and `ORDER BY` on different keys, hashing on the partition keys sends every row of a partition to the same receiver, so merging by the order keys leaves each partition ordered — the invariant `WindowAggregateOperator` documents it depends on. ### Rolling-upgrade boundary A server that has not been upgraded ignores `MailboxSendNode.isSort()` and sends its rows unsorted, while an upgraded receiver told `sortedOnSender=true` merges them as if they were sorted. **Upgrade servers before brokers.** Until every server runs this code, a broker running it can produce a window plan whose sender does not sort and whose receiver assumes it did, which returns wrongly ordered window results for the duration of a broker-first rollout. There is no persisted-data or plan-format change: `sort` and `sortedOnSender` are existing protobuf fields, only their value changes. ### Testing - `MailboxSendOperatorTest`: the sender sorts by the collation over duplicates and nulls, honours descending with nulls first, splits the sorted run across blocks, drops the buffered rows when the input ends in error, and stops sending when the exchange reports every receiver early terminated. - `SortedMailboxReceiveOperatorTest`: the merge over two senders, with duplicate keys, with nulls, with one sender empty, with every sender empty, plus the error and timeout paths. The existing cases cover the unchanged buffer-and-sort path. - `WindowFunctions.json` gains a `sorted_exchange_window_functions` group, validated against H2 on the two-server runtime harness: `RANK`/`DENSE_RANK` over duplicate order keys, descending order keys, a peer-based running aggregate, `ORDER BY`-only windows where every sender feeds one receiver, and filters that leave most or all senders with nothing to send. - Those new cases were checked to be non-vacuous: with the sender-side sort removed (an old server) but the receiver still merging, 5 of the 8 new queries fail, along with 173 of the pre-existing window cases. - `WindowFunctionPlans.json` expectations updated for the `isSortOnSender=[true]` flag on window exchanges; no other plan expectation moved. Full `pinot-query-planner` (1539 tests) and `pinot-query-runtime` (4599 tests) suites pass. -- 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]
