feat: size Iceberg scans from pruned file work by alexanderbianchi · Pull Request #787 · datafusion-contrib/datafusion-distributed

alexanderbianchi · 2026-10-20T14:05:33Z

What changes

Size eagerly planned Iceberg scans from the selected file work retained by #803. Row counts and file bytes come from FileScanTasks rather than whole-snapshot totals, and desired task counts are bounded by the number of selected files.

Empty selections report exact zero rows. Predicates or delete files make row estimates inexact; unknown file row counts remain unknown. Whole-snapshot column statistics are skipped for these filtered scans, including the metadata I/O that would compute unused statistics. Lazy scans retain their existing snapshot-based behavior.

The assigned source owns the resulting statistics. Task-local variants and decoded worker plans continue to report unknown statistics; this does not add selected-file column-statistics reconstruction, assignment policy, or routing changes.

Stack and review

Based on #803. Review the base-to-head diff: selected-file statistics, task sizing, and their integration tests. #788 builds on this for byte-balanced assignment.

The earlier implementation based on #773 is preserved at aa21dbe, retained by the archive/pr-787-on-773 tag. That commit and its ancestry remain available if we choose to continue with #773 instead.

Validation

Integration coverage checks selected-file totals when snapshot summaries are absent, inexact filtered statistics, pruning to one file, bounded task counts, and avoiding unnecessary column-statistics reads. Existing routing, result-parity, and distribution tests are retained. No performance claim.

The previous full suite passed (61 tests and one doctest). After the module/test cleanup rebase, routing, task-count, and statistics tests passed, plus all-target Clippy with warnings denied, formatting, and git diff --check. Routing snapshots were reviewed and rerun with updates disabled.

gabotechs

👍 nice!

gabotechs · 2026-10-22T09:44:33Z

