feat(datafusion): report IcebergTableScan metrics by geoffreyclaude · Pull Request #2521 · apache/iceberg-rust

geoffreyclaude · 2026-05-28T08:04:24Z

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 · 2026-05-29T20:19:44Z

+    baseline_metrics: BaselineMetrics,
+) -> Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> {
+    futures::stream::poll_fn(move |cx| {
+        let baseline_metrics = baseline_metrics.clone();


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) })

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 · 2026-05-29T20:20:21Z

+        let metrics = metrics.clone_inner();
+        assert_eq!(metrics.output_rows(), Some(3));
+        assert!(
+            metrics.elapsed_compute().is_some_and(|elapsed| elapsed > 0),


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.

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 · 2026-06-01T08:34:17Z

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.

github-actions · 2026-07-02T00:44:06Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

geoffreyclaude · 2026-07-02T07:27:19Z

@mbutrovich Could you please give this another look?

mbutrovich · 2026-07-02T13:05:22Z

@mbutrovich Could you please give this another look?

Sure thing, I will take a pass today. Thanks for keeping track of this!

mbutrovich

Thanks for the PR and addressing my previous feedback, @geoffreyclaude! LGTM

github-actions · 2026-08-03T00:33:31Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

github-actions · 2026-08-11T00:16:56Z

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

mbutrovich · 2026-08-11T18:14:51Z

I have reopened this since I think it's a valuable contribution.

github-actions · 2026-09-11T00:24:28Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

laskoviymishka

This already has @mbutrovich's approval, I'm just adding a second pair of eyes since it touches the public API surface. The metrics wiring reads clean and I've got no objection to merging; nothing below is blocking.

The one I'd genuinely tighten while we're here is the integration test. get_test_catalog_and_table() builds a table with no rows, so output_rows is 0 and assert_eq!(metrics.output_rows(), Some(output_rows)) passes whether or not the wrapper ever ran — BaselineMetrics starts that counter at 0. As written it can't catch a regression that bypasses metric collection. Inserting a small known row set and asserting against a count > 0 fixes that, and it also makes the neighbouring elapsed_compute > 0 check meaningful instead of flaky on a fast runner.

The other thing worth a note rather than a change: the unit test asserts output_batches/output_bytes, but the DF55 record_poll docs claim it only touches output_rows and end_time. If CI's green then it clearly does more than the docs say and we're fine — I'd just drop a one-line comment so the next reader doesn't "correct" the assertions.

A few smaller things I left inline:

the timestamp test helpers use MetricsSet::sum, which adds up point-in-time values — harmless at one partition, wrong beyond that
_timer has to stay a named binding; let _ = ... would silently zero elapsed_compute
reset_state could derive Clone instead of re-listing every field by hand

None of this blocks merge, i think can be merged as is. Will wait for day or two to merge

laskoviymishka · 2026-09-14T21:32:21Z

+        Some(self.metrics.clone_inner())
+    }
+
+    fn reset_state(self: Arc<Self>) -> DFResult<Arc<dyn ExecutionPlan>> {


minor, but reset_state re-lists every field by hand, so a new stateful field added later gets carried over silently unless someone remembers to touch this too.

Deriving Clone and writing it as let mut plan = (*self).clone(); plan.metrics = ExecutionPlanMetricsSet::new(); Ok(Arc::new(plan)) makes the intent — clone everything, reset the metrics — explicit and drops the boilerplate. Not blocking; the exhaustive literal is at least compile-safe today.

laskoviymishka · 2026-09-14T21:32:21Z

+    baseline_metrics: BaselineMetrics,
+) -> Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> {
+    futures::stream::poll_fn(move |cx| {
+        let _timer = baseline_metrics.elapsed_compute().timer();


worth a short comment that _timer has to stay a named binding — if someone tidies this to let _ = ...elapsed_compute().timer() the guard drops immediately and every elapsed_compute reading silently goes to 0, with the tests still passing. Quiet trap for the next editor.

laskoviymishka · 2026-09-14T21:32:21Z

+    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)


one thing I couldn't confirm from here: the DF55 docs for record_poll say it only updates output_rows and end_time, yet the unit test asserts output_batches == Some(1) and output_bytes == Some(...). If CI's green then record_poll clearly populates those in this version and we're fine — but the published docs contradict it, so the next person could reasonably assume the assertions are wrong and "fix" them.

A one-line comment noting that DF55's record_poll also tracks batches/bytes (or an explicit output_batches().add(1) if it turns out it doesn't) would head that off. Which is it in practice?

laskoviymishka · 2026-09-14T21:32:21Z

+    use super::stream_with_baseline_metrics;
+
+    #[test]
+    fn stream_with_baseline_metrics_records_rows_and_compute() {


this covers the happy single-batch path. Two cheap cases I'd add while it's fresh: an inner stream that yields Err(...) — assert the error passes through unchanged and elapsed_compute is still recorded, since error propagation is the likely real-world failure mode for a remote scan — and a two-batch stream to confirm output_rows accumulates across record_poll calls. Neither is blocking.

laskoviymishka · 2026-09-14T21:32:21Z

+        matches_metric: impl Fn(&MetricValue) -> bool,
+    ) -> Option<usize> {
+        metrics
+            .sum(|metric| matches_metric(metric.value()))


metric_value_as_usize uses MetricsSet::sum, which is right for the additive counters but off for StartTimestamp/EndTimestamp — those are point-in-time values, not quantities to add up. It works today only because there's a single partition, so there's exactly one matching entry; the moment this scan reports more than one output partition, sum adds two epoch-ns values and hands back garbage.

For the two timestamp helpers I'd iter().find(...) the matching metric instead of summing. Small thing, but it's a trap waiting on any future partitioning change.

laskoviymishka · 2026-09-14T21:32:21Z

+        let output_rows = batches.iter().map(|batch| batch.num_rows()).sum();
+
+        let metrics = scan_plan.metrics().expect("scan should expose metrics");
+        assert_eq!(metrics.output_rows(), Some(output_rows));


the table get_test_catalog_and_table() builds has no rows, so output_rows is 0 here and this assertion holds even if the wrapper never fires — BaselineMetrics initializes that counter to 0. So it can't actually catch a regression where metric collection gets bypassed.

I'd insert a small known row set before scanning and assert against that count (> 0). That also makes the elapsed_compute > 0 check below meaningful instead of flaky — an empty scan can finish inside the clock resolution on a fast runner and record 0ns. test_catalog_backed_provider_insert has the insert path. wdyt?

blackmwk · 2026-09-15T10:10:50Z

I think we should freeze this change. We are in the processing moving datafusion related things to apache/datafusion-iceberg repo.

geoffreyclaude · 2026-09-16T08:50:50Z

@laskoviymishka Thank you for the detailed review!

IIUC @blackmwk comment, and as discussed privately with @gabotechs, the recommended way forward would be to close this PR and re-open a similar one on https://github.com/apache/datafusion-iceberg once that repo is ready. I'll of course take that opportunity to integrate @laskoviymishka's suggestions.

blackmwk · 2026-09-17T07:08:40Z

@laskoviymishka Thank you for the detailed review!

IIUC @blackmwk comment, and as discussed privately with @gabotechs, the recommended way forward would be to close this PR and re-open a similar one on https://github.com/apache/datafusion-iceberg once that repo is ready. I'll of course take that opportunity to integrate @laskoviymishka's suggestions.

Yes, we should do iin the new repo. I'll close this pr for now.

geoffreyclaude force-pushed the fix/iceberg-scan-metrics branch from 9f036f5 to fe4a322 Compare May 28, 2026 08:25

geoffreyclaude marked this pull request as ready for review May 28, 2026 08:33

mbutrovich reviewed May 29, 2026

View reviewed changes

mbutrovich suggested changes May 29, 2026

View reviewed changes

geoffreyclaude force-pushed the fix/iceberg-scan-metrics branch 3 times, most recently from fb0a0ae to 6dc9f99 Compare June 1, 2026 08:32

geoffreyclaude requested a review from mbutrovich June 1, 2026 08:33

geoffreyclaude force-pushed the fix/iceberg-scan-metrics branch 2 times, most recently from e3293de to f94d61f Compare June 1, 2026 11:39

github-actions Bot added the stale label Jul 2, 2026

geoffreyclaude added 4 commits July 2, 2026 08:49

feat(datafusion): report IcebergTableScan metrics

eafd131

fix(datafusion): avoid cloning scan metrics per poll

b70116b

test(datafusion): cover all scan baseline metrics

8113fc0

chore(datafusion): update public api snapshot

c5b450e

geoffreyclaude force-pushed the fix/iceberg-scan-metrics branch from f94d61f to c5b450e Compare July 2, 2026 06:50

mbutrovich approved these changes Jul 2, 2026

View reviewed changes

github-actions Bot removed the stale label Jul 3, 2026

github-actions Bot added the stale label Aug 3, 2026

alexanderbianchi mentioned this pull request Aug 5, 2026

Expose detailed Iceberg DataFusion scan metrics DataDog/iceberg-rust#31

Open

This was referenced Aug 6, 2026

Support open-table formats datafusion-contrib/datafusion-distributed#533

Open

[EPIC] Iceberg integration datafusion-contrib/datafusion-distributed#599

Open

github-actions Bot closed this Aug 11, 2026

mbutrovich reopened this Aug 11, 2026

github-actions Bot removed the stale label Aug 12, 2026

gabotechs mentioned this pull request Aug 19, 2026

Moving the DataFusion integration into a separate repository #3029

Closed

github-actions Bot added the stale label Sep 11, 2026

laskoviymishka approved these changes Sep 15, 2026

View reviewed changes

github-actions Bot removed the stale label Sep 16, 2026

blackmwk closed this Sep 17, 2026

Conversation

geoffreyclaude commented May 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

What changes are included in this PR?

Are these changes tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

geoffreyclaude Jun 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

geoffreyclaude Jun 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

geoffreyclaude commented Jun 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Jul 2, 2026

Uh oh!

geoffreyclaude commented Jul 2, 2026

Uh oh!

mbutrovich commented Jul 2, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Aug 3, 2026

Uh oh!

github-actions Bot commented Aug 11, 2026

Uh oh!

mbutrovich commented Aug 11, 2026

Uh oh!

github-actions Bot commented Sep 11, 2026

Uh oh!

laskoviymishka left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

blackmwk commented Sep 15, 2026

Uh oh!

geoffreyclaude commented Sep 16, 2026

Uh oh!

blackmwk commented Sep 17, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

geoffreyclaude commented May 28, 2026 •

edited

Loading

geoffreyclaude Jun 1, 2026 •

edited

Loading

geoffreyclaude Jun 1, 2026 •

edited

Loading

geoffreyclaude commented Jun 1, 2026 •

edited

Loading

laskoviymishka left a comment •

edited

Loading