feat(datafusion): Add opt-in eager file scan planning with output partitioning by toutane · Pull Request #2671 · apache/iceberg-rust

toutane · 2026-06-18T16:24:37Z

Which issue does this PR close?

What changes are included in this PR?

This PR adds an opt-in eager scan planning path for the DataFusion integration.

When iceberg.enable_eager_scan_planning is enabled, IcebergTableProvider::scan() plans FileScanTasks during physical planning, groups them across DataFusion target_partitions, and exposes the resulting output partition count as UnknownPartitioning(N). Each execute(partition) call then reads only the task group assigned to that output partition through ArrowReaderBuilder.

The default behavior is unchanged: eager planning is disabled by default, so scans keep the existing lazy single-partition planning path unless explicitly enabled.

Eager planning builds a single TableScan per query: the scan that plans the FileScanTasks is retained and reused by every execute(partition) call to derive its ArrowReaderBuilder. Both the eager and the lazy path therefore source their reader settings from TableScan::arrow_reader_builder(), so the two cannot silently drift apart.

When a LIMIT is pushed into the scan, eager planning applies it as a per-partition bound. The final global bound is enforced above the scan by the optimized DataFusion plan, currently through CoalescePartitionsExec::fetch.

This is a narrower split from #2298, focused on file-level eager planning and output partition count reporting. Some of the broader design discussions and review feedback happened in #2298; this PR keeps only the first scoped step and intentionally does not include hash partitioning, size-aware bin-packing, or row-group/sub-file planning.

Public API changes:

Adds one public method to the core crate: iceberg::scan::TableScan::arrow_reader_builder() -> ArrowReaderBuilder. It extracts the reader construction that to_arrow() already performed, so the DataFusion integration can build a reader from a planned scan without duplicating the reader defaults.

Trade-offs:

