feat(datafusion): report IcebergTableScan metrics

[geoffreyclaude]

Which issue does this PR close?

• Closes Report DataFusion operator metrics in IcebergTableScan #2364.

What changes are included in this PR?

IcebergTableScan now owns a DataFusion ExecutionPlanMetricsSet, returns it from ExecutionPlan::metrics(), and resets it from ExecutionPlan::reset_state().

The scan output stream is wrapped in a small poll_fn adapter that records BaselineMetrics while the Iceberg stream is polled. This exposes the standard DataFusion operator metrics such as elapsed_compute, output_rows, output_batches, output_bytes, and completion timestamps in EXPLAIN ANALYZE.

Focused tests cover the metrics wrapper directly and the catalog-backed provider execution path, including reset-state behavior.

Are these changes tested?

• cargo test -p iceberg-datafusion stream_with_baseline_metrics_records_rows_and_compute --locked
• cargo test -p iceberg-datafusion test_catalog_backed_provider_scan_reports_metrics --locked
• cargo check -p iceberg-datafusion --locked
• cargo test -p iceberg-datafusion --locked
• cargo clippy -p iceberg-datafusion --all-targets --locked

[mbutrovich]

I don't think baseline_metrics.clone() is needed. BaselineMetrics::elapsed_compute() and record_poll() both take &self, so the captured value can be used directly. The clone is cheap (Arc/Count copies) but it's dead weight on every poll and reads as if there's a borrow problem to work around. Suggest:

futures::stream::poll_fn(move |cx| {
    let _timer = baseline_metrics.elapsed_compute().timer();
    let poll = stream.as_mut().poll_next(cx);
    baseline_metrics.record_poll(poll)
})

[geoffreyclaude]

Good catch. Removed the per-poll clone() in 19db703.

This is cleaner and also avoids dropping a cloned BaselineMetrics on every poll, which could record end_timestamp earlier than intended.

[mbutrovich]

stream_with_baseline_metrics_records_rows_and_compute asserts output_rows and elapsed_compute. The PR description also lists output_batches, output_bytes, and completion timestamps as exposed. Those are in fact recorded — BaselineMetrics::record_poll → batch.record_output(...) updates output_batches and output_bytes (see datafusion/physical-expr-common/src/metrics/baseline.rs:331) — so adding assert!(metrics.output_batches() == Some(1)) and assert!(metrics.output_bytes().is_some_and(|b| b > 0)) is a cheap regression guard that matches the documented contract.

[geoffreyclaude]

Agreed on covering the rest of the baseline metrics. I added assertions for output_batches, output_bytes, start_timestamp, and end_timestamp in the focused stream_with_baseline_metrics test in 19db703.

I adapted this through MetricsSet::sum(...) + MetricValue matching because DataFusion 53.1 does not expose MetricsSet::output_batches() / output_bytes() convenience methods.

[mbutrovich]

Minor suggestions, thanks for tackling this @geoffreyclaude!

[geoffreyclaude]

Minor suggestions, thanks for tackling this @geoffreyclaude!

@mbutrovich Thanks for the quick review! PR should now be updated with your suggestions, as two additional commits.

I also had to add a small public API snapshot update in f94d61f after rebasing onto current main, since the new check-public-api CI validates crates/integrations/datafusion/public-api.txt.