feat: embed eager Iceberg assignments in scan inputs

docs/upgrade/5.0.0.md

@@ -111,3 +111,14 @@ Some(Ok(DesiredTaskCountEventResponse::desired(0.25)))
 or `StorageFactory` to this crate, update those dependencies to the same revision.
 Alternatively, use `datafusion_distributed_iceberg::iceberg` for the Iceberg
 types you pass to its public APIs.
+
+## 5. Iceberg scan inputs
+
+`IcebergDataSource::feed()` now returns `Option<&WorkUnitFeed<IcebergWorkUnitFeed>>`:
+only lazy scans have a feed. Use `input()` to match `ScanInput`, or
+`assigned_files()` to inspect eager files by local partition. Custom work-unit
+feed registrations should use `and_then(IcebergDataSource::feed)` instead of `map`.
+
+Eager task plans now contain assigned file descriptors. Upgrade coordinators and
+workers together before enabling eager planning; older workers do not decode
+this new input. Lazy feed plans retain their existing representation.

iceberg/README.md

@@ -35,22 +35,26 @@ while building each physical scan, then deliver those same tasks to the reader a
 execution. Retained metadata is charged to the query memory pool, without a
 file-count limit. Eager planning uses metadata I/O up front. Scan statistics and
 desired distributed task counts use selected file work instead of snapshot totals;
-reader settings and file assignment are unchanged. Both modes use the same dispatcher
-and partition channels. Eager execution explicitly copies retained file descriptors
-into owned work units; lazy execution moves newly discovered descriptors into them.
+one Iceberg reader handles both inputs.
 
 ```text
-Planning                          Execution
-Eager: discover → retain files ──→ copy descriptors ─┐
-Lazy:  retain scan definition ───→ discover files ──┤
-                                                  ▼
-                                   shared task-first dispatcher
-                                                  ↓
-                                         partition channels
-                                                  ↓
-                                      local or remote readers
+Eager: discover → assign files → task scan with local file groups → reader
+Lazy:  retain scan definition → discover and dispatch via feed   → reader
 ```
 
