feat: plan eager Iceberg scans with source-owned files by alexanderbianchi · Pull Request #803 · datafusion-contrib/datafusion-distributed

alexanderbianchi · 2026-10-26T18:38:48Z

Eager files belong to the scan

This draft combines #773's eager file planning with source-owned execution input, directly against main. It explores landing the intended ownership model first, without introducing eager feeds and replacing them in a later PR. Lazy planning remains the default.

Eager:
  discover files during planning → source owns descriptors
  → assign task-local partition groups → router inspects those groups
  → serialize each task's groups → worker restores groups → reader

Lazy:
  scan definition → task plan with remote feed reference
  → coordinator discovers and dispatches descriptors → reader

AssignedIcebergDataSource holds shared descriptors and its partition groups directly. The source is the authority for routing, serialization, and execution. PlannedFiles only owns descriptors and their memory reservation; there is no separate AssignedFiles object, assignment-policy enum, or routing metadata copy.

IcebergDataSource retains the lazy feed lifecycle. Both sources share scan settings and reader execution. The planning boundary selects the source; eager execution does not pass through a work-unit feed.

Assignment and routing

Like DFD's FileScanConfig path, the scale-up handler produces task-local source variants in a DistributedLeafExec before routing. Files visit tasks first, then local partitions. Coordinator variants share descriptor storage; each worker plan carries only that task's assigned descriptors.

The existing RouteTaskEvent::task_specialized_plan is sufficient. Downcast the source to AssignedIcebergDataSource and borrow planned_files() or planned_partition_tasks(partition) before choosing a worker. Iceberg continues using its own reader and FileIO, without placeholder storage URLs.

Scope and tradeoffs

Includes eager discovery, query-memory accounting, task-local assignment, routing inspection, and descriptor transport. No file-count cap.
Keeps current snapshot statistics and task-size estimates. feat: size Iceberg scans from pruned file work #787's pruned sizing, feat: balance whole-file Iceberg work across tasks #788's greedy policy, and feat: reconstruct eager Iceberg column statistics #789's selected-file statistics are outside this draft.
Eager planning waits for file discovery and increases worker-plan payloads. Lazy planning retains discovery/read overlap and execution-time feed delivery.
Task-local variants drop whole-snapshot statistics rather than claim they describe a subset.
Lazy and assigned scans have separate protocol payloads with shared scan settings. Coordinators and workers must upgrade together; see docs/upgrade/5.0.0.md.

This follows the source-owned planning direction explored in datafusion-iceberg#67 and Gabriel's data-source ownership suggestion on #797. It is an alternative foundation to #773 and the experiments in #797/#799/#801; those PRs remain unchanged.

Validation

Integration coverage includes eager/lazy result parity, task-first distribution, routing after pruning, isolated UNION branches, descriptor round trips, repeated execution, and query-memory limits. Codec tests reject missing variants and malformed groups/descriptors.

Passed: full Iceberg integration suite (58 tests and one doctest), all-target Clippy with warnings denied, formatting, and documentation build.

alexanderbianchi · 2026-10-26T19:22:44Z

Hm this feels like it shouldn't be in the harness, should be like a trait on the data source maybe. ideally without messing up the code.

alexanderbianchi · 2026-10-26T19:23:12Z

            proto_converter,
        )?
        .ok_or_else(|| proto_error("IcebergDataSource is missing its partitioning"))?;
+        let input = match (proto.feed, proto.assigned_files) {


The fact we need this complex decoding is a red flag.

Maybe we could have 2 different protos at the top level with their own path to the same data source and share common componends.

alexanderbianchi · 2026-10-26T19:25:26Z

-///
-/// This distributed mechanism is transparent to this [DataSource].
+/// Reads Iceberg file tasks through one reader. Eager scans own an immutable
+/// assignment exposed by [`Self::planned_partition_tasks`]; lazy scans use a work-unit feed.


only used in tests? should be marked integration maybe? comment should specify that.

alexanderbianchi · 2026-10-26T19:25:51Z

+use crate::{IcebergConfig, IcebergDataSource};
+
+/// Materialize eager assignments after the planner has chosen the final task count.
+pub(crate) fn iceberg_scale_up(


Matches the file scan config path? we should explicitlymention that in the comment.

alexanderbianchi · 2026-10-26T19:26:37Z

+    /// Returns `None` for lazy scans and `Some` with an empty iterator for empty
+    /// eager scans. This borrows the execution assignment without I/O or cloning
+    /// descriptors; iteration order does not promise row ordering.
+    pub fn planned_files(&self) -> Option<impl Iterator<Item = &FileScanTask>> {


also only used in tests?

alexanderbianchi · 2026-10-26T19:27:08Z

        for task in 0..task_count {
            loads.push(Reverse((0_u128, 0_usize, task)));
-            if task % 256 == 0 {
-                tokio::task::yield_now().await;


why removing these?

alexanderbianchi · 2026-10-26T19:27:26Z

@@ -122,10 +123,11 @@ fn set_iceberg_integration(state: &mut SessionState, options: IcebergIntegration
        )),
    );
    state.set_distributed_desired_task_count_handler(iceberg_desired_task_count);


Hod does this differ from below?

alexanderbianchi commented Oct 26, 2026

View reviewed changes

feat: plan eager Iceberg scans with source-owned files

093942f

alexanderbianchi changed the title ~~feat: carry eager Iceberg assignments in task scans~~ feat: plan eager Iceberg scans with source-owned files Oct 26, 2026

alexanderbianchi force-pushed the codex/iceberg-assigned-scans branch from a07bff5 to 093942f Compare October 26, 2026 20:21

alexanderbianchi changed the base branch from alex.bianchi/iceberg-selected-column-bytes to main October 26, 2026 20:22

Navigation Menu

Repository navigation

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

feat: plan eager Iceberg scans with source-owned files#803
alexanderbianchi wants to merge 1 commit into
mainfrom
codex/iceberg-assigned-scans

alexanderbianchi commented Oct 26, 2026 •

edited

Loading

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

alexanderbianchi Oct 26, 2026

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

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

Uh oh!

Eager files belong to the scan

Assignment and routing

Scope and tradeoffs

Validation

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!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

alexanderbianchi commented Oct 26, 2026 •

edited

Loading