andygrove opened a new pull request, #1968:
URL: https://github.com/apache/datafusion-ballista/pull/1968
# Which issue does this PR close?
Closes #1958.
# Rationale for this change
Diagnosing shuffle-heavy queries (for example the SF100 TPC-H
Ballista-vs-Comet gap) is currently blind. `ShuffleReaderExec` exposes only
`BaselineMetrics` (`elapsed_compute`/`output_rows`), so the
shuffle-read/transport cost — the time spent fetching remote partitions, the
bytes moved, and the time blocked on the reduce-side in-flight governor (#1951)
— is unmeasurable. Separately, the executor's `LoggingMetricsCollector` renders
only the shuffle-writer wrapper metrics, so inner-operator compute
(`SortExec`/`SortMergeJoinExec`/`DataSourceExec` `elapsed_compute`/spill) never
appears even though DataFusion already collects it. Together these make it
impossible to attribute where a shuffle-heavy query spends its time.
# What changes are included in this PR?
1. **Fetch metrics on `ShuffleReaderExec`.** A `ShuffleReadMetrics` struct
(mirroring the existing `ShuffleWriteMetrics`) recorded per output partition
and threaded through `send_fetch_partitions` into the concurrent local/remote
fetch tasks:
- `fetch_time` — wall-time fetching remote partitions (Arrow-Flight fetch
+ buffering)
- `local_read_time` — wall-time opening node-local shuffle files
- `permit_wait_time` — a single combined timer over the three reduce-side
governor semaphores (request + per-address + byte), i.e. backpressure wait
(#1951)
- `decoded_bytes` — decoded in-memory Arrow footprint of fetched remote
partitions (not compressed wire bytes)
- `fetch_requests` — total remote fetch attempts issued, including retries
- `fetch_retries` — of those, how many were retries
- `local_partitions` / `remote_partitions` — the local/remote read split
All metrics go through the standard `MetricBuilder`/`MetricsSet`
machinery, so they also appear in `EXPLAIN ANALYZE`. The handles are
`Arc`-backed atomics cloned into the concurrent fetch tasks (no added locking).
Doc comments note that `fetch_time`/`permit_wait_time` are summed across
concurrent tasks (aggregate cost, not wall-clock) and that `fetch_time` is
additive to — not a re-slice of — the reader's existing `elapsed_compute`.
2. **Per-operator metrics in the plan display.**
`ShuffleWriterExec::Display` now renders the child plan with metrics
(`DisplayableExecutionPlan::with_metrics`, `ShowMetrics::Aggregated`) in
addition to statistics, so inner-operator `elapsed_compute`/spill/output-rows
surface in `LoggingMetricsCollector` output and `EXPLAIN ANALYZE`. (The
originally-considered `set_show_metrics` setter does not exist on
`DisplayableExecutionPlan`; `with_metrics` is the equivalent constructor and is
already used elsewhere in the repo.)
# Are there any user-facing changes?
No API changes. The additional metrics are visible in `EXPLAIN ANALYZE`
output and executor metrics logs.
--
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]
