feat(parquet): add bytes_processed scan-completion metric

[adriangb]

Which issue does this PR close?

• No issue filed yet. Happy to open one describing the problem before review if that is preferred.

Rationale for this change

Nothing today reports how far along a scan is. That is what a progress bar, a query-progress API, or a watchdog that cancels runaway queries needs, and both existing signals fail at it in opposite directions:

• bytes_scanned looks like the right metric, because its natural denominator — the size of the files in the plan — is known before the query runs. But it counts only the bytes actually fetched. A scan that prunes most of its row groups, or projects 3 of 50 columns, reports a small fraction of the file even once it has finished, so bytes_scanned / total file bytes understates progress by a factor that varies per query and is not knowable up front.
• files_processed only moves when a whole file completes. With high scan parallelism it reads 0% through the entire first wave of files, however much work is in flight, and then jumps.

The gap is not the unit, it is the numerator: a scan is done with a byte once it has either read it or proved it does not need it, and bytes_scanned counts only the first half.

What changes are included in this PR?

A new bytes_processed counter on ParquetFileMetrics, alongside bytes_scanned and sharing its metric type, category and per-file label:

bytes_scanned    — bytes fetched from the object store
bytes_processed  — bytes the scan is finished with, whether read or pruned

Its contract is one invariant:

Over the lifetime of a file — or, for a file split into byte ranges for parallelism, a range — bytes_processed advances by exactly effective_size(), monotonically.

so bytes_processed / total file bytes is a completion fraction that needs no statistics and no second metric to normalise against.

Credit lands a row group at a time:

event	credit
file/range pruned before open (file statistics, dynamic filter)	effective_size()
row groups the final access plan skips — range, statistics, bloom filter, limit, page index	compressed_size(rg), at open
row group dropped mid-scan by a dynamic filter	compressed_size(rg)
row group reached by the decoder	compressed_size(rg)
file closed for any reason (finished, LIMIT, early stop, error)	the remainder

Two details worth a reviewer's attention:

• ByteProgress (in metrics.rs) clamps every credit to the bytes left in the range and credits the remainder on Drop. The clamp absorbs the two inexactnesses of crediting by row group — a file is slightly larger than the sum of its row groups, and a row group is assigned to a byte range by its first page's offset — and Drop is what makes the invariant hold under LIMIT, EarlyStoppingStream and errors without instrumenting every exit path.
• row_group_in_range is extracted out of RowGroupAccessPlanFilter::prune_by_range so the open-time skip pass decides which row groups a range owns by the same rule, instead of duplicating the offset logic. Without it, one range of a split file would credit its siblings' row groups against its own budget and jump straight to 100% at open.

Folding every open-time pruning stage into a single pass over the final access plan is what keeps this free of double counting: range, statistics, bloom, limit and page-index pruning have all been applied by then, so there is no need to instrument five sites and reason about their overlap.

Cost is one atomic add per row group. Crediting a row group's bytes progressively as its rows decode — proportionally by rows resolved, trued up at the boundary — is deliberately left to a follow-up; it does not change this metric's contract.

Are these changes tested?

Yes. Six new tests in opener::test::bytes_processed, all asserting the invariant, for: a plain scan, a scan pruning row groups, a file pruned before open by a dynamic filter (also asserting bytes_scanned == 0), a byte-split file (each range credits exactly its own length, and the ranges sum to the file), a LIMIT that ends the scan early, and one that steps through the stream batch by batch to pin that credit advances during the scan rather than all at once on close.

The two carrying the real claims were mutation-checked: removing the mid-scan credit fails credit_advances_while_the_scan_runs, and removing the open-time credit fails row_group_pruning_is_credited_before_any_batch_is_read. The other four would also pass a trivial credit-everything-on-close implementation, so those two are the ones doing the work.

Also run: the full datafusion-datasource-parquet suite, the explain_analyze core tests, the four affected sqllogictest files, cargo fmt, cargo clippy --all-targets --all-features -- -D warnings, and rustdoc with -D warnings.

Are there any user-facing changes?