+        .map(|count| {
+            let count = node
+                .planned_file_count()
+                .map_or(count, |files| count.min(files as f64));


🤔 So we cannot byte-range split files here no?

It'd be great if somehow at some point we could use native DataFusion parquet reads for Iceberg.

gabotechs · 2026-10-22T09:49:11Z

+            .configure_session(|mut state| {
+                let mut config = IcebergConfig::default();
+                config.eager_file_planning = true;
+                state
+                    .config()
+                    .get_or_insert_default()
+                    .options_mut()
+                    .extensions
+                    .insert(config);
+                Ok(state)
+            })?
+            .build()


This occupies a fair amount of real-state, but the only important bit for a test reviewer is:

config.eager_file_planning = true;

Maybe we can add one more helper that says:

.iceberg_config_mut(|&mut iceberg_cfg| { ... })

Or something like:

.with_eager_file_planning()

## What changes Add opt-in eager Iceberg file discovery with `iceberg.eager_file_planning=true`. Physical planning reads manifests, applies Iceberg's pruning and delete matching, and retains the selected `FileScanTask`s on an `EagerIcebergDataSource`. Distributed planning assigns those files to task-local partitions before worker placement. Each worker receives only its assigned descriptors in its execution plan and reads them through the existing Iceberg reader. Lazy discovery remains the default. Both modes use `DataSourceExec`, shared reader configuration, and Iceberg's `FileIO`; eager planning does not read Parquet data. ### Lazy Planning uses snapshot statistics. During execution, the coordinator reads manifests and progressively supplies file tasks to workers through a remote work-unit feed. This overlaps discovery with reading and can avoid some discovery when execution stops early. Initial planning and worker placement happen without the complete selected-file set. ### Eager Planning completes discovery and retains the selected tasks before execution starts. The source owns the assignment that routing inspects and workers execute. This adds startup latency, retained metadata, and larger worker-plan payloads. In exchange, selected files are available for placement and future task-sizing, balancing, and statistics improvements. This matters especially when partition pruning reduces a large table to a handful of files. Fully consumed scans generally require the same discovery I/O in either mode. ```text Lazy: plan → remote feed reference → coordinator discovers and dispatches → reader Eager: discover → source owns files → assign task-local groups → route → serialize each task's groups → worker restores groups → reader ``` ## Why replace #773 #773 moves discovery earlier but retains the files behind the work-unit feed. Before routing, DFD converts native feeds into remote references, hiding the retained descriptors from the router. Keeping that design would require another way to preserve access to the files across the conversion. This PR instead lands source-owned eager execution directly against `main`, following [Gabriel's suggestion to retain routing-relevant information on the data source](#797 (comment)). It replaces #773 as the proposed foundation rather than merging eager feeds and immediately refactoring them. - **One authoritative assignment:** task-local sources share descriptor storage on the coordinator. Routing borrows the source's assigned files; serialization sends those same groups; workers restore them without reassignment. There is no separate routing metadata copy. - **Existing routing API:** like DFD's `FileScanConfig` path, the scale-up handler creates task-specific `DistributedLeafExec` variants. `RouteTaskEvent::task_specialized_plan` exposes the selected variant before placement. No second plan view or work-unit-feed extension is needed. - **Shared reading, separate delivery:** lazy scans keep their feed lifecycle; eager scans carry assigned files. Both produce file-task streams for the same reader implementation. Independent codecs preserve the existing lazy payload. Coordinator and worker codec registrations must match. - **Iceberg-native I/O:** the structure resembles file scans without forcing Iceberg into `FileScanConfig` or supplying a placeholder object-store URL. - **Bounded retention:** descriptors are charged to the query memory pool, with no arbitrary file-count cap. Abandoned eager planning cancels its discovery future. Lazy remains available for progressive startup and early termination. Keeping it means maintaining two discovery/delivery lifecycles; this PR isolates that difference rather than making eager execution depend on a feed. ## Relationship to upstream datafusion-iceberg #67 Like [#67](apache/datafusion-iceberg#67), this retains native `FileScanTask`s with a private memory-reservation owner, exposes borrowed accessors on the source, and feeds local partition streams into the Iceberg reader. Moving ownership out of DFD's feed brings the eager representation closer to upstream. DFD adds distributed assignment, placement, and reconstruction of assigned groups from worker plans. The implementations are not interchangeable yet: #67 uses one source with optional eager planning and local lazy discovery; this PR uses separate eager/lazy source types and retains DFD's remote feeds for lazy scans. Upstream would need to support reconstructing an already assigned scan without rediscovery or repartitioning before it could replace this implementation directly. ## Source and scheduling boundary The eager source represents local file groups; it does not choose distributed task assignments. `with_file_groups` derives a source using positions in `planned_files()` (retained-file order), preserving supplied group order and sharing descriptors and their memory reservation. It rejects invalid or repeated positions and resets metrics/statistics for the resulting scan. The scale-up handler chooses task assignments through this public API. The codec uses public schema, file, and partition accessors and reconstructs received native descriptors with `from_file_groups`, without rediscovery. `PlannedFiles` remains private. Reader configuration and shared delete-loader behavior still need reconciliation with upstream #67; this change separates scheduling without changing reader behavior. ## Scope and follow-ups This includes eager discovery, descriptor retention, task-first assignment, routing access, and transport. It preserves current snapshot statistics and task-size estimates; task-local eager variants and decoded plans report unknown statistics instead of applying whole-snapshot aggregates to a subset. Selected-file sizing (#787), byte balancing (#788), and selected-file statistics (#789) remain separate. Affinity policy can use the existing routing hook outside this crate. Better selected-file metrics still need iceberg-rust support to avoid rereading manifests. This PR makes no performance claim. ## How to review Start with [the eager-planning feature commit](d6c8b6a). The first four commits extract existing statistics, reader configuration, lazy scan construction, and task-first partition calculation. The next two commits address review feedback separately: [module moves and source names](bd68327), then [readable routing tests](bad9344). In the final tree, follow `table_provider/scan.rs` → `eager/data_source.rs` → the scale-up handler → `eager/codec.rs`. Lazy sources and codecs live under `lazy/`, with shared reading in `common/scan_config.rs`. The old `IcebergDataSource` and `IcebergCodec` exports are removed; consumers must use `LazyIcebergDataSource` and `LazyIcebergCodec`, as documented in the upgrade guide. The latest [source/scheduling separation commit](96f55ed) adds these public construction/regrouping operations and moves task assignment into the scale-up handler. ## Validation Coverage includes eager/lazy result parity, per-task distribution snapshots, placement after pruning, isolated UNION branches, repeated scan execution, and query-memory limits. Routing snapshots show the assigned files, selected worker, and query results; normal distributed execution exercises transport. Codec tests reject missing configuration and malformed groups/descriptors. The full Iceberg suite previously passed (61 tests and one doctest). After the grouping-test cleanup, all library tests, grouping tests, and routing tests passed with snapshot updates disabled. All-target Clippy with warnings denied and formatting also passed. Public grouping tests snapshot partition groups and actual dates/counts read after regrouping and reconstruction. Invalid cases are labeled explicitly; reservation lifetime is tested separately. The codec validation test uses `encode_config` through public source accessors without a synthetic plan roundtrip. Existing routing/distribution snapshots remain unchanged.

This was referenced Oct 20, 2026

feat: balance whole-file Iceberg work across tasks #788

Open

feat: reconstruct eager Iceberg column statistics #789

Draft

feat: route Iceberg tasks by file affinity with placement metrics #790

Closed

alexanderbianchi added this pull request to stack #791 October 20, 2026 14:58

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from f75ccd2 to c5f96f5 Compare October 20, 2026 23:22

alexanderbianchi removed this pull request from stack #791 October 21, 2026 00:17

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from c5f96f5 to cbefc78 Compare October 21, 2026 00:17

alexanderbianchi added this pull request to stack #794 October 21, 2026 00:18

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from cbefc78 to 09db90e Compare October 21, 2026 00:20

alexanderbianchi marked this pull request as ready for review October 21, 2026 05:08

alexanderbianchi mentioned this pull request Oct 21, 2026

Preserve column statistics on the eager Iceberg path #795

Open

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from 6f68dad to 3c25b55 Compare October 21, 2026 05:28

gabotechs force-pushed the alex.bianchi/iceberg-pruned-task-count branch from 3c25b55 to aa21dbe Compare October 21, 2026 10:53

alexanderbianchi mentioned this pull request Oct 22, 2026

feat: plan Iceberg file tasks before execution apache/datafusion-iceberg#67

Open

gabotechs approved these changes Oct 22, 2026

View reviewed changes

alexanderbianchi mentioned this pull request Oct 26, 2026

feat: plan eager Iceberg scans with source-owned files #803

Merged

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from aa21dbe to d9ad48f Compare October 26, 2026 23:52

alexanderbianchi removed this pull request from stack #794 October 26, 2026 23:54

alexanderbianchi changed the base branch from feat/iceberg-eager-file-planning to codex/iceberg-assigned-scans October 26, 2026 23:54

alexanderbianchi added this pull request to stack #804 October 26, 2026 23:54

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch 3 times, most recently from 1b68bd3 to a78bc2d Compare October 27, 2026 16:19

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from a78bc2d to c6b5ecf Compare October 27, 2026 16:31

Base automatically changed from codex/iceberg-assigned-scans to main October 27, 2026 19:06

feat: size assigned Iceberg scans from selected file work

57606f4

alexanderbianchi force-pushed the alex.bianchi/iceberg-pruned-task-count branch from c6b5ecf to 57606f4 Compare October 27, 2026 19:06

alexanderbianchi merged commit f108d55 into main Oct 27, 2026
38 checks passed

alexanderbianchi deleted the alex.bianchi/iceberg-pruned-task-count branch October 27, 2026 19:27

Navigation Menu

Repository navigation

feat: size Iceberg scans from pruned file work - #787

feat: size Iceberg scans from pruned file work#787
alexanderbianchi merged 1 commit into
mainfrom
alex.bianchi/iceberg-pruned-task-count

alexanderbianchi commented Oct 20, 2026 •

edited

Loading

Uh oh!

gabotechs left a comment

Uh oh!

gabotechs Oct 22, 2026

Uh oh!

gabotechs Oct 22, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

alexanderbianchi commented Oct 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes

Stack and review

Validation

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

gabotechs Oct 22, 2026

Choose a reason for hiding this comment

Uh oh!

gabotechs Oct 22, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

alexanderbianchi commented Oct 20, 2026 •

edited

Loading