Skip to content

Add shuffle-read fetch metrics and surface per-operator metrics in plan display#1968

Open
andygrove wants to merge 6 commits into
apache:mainfrom
andygrove:shuffle-read-metrics
Open

Add shuffle-read fetch metrics and surface per-operator metrics in plan display#1968
andygrove wants to merge 6 commits into
apache:mainfrom
andygrove:shuffle-read-metrics

Conversation

@andygrove

Copy link
Copy Markdown
Member

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 (feat: bound shuffle fetch with a reduce-side in-flight governor #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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add shuffle-read (fetch) metrics to ShuffleReaderExec and surface per-operator metrics in executor logs

1 participant