[MSE] Add opt-in sender-sorted merging for global ordered windows - #19396

Codecov Report

❌ Patch coverage is 75.21008% with 118 lines in your changes missing coverage. Please review.
✅ Project coverage is 68.59%. Comparing base (f818bae) to head (85be3b0).

@@             Coverage Diff              @@
##             master   #19396      +/-   ##
============================================
+ Coverage     68.55%   68.59%   +0.04%     
  Complexity     1486     1486              
============================================
  Files          3525     3531       +6     
  Lines        229852   230210     +358     
  Branches      36507    36572      +65     
============================================
+ Hits         157568   157907     +339     
+ Misses        59965    59953      -12     
- Partials      12319    12350      +31     

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

We are already working on something like that in #19120, #19121 and #19122. I would recommend coordinating our efforts.

I've been working on my own fix for SortedMailboxReceiverOperator, which goes back to what used to be:

┌───────────────────────────────────┬─────────────────────────┬──────────────────────┐
│                                   │ #19396                  │      My branch       │
├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
│                                   │ sortOnSender =          │ sortOnReceiver =     │
│ PinotWindowExchangeNodeInsertRule │ **true**, keeps         │ **false** + explicit │
│                                   │ sortOnReceiver = true   │  Sort                │
├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
│                                   │ +336 lines — k-way      │                      │
│ SortedMailboxReceiveOperator      │ merge over sorted       │ @Deprecated          │
│                                   │ senders                 │                      │
├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
│ MailboxSendOperator               │ +143 — sorts sender     │ untouched            │
│                                   │ output, 10k blocks      │                      │
├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
│                                   │ new sortedOnSender      │                      │
│ Mailbox protocol                  │ confirmation over gRPC  │ untouched            │
│                                   │ + in-memory             │                      │
├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
│ PinotSortExchangeNodeInsertRule   │ comment only —          │ sortOnReceiver =     │
│                                   │ deliberately deferred   │ false                │
└───────────────────────────────────┴─────────────────────────┴──────────────────────┘

The idea I have is to stop generating SortedMailboxReceiveOperator and substitute that with a sort on the receiver side (with optional sort on the sender side when the limit is small). Once we have that, we can start thinking about recovering SortedMailboxReceiveOperator as a k-way merge when the senders guarantee data is sent in order, which is what you and #19121 are doing (although #19121 keeps both the current and the k-way merge).

Also, I don't think MailboxSendOperator should sort on send. Instead, we should add a sort operator on the sender opchain and keep MailboxSendOperator agnostic about ordering.

@gortiz I rebased this onto current master and refactored it around the coordination feedback.

• Sender sorting is now owned by an explicit SortNode / SortOperator; MailboxSendOperator no longer performs sorting and only advertises a guarantee it can structurally verify.
• Ordered-window fan-in now uses a dedicated SortedMailboxMergeReceiveOperator, leaving the legacy SortedMailboxReceiveOperator execution path intact. That isolates the overlap with Add streaming k-way merge to SortedMailboxReceiveOperator #19121 while Add streaming selection ORDER BY combine for physically sorted segments #19120 and Add streaming sorted merge join with colocation support #19122 remain complementary producer/consumer work.
• This follows the ownership direction in [MSE] Build ORDER BY on SortOperator instead of SortedMailboxReceiveOperator #19412. If [MSE] Build ORDER BY on SortOperator instead of SortedMailboxReceiveOperator #19412 lands first, the remaining overlap should be mechanical rather than two competing sender-sort implementations.
• Mixed-version behavior is safe in either upgrade direction: missing per-block confirmation falls back to the legacy full receiver sort before output, while old receivers retain sortOnReceiver = true.

Validation on head f151331206ca6a13157b1d93e55bb889d4c39c25: the 20-module runtime reactor passed (planner 1,540 tests; runtime 4,640 tests, 18 skips), 348 focused planner/runtime tests passed, all four hygiene gates passed for planner/runtime/perf, and the eight-domain Pinot review completed with no remaining findings. The refreshed 400k-row matched JMH comparison was noisy in both directions, so I reported the raw values in the PR body without making a performance claim.