• EXPLAIN ANALYZE on a parquet scan gains a bytes_processed= entry, next to bytes_scanned. Four sqllogictest files are updated; the diff there is exactly eight added bytes_processed= fragments and nothing else — existing <slt:ignore> markers and pinned values are preserved.
• A bullet in docs/source/user-guide/explain-usage.md documenting the metric next to bytes_scanned.
• ParquetFileMetrics gains a public field. The struct is documented as subject to change and is normally built through ParquetFileMetrics::new, but external code constructing it with a struct literal would need updating — flagging in case this warrants the api change label.

---

🤖 Generated with Claude Code

https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3

---

Generated by Claude Code

[saadtajwar]

This is really nice! Just left a small code-styling suggestion

[saadtajwar]

nit: I think we could make this a bit cleaner by collapsing 1519-1522, suggestion below:

let in_range_bytes: u64 = rg_metadata
    .iter()
    .filter(|rg_meta| {
        prepared
            .file_range
            .as_ref()
            .is_none_or(|range| row_group_in_range(rg_meta, range))
    })
    .map(row_group_bytes)
    .sum();
let skipped_bytes: u64 = in_range_bytes.saturating_sub(rg_plan.iter().map(|e| e.bytes).sum());

[adriangb]

Thanks addressed!

[Copilot]

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a new Parquet scan completion metric (bytes_processed) that advances monotonically to the file/range effective_size(), enabling reliable scan-progress reporting even with pruning and projection pushdown.

Changes:

• Introduces ParquetFileMetrics::bytes_processed plus ByteProgress to safely and monotonically account bytes (including on early stop / errors via Drop).
• Updates Parquet scan/open/decoder logic to credit processed bytes at open-time (for pruned RGs) and during decode (per RG), including split-range correctness via shared row_group_in_range.
• Updates EXPLAIN ANALYZE outputs, sqllogictest expectations, and user docs to include bytes_processed.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
docs/source/user-guide/explain-usage.md	Documents the new bytes_processed EXPLAIN metric and its interpretation.
datafusion/sqllogictest/test_files/*.slt	Updates expected EXPLAIN ANALYZE metric strings to include bytes_processed.
datafusion/datasource-parquet/src/row_group_filter.rs	Extracts row_group_in_range to unify range ownership logic for row groups.
datafusion/datasource-parquet/src/push_decoder.rs	Credits bytes_processed as row groups are pruned/decoded/advanced.
datafusion/datasource-parquet/src/opener/mod.rs	Creates and wires ByteProgress, credits open-time skipped bytes, and adds invariant tests.
datafusion/datasource-parquet/src/opener/early_stop.rs	Ensures inner stream is dropped immediately on early stop to complete byte accounting promptly.
datafusion/datasource-parquet/src/metrics.rs	Adds bytes_processed metric and ByteProgress guard implementation.
datafusion/core/tests/sql/explain_analyze.rs	Asserts EXPLAIN ANALYZE output contains bytes_processed=.

Suppressed comments (1)

datafusion/datasource-parquet/src/metrics.rs:1

• Adding a new pub field to a public struct is a semver-breaking change for external users constructing ParquetFileMetrics with a struct literal. If ParquetFileMetrics is intended to be publicly constructible, consider mitigating this by marking the struct #[non_exhaustive] (if not already), or making fields private and exposing construction/access via new()/accessors so future metric additions don’t break downstream builds.

// Licensed to the Apache Software Foundation (ASF) under one

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 98.12332% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.32%. Comparing base (c429919) to head (d37ca45).
⚠️ Report is 27 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/datasource-parquet/src/push_decoder.rs	86.11%	3 Missing and 2 partials ⚠️
...fusion/datasource-parquet/src/opener/early_stop.rs	98.90%	1 Missing ⚠️
...afusion/datasource-parquet/src/row_group_filter.rs	87.50%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24524      +/-   ##
==========================================
+ Coverage   81.24%   81.32%   +0.07%     
==========================================
  Files        1113     1117       +4     
  Lines      392744   396255    +3511     
  Branches   392744   396255    +3511     
==========================================
+ Hits       319090   322251    +3161     
- Misses      54900    55182     +282     
- Partials    18754    18822      +68     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.