[alexanderbianchi]

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 FileScanTasks 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.

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. 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, this retains native FileScanTasks 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. 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, then readable routing tests. 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 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.

[gabotechs]

👍 The approach looks good, my suggestions are mostly around organization, testing and some questions.

Glad to see you've found a good way of representing the eager path with existing tooling!

[gabotechs]

We typically avoid to commit anything that is not pure use plumbing to mod.rs files.

Even if it's trivial, I'd just place this in a separate file for consistency.

[gabotechs]

Suggested change

	Some(projection) => Arc::new(schema.project(projection).unwrap()),
	Some(projection) => Arc::new(schema.project(projection)?),

[gabotechs]

🤔 I'm thinking about the naming, what about renaming them:

• EagerIcebergDataSource
• LazyIcebergDataSource
  ?

Those are the terms we use typically while talking about it.

[gabotechs]

I wonder how this would look like once apache/datafusion-iceberg#67 lands.

I imagine we no longer want to maintain here any extra param determining whether we want the discovery to be lazy or not, we'd just follow whatever we decide in apache/datafusion-iceberg#67 instead.

Should we be shipping apache/datafusion-iceberg#67 first? or do you think it does not matter much.

[gabotechs]

If we've done eager file planning, the stats should not be coming from the table snapshot no? I imagine they'll not take pruning into account.

Is this what you had in mind for #787? probably worth a TODO comment linking the other PR if that's the case.

[gabotechs]

The ok_or_some_err!() macro can help you reduce some boilerplate here.

[gabotechs]

These tests are a bit hard to read. Integration tests serve also as live documentation, humans should be able to take a quick glance at a test and immediately know what's happening.

Typically we heavy really on insta and plan snapshots for this: they do not only capture behavior, they are also very easy to review by humans.

I remember giving similar feedback when the task routing API was introduced, and this was the result:

https://github.com/datafusion-contrib/datafusion-distributed/blob/14beca0048b939366e2583522402b604285c6585/tests/task_estimator_test.rs

[gabotechs]

Why this proto roundtrip here? this test overall look very far from how someone would use the project in their codebase.

The challenge with having so much synthetic code during integration testing is that they stop resembling real scenarios that people would typically find in their codebases, and a good part of the coverage does not go to testing real scenarios, it goes to testing the synthetic test preparation code instead.

Same as above, if you find ways of expressing these test with something closer how people would use the project, that'd be great.

[gabotechs]

🤔 is this something we want to assert? I'd expect them to not return the same stats, as eager file planning stats should be better.

[gabotechs]

Part of the charm of the IcebergTestHarness is that you should not need helpers for building the harness, and the harness configuration is explicit during the build process.

For example, here, I'm unable to understand what this test is doing without navigating to the harness() function and reading it first.

For example, with something like this:

        let eager = IcebergTestHarness::builder()
            .with_workers(2)
            .with_column_stats_enabled()
            .with_eager_file_planning()
            .build()
            .await?;

        let lazy = IcebergTestHarness::builder()
            .with_workers(2)
            .with_column_stats_enabled()
            .build()
            .await?;

It's very easy for a human to know what's the testing setup without doing any code jump. It's more verbose, but the verbosity here plays in the reviewers favor, as the cognitive price it takes to read that is less than if they need to jump through the code understanding what the helper function does.

[gabotechs]

I see most of my comments have been address or are on the right track, so +1 in this now. Pretty epic work here 👍

[gabotechs]

It should be fine to just break the API here, and don't think we need to alias this.