[MSE] Support sender-side sorting for sort exchanges

[xiangfu0]

Fixes #19395.

PinotLogicalSortExchange already carried sender/receiver sort flags, but the runtime did not honor them: senders streamed unsorted input and SortedMailboxReceiveOperator buffered and sorted the complete exchange. This change implements sender-side sorting for window exchanges and lets receivers k-way merge the sorted streams, so a window with ORDER BY can begin producing rows without retaining the full exchanged input at one receiver.

Runtime behavior

• MailboxSendOperator sorts the complete sender input with the exchange collation, including direction and null ordering, then emits sequential blocks of at most 10k rows. Existing exchange implementations preserve the order of each destination's subsequence.
• SortedMailboxReceiveOperator merges one cursor per sender when the plan requests sender sorting. It drains whichever mailbox is ready to avoid cross-receiver backpressure deadlocks and emits merged blocks of at most 10k rows.
• The existing buffer-everything-and-sort path remains the fallback for plain ORDER BY exchanges and any sender that does not confirm the new protocol.
• A single sorted sender bypasses the merge heap without eagerly deserializing its blocks.
• Merge and sender-sort loops enforce the active deadline and periodically sample query resource usage.

PinotWindowExchangeNodeInsertRule enables sender sorting for ordered window exchanges. Plain sort and aggregate exchanges retain receiver-side sorting.

Mixed-version safety

Every data block from a sorting sender now carries a sortedOnSender confirmation through both in-memory and gRPC mailboxes. The gRPC confirmation is transport metadata rather than part of the serialized data block.

An upgraded receiver tentatively merges only confirmed streams. Before it can emit its first row it must have a head row from every non-empty live sender; if any sender is old (or uses a mailbox implementation that cannot carry the confirmation), that sender's first data block has no marker and the receiver folds all tentatively buffered rows into the existing full buffer-and-sort path. The confirmation is checked on every data block. An old empty sender is harmless because it contributes no rows.

Older receivers ignore the additional metadata and continue to full-sort. The new SendingMailbox.send(data, sortedOnSender) method is a default method, so existing mailbox implementations remain source- and binary-compatible and cause the safe fallback. No server-before-broker upgrade ordering is required.

Review fixes

The review also tightened the implementation beyond the initial PR:

• Preserved every already-buffered and not-yet-emitted row when switching from tentative merge to full sort.
• Replaced repeated fan-in-wide liveness scans with identity-based cursor sets and targeted EOS updates.
• Kept synthetic timeout/error blocks from being attributed to the previously read sender.
• Released sender-side sorted chunk lists as they were transmitted instead of retaining the complete run twice.
• Released each receiver cursor's final block as soon as it was drained instead of retaining it behind a straggler.
• On early termination, released merge read-ahead immediately while still draining raced data to preserve EOS stats
  and sender errors.
• Added active-deadline/resource checks to long reads, sorts, and merge loops.
• Kept the existing cancellation and backpressure paths intact.

Tests and hygiene

• Production dispatch oracles construct MailboxSendOperator from a real MailboxSendNode, verify sorted output reaches the terminal SendingMailbox through send(data, true), reject the legacy one-argument path, and cover the nested multi-receiver exchange.
• Mixed-version tests cover legacy senders over both transports, fallback after tentative buffering, empty senders, confirmation on every fragmented block, single-sender pass-through, errors/timeouts, early termination, and 10k-block fan-in stress.
• Two-server H2-backed window cases cover global and partitioned ordered windows, duplicates, null ordering, filters, and empty streams.
• ./mvnw -pl pinot-query-runtime -am test: 20-module reactor success; planner 1,539 tests, runtime 4,613 tests, 0 failures/errors (18 runtime skips), followed by the final runtime-only suite at 4,616 tests, 0 failures/errors (18 skips).
• spotless:apply, checkstyle:check, license:format, and license:check pass for
  pinot-query-planner,pinot-query-runtime,pinot-perf.

Base/head benchmark

Exact comparison:

• Base: bbbed251444e00d58bdcbe59a6f233902eb7e79c
• Head: afe7a5db057776482049fdec6f93b15695043cea
• JDK 25.0.4, two servers, four 10k-row segments, one local plus one gRPC sender for the global fan-in.
• JMH: 1 fork, 2 x 3s warmups, 4 x 3s measurements, legacy stats mode, with JFR and memory-pool profilers.

Workload	Base	Head	Delta	Peak managed memory
Global ordered window	15.666 +/- 5.245 ms/op	16.092 +/- 6.723 ms/op	+2.72%	2,766,036 -> 2,772,254 KiB (+0.22%)
Partitioned ordered window	12.187 +/- 6.063 ms/op	12.434 +/- 3.914 ms/op	+2.03%	2,817,272 -> 2,800,814 KiB (-0.58%)

Both point deltas are small relative to the reported uncertainty, and the intervals overlap heavily, so this run does not support a latency improvement or regression claim. Peak process RSS was 3,269,705,728 bytes for the base and 3,323,641,856 bytes for the head (+1.65%). The recordings cover one row scale and a fixed two-sender topology. JFR captured allocation behavior, but this JDK/JMH combination did not provide a reliable normalized bytes/op result. Both benchmark forks completed their measurements and results before JMH forcibly stopped leftover integration-test threads during teardown.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.72810% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.59%. Comparing base (bbbed25) to head (afe7a5d).

Files with missing lines	Patch %	Lines
...runtime/operator/SortedMailboxReceiveOperator.java	89.88%	7 Missing and 10 partials ⚠️
...apache/pinot/query/mailbox/GrpcSendingMailbox.java	66.66%	7 Missing ⚠️
...ot/query/runtime/operator/MailboxSendOperator.java	91.07%	2 Missing and 3 partials ⚠️
...y/runtime/operator/exchange/BroadcastExchange.java	33.33%	2 Missing ⚠️
...uery/runtime/operator/exchange/RandomExchange.java	66.66%	2 Missing ⚠️
...query/runtime/operator/exchange/BlockExchange.java	93.33%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19396      +/-   ##
============================================
+ Coverage     57.71%   67.59%   +9.88%     
- Complexity        7     1430    +1423     
============================================
  Files          2686     3487     +801     
  Lines        163987   224441   +60454     
  Branches      26627    35424    +8797     
============================================
+ Hits          94640   151716   +57076     
+ Misses        61352    60685     -667     
- Partials       7995    12040    +4045     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.59% <89.72%> (+9.88%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.59% <89.72%> (+9.88%)	⬆️
unittests	67.59% <89.72%> (+9.88%)	⬆️
unittests1	57.73% <89.72%> (+0.02%)	⬆️
unittests2	39.27% <0.00%> (?)

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.