andygrove opened a new pull request, #2765:
URL: https://github.com/apache/datafusion-ballista/pull/2765
# Which issue does this PR close?
Closes #1953.
# Rationale for this change
When a query reads the same subplan more than once, such as a common table
expression joined to itself or TPC-H q15's revenue view, the adaptive planner
places one exchange per copy and runs the subplan once per reader. #1954 added
exchange reuse to the static planner only, which now runs only when AQE is
switched off, so it was closed in favour of doing this in AQE.
# What changes are included in this PR?
- **`ReuseExchangeRule`**, the last rule of the AQE replan chain (after an
embedder's replan rules, so keys cover each stage as it will run).
- Each `ExchangeExec` is keyed by the protobuf encoding of its subtree
under the job's physical extension codec. Nested exchanges are encoded by what
they write (partitioning, broadcast), never by stage id. A subtree the codec
can't encode is left alone.
- Each duplicate becomes a reader of one stage. It shares the canonical
exchange's stage id, shuffle partitions, column statistics and range routing,
and keeps its own coalesce and skew-join slots, since each consuming stage
makes those decisions for its own alignment group.
- The canonical exchange of a key is one already resolved, else one
already running, else the first met bottom-up. An exchange already running a
stage of its own is never re-pointed, so a copy found after its twin finished
just reads the finished output.
- Never shared: subtrees with a volatile expression (`random()`), and
range-shuffled subtrees and the window rewrites built on them, whose completion
reads state off a single consumer.
- A pending stage whose readers diverged, for example one reader turned
into a broadcast, is split back into separate stages before launch.
- **`identify_runnable_stages`** launches each stage id once, since every
reader of a shared stage is found.
- **Helpers on `ExchangeExec`:** `share_stage_of`, `shares_stage_with`,
`with_own_stage`.
- **New config `ballista.planner.adaptive.reuse_exchange`**, default `true`.
- **Tests:** planner and graph tests for union, self-join and q15-shaped
queries, the config switch, volatile subtrees, late duplicates, already-running
stages and diverged readers; a test that losing a shared stage's output rolls
back that stage once along with its reader; and an end-to-end standalone test
comparing rows with reuse on and off against the static planner.
Losing the executor that holds a finished stage's output still can't be
recovered under AQE, with or without reuse (#1986, #2539). This PR doesn't
change that: a shared stage is reset once and its readers are rolled back, as
for a single-reader stage.
# Are there any user-facing changes?
Yes. Queries that read a subplan more than once run fewer stages under AQE.
The new config `ballista.planner.adaptive.reuse_exchange` (default `true`)
switches it off. It's documented in the configuration reference and in the
tuning guide's AQE section. `SessionConfigExt` gains
`ballista_adaptive_reuse_exchange_enabled` /
`with_ballista_adaptive_reuse_exchange_enabled`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