Rebased onto master after #19412 and kept the responsibilities separate: sender sorting owns ordering; #19396 k-way merges only confirmed-sorted local/gRPC streams and safely falls back for mixed/custom transports.

Final coverage review added two targeted regressions for queued read-ahead preservation during fallback and buffered-cursor cleanup on sender error. The focused class passes 21/21, module formatting/checkstyle/license gates pass, and Codecov improved to 91.37% patch coverage.

Exact head 9642c66b99 is conflict-free and all 13 check runs pass. The linter setup failure was a transient Maven Central resolution miss; retrying only that job passed. No review threads remain; maintainer approval is the only remaining gate.

Rebased and simplified on top of master 257a0e7df2 (which includes #19412); the updated PR head is 7ecc01c0fd. The patch shrank from 38 to 27 files and from 2,379 to 1,957 changed lines, with the benchmark harness kept temporary and out of the PR.

Fresh matched JMH results on JDK 25, 400k rows, two servers, local + gRPC fan-in:

Both JMH-reported intervals are non-overlapping. Global is the stronger signal: every PR measurement was below every baseline measurement. The partitioned aggregate also favors the PR, but visible fork variance means 25.5% is this run's estimate, not a stable production-wide effect size.

Local validation passed: 89 focused runtime tests, all 243 QueryCompilationTest cases, and planner/runtime Spotless, Checkstyle, and license gates. Fresh exact-head CI is now running after the force-with-lease update.

I'd like to use this PR to fix how MSE assigns the responsibility for ordering. Today, ordering should be a requirement that the logical plan guarantees. Instead, the physical plan and the runtime keep getting extra machinery to guarantee it: sort/sortedOnSender on the exchange and mailbox nodes, SortedMailboxReceiveOperator, and now a per-block sorted.on.sender marker. I consider the marker a blocker.

The runtime has to trust the plan. Re-checking ordering on every block is like a JIT adding checks for invariants that the front-end compiler already proved. It is also fragile: MailboxSendOperator.isSortedOnSender() requires a SortOperator right below the send, so any order-preserving operator we add there later (a buffer, for example) silently disables the optimization. And it goes through the whole mailbox layer (GrpcSendingMailbox, InMemorySendingMailbox, MailboxService, ReceivingMailbox, MailboxContentObserver, ChannelUtils, BlockingMultiStreamConsumer) for an opt-in feature.

What I'd suggest instead is to express ordering with simple plan nodes, not flags:

• The sender has no ordering flags. If the receiver needs ordered streams, the plan guarantees the order upstream: usually with a Sort, but also by construction, for example after a sorted join.
• The receiver uses a new plan node: a receive that k-way merges streams that are each already sorted, with an optional limit/offset. That's SortedMailboxMergeReceiveOperator, and the optional limit also covers the streaming ORDER BY LIMIT in [MSE] Share ordering proof and stream ordered mailbox results #19711.
• When the plan doesn't guarantee order upstream, it uses a plain receive and a Sort downstream, as [MSE] Build ORDER BY on SortOperator instead of SortedMailboxReceiveOperator #19412 already does.

On the Calcite side, this could be a new PinotKWayMergeSortExchange: an exchange that requires its input to be sorted on the collation and preserves that order in its output. PlanFragmenter would turn it into a plain MailboxSendNode plus the new merge receive node. A rule similar to PinotSortExchangeCopyRule could push the limit/offset of a Sort above it into the receive.

In practice, for this PR that means removing the mailbox metadata and the sortedOnSender parameter of offer()/offerRaw(), MseBlockWithStats.isSortedOnSender(), isLastBlockSortedOnSender(), MailboxSendOperator.isSortedOnSender(), hasExplicitSortInput() and the full-sort fallback in the merge receiver. A per-row order check is fine in tests, but not as a production path.

Compatibility gets simpler too. On a new server:

For a new broker with old servers, it's the broker's responsibility not to send the new node and to keep generating a plan that old servers understand, i.e. a plain receive with a Sort downstream. We already have the mechanism for this: SendStatsPredicate in SAFE mode watches the instance configs and detects when any broker or server reports a different Pinot version. Today only servers use it to decide whether to send MSE stats, but the broker can reuse the same logic and only pick PinotKWayMergeSortExchange when the cluster is homogeneous. As a last line of defense, an old server can't deserialize the new node (PlanNodeDeserializer throws Unsupported PlanNode type), so the query fails loudly instead of returning wrong results.

With this, sortedOnSender and sort in plan.proto and the flags in PinotLogicalSortExchange can be deprecated, and SortedMailboxReceiveOperator can go away. Old ORDER BY plans already have a Sort with the fetch above the receive, so they get the bounded implementation with a LIMIT. For old window plans, the server injects a Sort without a fetch. That part can be a follow-up.

@gortiz updated at 01af90617f. Ordering is now a logical-plan guarantee: an explicit Sort feeds PinotKWayMergeSortExchange, and fragmentation creates a plain sender plus a distinct merge receive with serialized optional fetch/offset. Mailbox block markers, sender structural checks, production row-order checks, and the new merge receiver's full-sort fallback are removed.

The broker gate requires matching, non-SNAPSHOT broker/server release versions and fails closed on missing/unreadable configs, registration failure or disconnect; multi-cluster queries disable it. Unsupported clusters use plain receive plus an explicit MAX-fetch Sort, preserving complete window input. Legacy sort=true plans still use the deprecated full-sort receiver; its replacement and old-flag cleanup remain follow-up work. The gate is a point-in-time snapshot, so a concurrent version/topology change can still cause a loud unsupported-node query failure.

Recent unchanged planner/window coverage passed 962 cases, including full-input following/RANGE regressions. This latest commit only corrects the existing SNAPSHOT assertion (+7/-3 lines); both broker gate tests, JDK25 warning-enabled compilation, formatting, checkstyle and license checks passed, with no warnings on added lines. All 17 reported checks were green at b047dd4; CI is running on the new head. Please re-review the architecture and retained legacy compatibility path.

@gortiz rebased on master f818bae6d7 and pushed 85be3b0e80e4265ceeb6258e7db4582c4415511c with the review fixes. All 14 inline requests have individual evidence-backed replies and are resolved.

The current rebased gate reads only cached state on an enabled query, and shared incremental tracking watches live participant names and config notifications. The planner-internal ExchangeNode uses an explicit k-way mode; ordinary and merge receives are siblings under BaseMailboxReceiveNode with dedicated receive visitors. Legacy flags are deprecated. The single-run requirement is a serialized node hint, unused merge limits are removed, and the merge uses passive receive deadlines.

The current hybrid integration fixture compares receiver-sort and merge results through real cluster planner/proto/server/opchain execution for ROWS/FOLLOWING and RANGE/peer frames with nulls, duplicate keys, DESC NULLS FIRST and all 16 rows (8 offline + 8 realtime). It asserts complete results and a physical merge plan, but does not assert local/gRPC mailbox statistics; its test-only override resets the supplier to false in finally. The earlier integration pass and 968 planner/serde/plan-fixture cases, 177 focused runtime cases, tracker/capability/query-option tests and scoped formatting/license/style/warning checks are historical local validation, not a claim about fresh rebased-head CI.

For the non-inline comments: the earlier Calcite receiver MAX-fetch hardening, 124 JSON plan edits and unrelated annotations dependency are removed. The physical converter still encodes MAX_VALUE for complete-input window fallbacks on older servers that ignore the single-run hint. The query option is windowKWayMerge and the broker key is pinot.broker.multistage.window.k.way.merge. The no-op local variable and extraction wrapper are removed. The compact description now includes the config key, merge oneof tag 19, dedicated receive visitor and explicit SNAPSHOT exclusion. The final formatting cleanup removes the extra SenderCursor blank line (-1 line); the restack has no functional source changes. Heap remove/add and single-remaining-stream improvements are left out as the informational performance follow-up, without adding speculative optimization or benchmark code.

The feature remains opt-in and the memory limitation tracked by #19395 remains open. This update does not claim universal performance superiority or a validated AUTO selector. Focused stack validation passed 603 cases and the separate AUTO suite passed 417 cases on functionally identical trees; 36 scoped hygiene goals and four final runtime checks passed. All current-head CI suites now pass on the five final stack heads, without a failed-job rerun. Maintainer re-review and approval remain required; green CI does not establish acceptance.