Add dependency-ordered staged dispatch

[wirybeaver]

PR flow

Staged dispatch with materialized exchange: plan marked exchange, build dependency graph, dispatch groups when ready, serve via file-backed mailbox, broker validates outputs.

flowchart TD
  N0["Mark exchange as materialized #40;F5#44; F6#41;"]:::stModified
  N1["Build dispatch graph #40;barriers for materialized edges#41; #40;F22#41;"]:::stAdded
  N2["Check for ready groups #40;dependencies met#41; #40;F22#41;"]:::stAdded
  N3["Dispatch group #40;run reducer if stage 0 present#41; #40;F22#41;"]:::stModified
  N4["Server#58; Create materialized sending mailbox per partition #40;F18#41;"]:::stModified
  N5["Server#58; Commit partition on EOS and return handle #40;F16#41;"]:::stAdded
  N6["Server#58; Include handle in OpChainComplete #40;F26#41;"]:::stModified
  N7["Broker#58; Collect handle and wait for workers#47;streams #40;F25#41;"]:::stModified
  N8["Broker#58; Mark group as completed #40;F22#41;"]:::stAdded
  N9["Broker#58; Validate materialized outputs #40;after all groups#41; #40;F22#41;"]:::stModified
  N0 -->|"materialized edge input"| N1
  N1 -->|"graph used for readiness"| N2
  N2 -->|"dispatch ready group"| N3
  N3 -->|"materialized send node #45;#62; create mailbox"| N4
  N4 -->|"send data #45;#62; commit on EOS"| N5
  N5 -->|"commit #45;#62; include handle in OpChainComplete"| N6
  N6 -->|"broker collects handle and waits"| N7
  N7 -->|"workers#47;streams done #45;#62; mark group completed"| N8
  N8 -->|"completed group may enable others"| N2
  N8 -->|"after all groups#58; validate outputs"| N9
  classDef stAdded fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
  classDef stModified fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
  classDef stRemoved fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
  classDef stUnchanged fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px

Loading

AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing

Partial evidence: 3 file patches omitted; 0 truncated.

Diff evidence

• F5: pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java — before · after
• F6: pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/MaterializedExchangePlanner.java — after
• F16: pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/materialized/MaterializedSendingMailbox.java — after
• F18: pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java — before · after
• F22: pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java — before · after
• F25: pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/streaming/StreamingQuerySession.java — before · after
• F26: pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java — before · after

• Regenerate PR flow

Summary

Adds Phase 2 of the AQE prerequisite stack: opt-in staged dispatch with unchanged worker assignments.

Depends on #19471.

Design

                    materialized dependency
producer group A  ----------------------------+
      | all unique workers complete            |
      | all streams emit ServerDone             v
      +----------------------------------> consumer group B
                                               |
                                      root group contains 0?
                                               |
                                    run reducer locally

streaming / pipeline-breaker edge:
    stage X ===== contracted dispatch group ===== stage Y

The broker builds a deterministic dispatch-group DAG. Materialized edges form completion barriers; streaming-connected stages are dispatched together. A group transitions only after every expected (stageId, workerId) reports success and every stream closes. Duplicate or unexpected reports cannot release the barrier.

Invariants

• default OFF via stagedDispatch=true
• one absolute deadline across all waves
• assignments, stage IDs, and worker IDs remain immutable
• a failed worker or stream prevents downstream dispatch
• stage 0 participates in dependency ordering but executes locally
• no retries or running-stage replacement

Test Plan

• graph/session/observer/dispatcher tests: 110 passed
• Spotless, Checkstyle, and license checks passed for pinot-common, pinot-spi, and pinot-query-runtime

[codecov-commenter]

Codecov Report

❌ Patch coverage is 71.23936% with 304 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.74%. Comparing base (e6787e1) to head (22f5c59).

Files with missing lines	Patch %	Lines
.../pinot/query/service/dispatch/QueryDispatcher.java	20.43%	105 Missing and 4 partials ⚠️
...mailbox/materialized/MaterializedMailboxStore.java	78.57%	42 Missing and 15 partials ⚠️
...ilbox/materialized/MaterializedSendingMailbox.java	46.80%	23 Missing and 2 partials ⚠️
...pinot/query/mailbox/channel/GrpcMailboxServer.java	77.01%	11 Missing and 9 partials ⚠️
...ailbox/materialized/MaterializedMailboxWriter.java	65.45%	17 Missing and 2 partials ⚠️
.../planner/physical/MaterializedExchangePlanner.java	41.93%	4 Missing and 14 partials ⚠️
...org/apache/pinot/query/mailbox/MailboxService.java	79.26%	10 Missing and 7 partials ⚠️
...not/query/service/dispatch/StageDispatchGraph.java	93.95%	4 Missing and 5 partials ⚠️
...e/operator/MaterializedMailboxReceiveOperator.java	86.53%	5 Missing and 2 partials ⚠️
...y/mailbox/materialized/MaterializedMailboxKey.java	73.68%	2 Missing and 3 partials ⚠️
... and 8 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19472      +/-   ##
============================================
+ Coverage     67.69%   67.74%   +0.05%     
  Complexity     1430     1430              
============================================
  Files          3488     3495       +7     
  Lines        224306   225327    +1021     
  Branches      35416    35582     +166     
============================================
+ Hits         151850   152658     +808     
- Misses        60447    60575     +128     
- Partials      12009    12094      +85     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.74% <71.23%> (+0.05%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.74% <71.23%> (+0.05%)	⬆️
unittests	67.74% <71.23%> (+0.05%)	⬆️
unittests1	57.92% <71.23%> (+0.12%)	⬆️
unittests2	39.24% <2.83%> (-0.17%)	⬇️

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.