fix: count Parquet decoding in the scan's elapsed_compute

[akashjainn]

Which issue does this PR close?

• Part of Improve metrics in DataSourceExec with Parquet source #18195. This is the last unchecked item, "Fix elapsed_compute baseline metrics not counting issue".

Rationale for this change

EXPLAIN ANALYZE reports an elapsed_compute for a Parquet scan that is far too small. For a single-partition SELECT * over a 1M row file, the scan takes about 155 ms and reports about 0.1 ms of compute, so the metric is useless for telling whether a query is bound by the scan.

#20767 added an elapsed_compute timer in PushDecoderStreamState::transition, but it wraps copy_arrow_reader_metrics and project_batch only. The decoding happens before the timer starts (in try_decode at the time of that PR, in reader.next() today), so the metric went from a few nanoseconds to a fraction of a millisecond and is still missing where the time goes. The CSV fix in #18901 times reader.next(), and this PR does the equivalent for Parquet.

What changes are included in this PR?

PushDecoderStreamState::transition now runs one timer for the whole transition and pauses it only across get_byte_ranges(...).await. Everything else in the loop is CPU work: decoding in reader.next(), row group pruning and decoder rebuilds at row group boundaries, try_next_reader, push_ranges and the projection. That matches the guidance in the issue to count metadata handling and payload decoding.

On the overhead concern raised on #20767: there is still one timer start per batch, as there is today. The only additional timer operations are a stop and a restart around each byte range fetch, which happens per row group rather than per batch.

The timer is created from a clone of the Time metric, which shares the underlying counter, so the guard does not borrow self across the &mut self calls in the loop. It records when it is dropped, which covers every return.

What is the testing strategy for this PR?

New test parquet_scan_elapsed_compute_includes_decoding in datafusion/core/tests/sql/explain_analyze.rs. It writes a 1M row Parquet file, scans it with one target partition so the scan runs on one thread, and asserts that the scan's elapsed_compute is at least a quarter of the wall time of the scan. The existing tests only assert that elapsed_compute is greater than zero, which a timer that misses decoding still satisfies. This is also an answer to the question on #20767 of whether the metric can be tested: comparing against wall time measured in the same run avoids depending on machine speed.

Measured locally in a debug build, three runs each, within 1% of each other:

	elapsed_compute	wall time
main	0.10 ms	155 ms
this PR	151 ms	155 ms

The test fails on main and passes with this change, and it passed five consecutive runs. The threshold of a quarter leaves a wide margin on both sides, and because it compares against wall time measured in the same run it does not depend on machine speed.

Is the new number right? Two checks, done with a throwaway test that is not part of this PR, on the same 1M row file:

• Against arrow-rs alone. Decoding the file with ParquetRecordBatchReaderBuilder at a batch size of 8192, with no DataFusion involved, took 145 to 148 ms over three runs. The scan's elapsed_compute with this change was 145 to 147 ms.
• I/O is excluded. With the file served from an InMemory store wrapped in object_store's ThrottledStore, which sleeps on every get:

injected delay per get	wall time	elapsed_compute
0 ms	149 ms	149 ms
100 ms	274 ms	167 ms
300 ms	471 ms	167 ms

Tripling the delay adds 200 ms of wall time and leaves elapsed_compute unchanged. The step from 149 to 167 ms is CPU time and not a leak of I/O wait: wall time minus the injected delay is about 171 to 174 ms in those runs, so the work itself was slower once the thread had gone idle.

The metric also scales with the work: 500k rows report 77 ms against 147 ms for 1M, a single Int64 column reports 38 ms, and a single string column 76 ms. In every case elapsed_compute stayed below the wall time, by 0.5 to 5 ms.

Also run locally: cargo fmt --all -- --check, cargo clippy --all-targets --all-features -- -D warnings for datafusion-datasource-parquet and for the core_integration test target, cargo test -p datafusion-datasource-parquet (267 passed), the sql::explain_analyze tests (30 passed) and the parquet_integration tests (242 passed).

Are there any user-facing changes?

elapsed_compute for Parquet scans in EXPLAIN ANALYZE becomes much larger, because it now includes decoding. No API changes.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.36%. Comparing base (cb14804) to head (d707e5e).
⚠️ Report is 5 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #25486   +/-   ##
=======================================
  Coverage   82.36%   82.36%           
=======================================
  Files        1137     1137           
  Lines      432962   433027   +65     
  Branches   432962   433027   +65     
=======================================
+ Hits       356610   356667   +57     
- Misses      54825    54827    +2     
- Partials    21527    21533    +6     

☔ 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.

[jayzhan211]

Thanks @akashjainn, make sense to me