+Eager scans use `ScanInput::AssignedFiles`; lazy scans use
+`ScanInput::WorkUnitFeed`. Once DFD chooses the distributed task count, the
+scale-up handler creates task-specific eager scans with concrete file groups.
+The existing routing event can inspect `IcebergDataSource::assigned_files()`
+in its `task_specialized_plan`, without accessing a coordinator-side feed.
+
+The codec sends only the assigned groups for that task, preserving local
+partition membership. This increases plan payload size and shifts descriptor
+transport from execution-time feed messages to worker setup. Greedy assignment
+runs synchronously in the existing scale-up hook; this draft does not claim a
+performance improvement. Reader execution clones descriptors as needed, while
+coordinator-side assignments share the retained descriptors and memory reservation.
+
 Configure the session before creating the `SessionState`:
 
 ```rust

iceberg/src/codec.rs

@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
+use crate::{AssignedFiles, ScanInput};
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::common::{Result, internal_err};
+use datafusion::common::{Result, exec_datafusion_err, internal_err};
 use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::ExecutionPlan;
@@ -69,11 +70,6 @@ impl PhysicalExtensionCodec for IcebergCodec {
             .schema
             .ok_or_else(|| proto_error("IcebergDataSource is missing its schema"))?;
         let schema = SchemaRef::new((&schema).try_into()?);
-        let feed = proto
-            .feed
-            .ok_or_else(|| proto_error("IcebergDataSource is missing its work-unit feed"))?;
-        let feed =
-            WorkUnitFeed::<IcebergWorkUnitFeed>::from_proto(WorkUnitFeedProto { id: feed.id })?;
         let decode_ctx = PhysicalPlanDecodeContext::new(ctx, self);
         let partitioning = parse_protobuf_partitioning(
             proto.partitioning.as_ref(),
@@ -82,6 +78,45 @@ impl PhysicalExtensionCodec for IcebergCodec {
             proto_converter,
         )?
         .ok_or_else(|| proto_error("IcebergDataSource is missing its partitioning"))?;
+        let input = match (proto.feed, proto.assigned_files) {
+            (Some(feed), None) => {
+                ScanInput::WorkUnitFeed(Box::new(WorkUnitFeed::<IcebergWorkUnitFeed>::from_proto(
+                    WorkUnitFeedProto { id: feed.id },
+                )?))
+            }
+            (None, Some(files)) => {
+                if files.partitions.len() != partitioning.partition_count()
+                    || files.partitions.is_empty()
+                {
+                    return internal_err!("Iceberg assigned files do not match scan partitioning");
+                }
+                let groups = files
+                    .partitions
+                    .into_iter()
+                    .map(|group| {
+                        group
+                            .tasks
+                            .into_iter()
+                            .map(|bytes| {
+                                rmp_serde::from_slice(&bytes).map_err(|error| {
+                                    exec_datafusion_err!("Invalid assigned Iceberg file: {error}")
+                                })
+                            })
+                            .collect::<Result<Vec<_>>>()
+                    })
+                    .collect::<Result<Vec<_>>>()?;
+                ScanInput::AssignedFiles(AssignedFiles::from_partitions(
+                    groups,
+                    ctx,
+                    files.may_filter_rows,
+                )?)
+            }
+            _ => {
+                return internal_err!(
+                    "Iceberg scan requires exactly one input: assigned files or feed"
+                );
+            }
+        };
         let fetch = proto
             .fetch
             .map(usize::try_from)
@@ -100,7 +135,7 @@ impl PhysicalExtensionCodec for IcebergCodec {
             table_snapshot: None,
             iceberg_file_io,
             iceberg_runtime: self.iceberg_runtime.clone(),
-            feed,
+            input,
         }))
     }
 
@@ -120,12 +155,44 @@ impl PhysicalExtensionCodec for IcebergCodec {
             return internal_err!("expected DataSourceExec wrapping IcebergDataSource");
         };
 
-        let feed = source.feed.to_proto();
+        let (feed, assigned_files) = match source.input() {
+            ScanInput::WorkUnitFeed(feed) => (
+                Some(pb::WorkUnitFeed {
+                    id: feed.to_proto().id,
+                }),
+                None,
+            ),
+            ScanInput::AssignedFiles(files) => {
+                let partitions = (0..files.partition_count())
+                    .map(|partition| {
+                        let tasks = files
+                            .partition(partition)?
+                            .map(|file| {
+                                rmp_serde::to_vec_named(file).map_err(|error| {
+                                    exec_datafusion_err!(
+                                        "Cannot encode assigned Iceberg file: {error}"
+                                    )
+                                })
+                            })
+                            .collect::<Result<Vec<_>>>()?;
+                        Ok(pb::FileGroup { tasks })
+                    })
+                    .collect::<Result<Vec<_>>>()?;
+                (
+                    None,
+                    Some(pb::AssignedFiles {
+                        partitions,
+                        may_filter_rows: files.planned.may_filter_rows(),
+                    }),
+                )
+            }
+        };
         let proto = pb::IcebergDataSource {
             schema: Some(datafusion_proto::protobuf::Schema::try_from(
                 source.schema.as_ref(),
             )?),
-            feed: Some(pb::WorkUnitFeed { id: feed.id }),
+            feed,
+            assigned_files,
             partitioning: Some(serialize_partitioning(
                 &source.partitioning,
                 self,
@@ -168,7 +235,10 @@ mod tests {
             decoded.partitioning.to_string()
         );
         assert_eq!(source.fetch, decoded.fetch);
-        assert_eq!(source.feed.to_proto(), decoded.feed.to_proto());
+        assert_eq!(
+            source.feed().unwrap().to_proto(),
+            decoded.feed().unwrap().to_proto()
+        );
         assert_eq!(
             source.iceberg_file_io.config().props(),
             decoded.iceberg_file_io.config().props()