Eager planning does catalog/metadata work during TableProvider::scan(), so it is kept opt-in
Task grouping is round-robin and count-based, not size-aware. This is simple and deterministic, but can be imbalanced when file sizes vary (tracked by [iceberg-datafusion] Balance eager scan task groups by estimated read size #2962)
Parallelism is at the FileScanTask level only. A table with one large file will not benefit from this change
The scan reports UnknownPartitioning(N), not hash partitioning. This exposes the number of output partitions without claiming stronger partitioning semantics
Eager planning retains the planning TableScan, and hence its PlanContext and warm evaluator caches, for the lifetime of the physical plan

Follow-up work:

Add size-aware task bin-packing using file_size_in_bytes (Plan file scan task according scan file size. #128)
Add sub-file / row-group level parallelism as the longer-term direction (EPIC: Support parallel scan in iceberg-datafusion #1604)
Cache planned tasks for repeated scans with the same snapshot, projection, filter, and target partition count (tracked by [iceberg-datafusion] Cache eager scan planning across equivalent scans #2963)
Investigate stronger output partitioning declarations for safe identity-partitioned cases
Add benchmarks for multi-file tables and skewed file-size distributions

Are these changes tested?

Yes. Integration tests cover:

eager scan planning disabled by default (test_multi_file_scan_defaults_to_single_lazy_partition)
enabling eager scan planning through iceberg.enable_eager_scan_planning, including at runtime via SET (test_set_enable_eager_scan_planning)
exposing multiple scan output partitions, clamped to the planned task count so that no empty partition is exposed (test_multi_file_scan_produces_multiple_partitions)
preserving query results between the lazy path, single-partition eager planning, and multi-partition eager planning (test_multi_partition_scan_matches_single_partition_results)
enforcing a global LIMIT across multiple output partitions (test_multi_partition_scan_enforces_global_limit)
enforcing a global ORDER BY ... LIMIT across multiple output partitions (test_multi_partition_ordered_scan_enforces_global_limit)
rejecting non-empty children on the leaf scan node (test_iceberg_table_scan_rejects_non_empty_children)

AI Disclosure

This PR was developed with AI assistance, using Claude Code (Opus 4.8)

toutane · 2026-06-30T15:24:17Z

👋 cc @mbutrovich @timsaucer

mbutrovich

First pass, thanks @toutane!

toutane · 2026-07-10T12:58:06Z

Thanks @mbutrovich, addressed all four comments in d99603f

toutane · 2026-07-16T08:38:37Z

@mbutrovich Could you please give this another look?

mbutrovich

Latest pass. Thanks for continuing to chip away at this, @toutane!

Second review round for apache#2671.

toutane · 2026-07-27T15:39:44Z

Hello @mbutrovich, thanks for this second pass! I addressed your comments in 1b0504d.

mbutrovich

Latest pass, thanks @toutane!

* feat(io): route FileIO storage through runtime (cherry picked from commit 331effe) * feat(catalog): propagate runtime to table FileIO (cherry picked from commit 96deb60) * feat(datafusion): propagate Iceberg runtime through providers (cherry picked from commit 481b283) * feat(runtime): add Datadog runtime integration hooks * feat(datafusion): add opt-in eager scan planning (cherry picked from commit f0b4c27) * test(datafusion): cover lazy and eager scan planning (cherry picked from commit 8ae6eb6) * fix(datafusion): address eager scan review feedback (cherry picked from commit 2380cbf) * fix(datafusion): address eager scan review feedback Second review round for apache#2671. --------- Co-authored-by: Geoffrey Claude <geoffrey.claude@datadoghq.com>

* feat(datafusion): add opt-in eager scan planning (cherry picked from commit f0b4c27) * test(datafusion): cover lazy and eager scan planning (cherry picked from commit 8ae6eb6) * fix(datafusion): address eager scan review feedback (cherry picked from commit 2380cbf) * fix(datafusion): address eager scan review feedback Second review round for apache#2671. --------- Co-authored-by: Geoffrey Claude <geoffrey.claude@datadoghq.com>

* feat(datafusion): add opt-in eager scan planning (cherry picked from commit f0b4c27) * test(datafusion): cover lazy and eager scan planning (cherry picked from commit 8ae6eb6) * fix(datafusion): address eager scan review feedback (cherry picked from commit 2380cbf) * fix(datafusion): address eager scan review feedback Second review round for apache#2671.

toutane · 2026-07-30T09:24:23Z

Thank for this pass @mbutrovich!

mbutrovich

Another pass, thanks for your patience on this PR, @toutane! I think it's getting close!

Eager scan planning built N+1 TableScans per query: one in plan_file_task_groups to call plan_files(), then one per output partition in execute() purely to read .arrow_reader_builder() off it. arrow_reader_builder() reads nothing from the PlanContext, only file_io, runtime, batch size and the row-group/row-selection flags. So the work TableScanBuilder::build() does on those per-partition calls (snapshot resolution, per-column schema validation, field id resolution, predicate binding, name-mapping JSON parse, three fresh caches) was entirely discarded. Introduce EagerScanPlan, which keeps the Arc<TableScan> that planned the tasks next to the tasks themselves, and have execute() derive its reader from it. The eager path now builds one TableScan per query. This keeps both scan paths routed through build_table_scan and TableScan::arrow_reader_builder, so their reader settings cannot drift apart. The eager reader is now derived from the very instance that planned the tasks, making that equivalence structural rather than a call convention. .build() is still called per partition, so each partition keeps its own ArrowReader and DeleteFilter as before.

mbutrovich

Thanks for staying with this through several rounds, this is close. Two small suggestions inline while we wait for a committer to review, both about test coverage rather than correctness of the feature itself.

mbutrovich · 2026-08-04T19:55:32Z

+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+fn group_file_scan_tasks_round_robin(


Every test that exercises this function uses a task count equal to (or clamped down to) the partition count, so each group ends up with exactly one task. Would it be worth adding a direct #[test] here for an uneven split, e.g. 5 tasks over 3 partitions landing as [2, 2, 1]? It's a small pure function with no I/O, so a unit test seems cheap, and right now nothing would catch a change that preserves total task/row counts but breaks the round-robin balance itself.

Addressed in f66c9b9 viatest_group_file_scan_tasks_round_robin_uneven_split

mbutrovich · 2026-08-04T19:55:59Z

+    // The physical optimizer absorbs the initial GlobalLimitExec into
+    // CoalescePartitionsExec; its fetch enforces the global bound in the final plan.
+    let global_limit_coalescer = plan
+        .downcast_ref::<CoalescePartitionsExec>()


This is a good end-to-end check that the global limit is actually enforced. One thought: it pins today's DataFusion optimizer output shape (GlobalLimitExec absorbed into CoalescePartitionsExec::fetch). If a future DataFusion bump changes how that composes, this assertion could fail with no regression in this crate's code. Might be worth a short comment here noting that a failure after a datafusion version bump likely means the plan shape changed, not that eager scanning broke, so whoever hits it later doesn't have to rediscover that.

Done in ab4e8ff

toutane · 2026-08-05T11:28:46Z

Thanks for staying with this through several rounds, this is close. Two small suggestions inline while we wait for a committer to review, both about test coverage rather than correctness of the feature itself.

Thank you for continuing to review this code, @mbutrovich.
Do you know of any committer who might be able to take this on?

timsaucer

I think you've done a great job pushing through on this topic. I know there have been many iterations and I think it's paid off because this looks like a very clean PR. I don't have committer access, so I can only provide an extra "approve" that doesn't carry weight.

The DataFusion parts specifically look very good to me. I am unable to test at scale myself, so I will have to rely on others or on the included tests.

The comments I have are all documentation, not code issues.

timsaucer · 2026-08-05T11:35:20Z

+                    // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+                    .with_data_file_concurrency_limit(1)
+                    .build()
+                    // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.


Is there an open issue for this? If so we should link here so future developers can track this TODO status.

Done: #2964

timsaucer · 2026-08-05T11:36:50Z

+    // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+    // because projection, predicate binding, snapshot schema, and delete planning can differ
+    // between scans. Catalog-backed providers also need fresh metadata on each scan.
+    // TODO: Revisit provider-level caching for static tables with a precise cache key.


Recommend opening an issue and linking here.

Done: #2963

timsaucer · 2026-08-05T11:37:17Z

+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.


Recommend opening an issue and linking here.

Done: #2962

toutane · 2026-08-05T14:54:59Z

I think you've done a great job pushing through on this topic. I know there have been many iterations and I think it's paid off because this looks like a very clean PR. I don't have committer access, so I can only provide an extra "approve" that doesn't carry weight.

The DataFusion parts specifically look very good to me. I am unable to test at scale myself, so I will have to rely on others or on the included tests.

The comments I have are all documentation, not code issues.

@timsaucer, thanks a lot for your message, really appreciate it!

For testing, we're going to shadow real traffic in our infra. It won't be a standard benchmark, but it should at least tell us whether things are improving on real-world queries.

That said, I think having a standard way to benchmark this kind of optimization would be really valuable for the iceberg-datafusion crate. Not sure if that's already been discussed somewhere?

Thanks again for the comments - I opened three issues to track the TODOs.

toutane marked this pull request as ready for review June 19, 2026 08:13

This was referenced Jun 19, 2026

Enable parallel file-level scanning for IcebergTableScan Datafusion Integration #2220

Open

feat!(datafusion): enable parallel file scanning with eager task bucketing #2298

Closed

toutane mentioned this pull request Jun 30, 2026

feat(datafusion): Declare hash partitioning for identity spec toutane/iceberg-rust#8

Draft

mbutrovich self-requested a review July 9, 2026 15:05

mbutrovich reviewed Jul 10, 2026

View reviewed changes

toutane force-pushed the datafusion-eager-file-scan-planning branch 2 times, most recently from 6a4f1a1 to be6d53e Compare July 10, 2026 13:44

toutane requested a review from mbutrovich July 15, 2026 08:28

toutane force-pushed the datafusion-eager-file-scan-planning branch from be6d53e to 528019a Compare July 17, 2026 14:28

mbutrovich reviewed Jul 17, 2026

View reviewed changes

toutane force-pushed the datafusion-eager-file-scan-planning branch from 528019a to 2380cbf Compare July 22, 2026 15:26

toutane added 4 commits July 27, 2026 17:27

feat(datafusion): add opt-in eager scan planning

0785d94

test(datafusion): cover lazy and eager scan planning

29acf0f

fix(datafusion): address eager scan review feedback

c6f7a1f

fix(datafusion): address eager scan review feedback

1b0504d

Second review round for apache#2671.

toutane force-pushed the datafusion-eager-file-scan-planning branch from 2380cbf to 1b0504d Compare July 27, 2026 15:37

toutane mentioned this pull request Jul 28, 2026

[branch-0.10] Add opt-in eager scan planning for DataFusion scans DataDog/iceberg-rust#27

Merged

mbutrovich self-requested a review July 28, 2026 16:22

mbutrovich reviewed Jul 28, 2026

View reviewed changes

Comment thread crates/integrations/datafusion/src/config.rs

Comment thread crates/integrations/datafusion/src/physical_plan/scan.rs

toutane requested a review from mbutrovich July 29, 2026 15:13

test(datafusion): cover eager scan global limits

5ae5544

toutane force-pushed the datafusion-eager-file-scan-planning branch from 5cd31e5 to 5ae5544 Compare July 30, 2026 08:18

test(datafusion): cover ordered eager scan limits

1082c50

mbutrovich reviewed Jul 31, 2026

View reviewed changes

Comment thread crates/integrations/datafusion/src/physical_plan/scan.rs Outdated

toutane added 2 commits August 3, 2026 11:24

Merge branch 'main' into datafusion-eager-file-scan-planning

cc926f3

toutane requested a review from mbutrovich August 3, 2026 13:38

Merge branch 'main' into datafusion-eager-file-scan-planning

42f9846

mbutrovich reviewed Aug 4, 2026

View reviewed changes

toutane added 2 commits August 5, 2026 13:04

docs: clarify DataFusion plan shape assertion

ab4e8ff

test: cover uneven scan task distribution

f66c9b9

toutane requested a review from mbutrovich August 5, 2026 11:29

timsaucer approved these changes Aug 5, 2026

View reviewed changes

This was referenced Aug 5, 2026

[iceberg-datafusion] Balance eager scan task groups by estimated read size #2962

Open

[iceberg-datafusion] Cache eager scan planning across equivalent scans #2963

Open

Allow ArrowReader to read shared FileScanTasks #2964

Open

docs(datafusion): link eager scan follow-up issues

4759f12

NoahKusaba mentioned this pull request Aug 5, 2026

feat: Add Iceberg Support apache/datafusion-ballista#2217

Draft

		// TODO: Replace this naive round-robin grouping with size-based grouping once the
		// first parallel scan path is stable. Keep this v1 simple and deterministic.

Uh oh!

Conversation

toutane commented Jun 18, 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?

Public API changes:

Trade-offs:

Follow-up work:

Are these changes tested?

AI Disclosure

Uh oh!

toutane commented Jun 30, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

toutane commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

toutane commented Jul 16, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

toutane commented Jul 27, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

toutane commented Jul 30, 2026

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

toutane Aug 5, 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!

Choose a reason for hiding this comment

Uh oh!

toutane commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

timsaucer left a comment

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!

toutane commented Aug 5, 2026

Uh oh!

Reviewers

toutane commented Jun 18, 2026 •

edited

Loading

toutane commented Jul 10, 2026 •

edited

Loading

toutane Aug 5, 2026 •

edited

Loading

toutane commented Aug 5, 2026 •

edited

Loading