feat: plan Iceberg file tasks before execution

crates/datafusion/README.md

@@ -20,3 +20,40 @@
 # Apache Iceberg DataFusion Integration
 
 This crate contains the integration of Apache DataFusion and Apache Iceberg.
+
+## Optional file planning before execution
+
+By default, the Iceberg reader discovers files when execution starts. To select
+`FileScanTask`s during `TableProvider::scan()` instead, register the planning
+options on the DataFusion session before planning a query:
+
+```rust
+use datafusion::prelude::SessionConfig;
+use iceberg_datafusion::IcebergPlanningConfig;
+
+let config = SessionConfig::new()
+    .with_option_extension(IcebergPlanningConfig::default())
+    .set_bool("iceberg.eager_file_planning", true);
+```
+
+### Behavior
+
+- Lazy planning is the default. Eager planning selects files before execution;
+  it reads metadata, not Parquet data.
+- Selected tasks are assigned round-robin to at most `target_partitions`.
+  An empty scan has one empty partition. Each eager partition reads one file
+  at a time when there are multiple output partitions.
+- Retained tasks have an estimated reservation in the query memory pool.
+  Exceeding the pool fails planning; there is no file-count cap or truncation.
+- Both modes use the same reader implementation. Reader clones share the
+  delete loader across partitions. Scan statistics remain unknown.
+
+### Trade-offs
+
+Eager planning completes file discovery and retains tasks even for queries
+that stop early. Repeated provider scans repeat planning. The planning caller
+must support metadata I/O: Iceberg's initial manifest-list read still runs on
+that runtime, even when the table has a separate I/O runtime.
+
+Register only one configuration extension under the `iceberg` prefix per session.
+DataFusion Distributed currently uses that same prefix for its own options.

crates/datafusion/src/config.rs

@@ -0,0 +1,31 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::extensions_options;
+use datafusion::config::ConfigExtension;
+
+extensions_options! {
+    /// Opt-in file planning during physical planning for local Iceberg scans.
+    pub struct IcebergPlanningConfig {
+        /// Discover and retain file tasks in TableProvider::scan instead of at execution.
+        pub eager_file_planning: bool, default = false
+    }
+}
+
+impl ConfigExtension for IcebergPlanningConfig {
+    const PREFIX: &'static str = "iceberg";
+}

crates/datafusion/src/lib.rs

@@ -18,6 +18,9 @@
 mod catalog;
 pub use catalog::*;
 
+mod config;
+pub use config::IcebergPlanningConfig;
+
 mod error;
 pub use error::*;
 

crates/datafusion/src/read/file_planning.rs

@@ -0,0 +1,81 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Query-local file work selected before execution, independently of task assignment.
+
+use std::mem::{size_of, size_of_val};
+
+use datafusion::common::Result;
+use datafusion::execution::TaskContext;
+use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use futures::TryStreamExt;
+use iceberg::scan::{FileScanTask, TableScan};
+
+use crate::to_datafusion_error;
+
+/// File work selected by one Iceberg scan before assignment to output partitions.
+///
+/// The tasks retain their bound predicates, projection, and delete references. Keeping
+/// this value alive also keeps the query memory reservation for the retained tasks.
+/// This is not a reader and contains no coordinator-local reader state.
+#[derive(Debug)]
+pub(crate) struct PlannedFileScan {
+    tasks: Vec<FileScanTask>,
+    _reservation: MemoryReservation,
+}
+
+impl PlannedFileScan {
+    /// Tasks in Iceberg's planning order. Assign each once per execution.
+    pub(crate) fn tasks(&self) -> &[FileScanTask] {
+        &self.tasks
+    }
+
+    pub(crate) async fn collect(scan: TableScan, context: &TaskContext) -> Result<Self> {
+        let reservation =
+            MemoryConsumer::new("Iceberg file planning").register(context.memory_pool());
+        let mut planned = Self {
+            tasks: Vec::new(),
+            _reservation: reservation,
+        };
+        let mut stream = scan.plan_files().await.map_err(to_datafusion_error)?;
+        while let Some(task) = stream.try_next().await.map_err(to_datafusion_error)? {
+            // Allow for spare Vec capacity and allocator overhead; schemas and specs
+            // are shared Arcs, but paths and delete references are owned by each task.
+            planned
+                ._reservation
+                .try_grow(retained_size(&task).saturating_mul(2))?;
+            planned.tasks.push(task);
+        }
+        Ok(planned)
+    }
+}
+
+fn retained_size(task: &FileScanTask) -> usize {
+    let mut bytes = size_of::<FileScanTask>()
+        + task.data_file_path().len()
+        + size_of_val(task.project_field_ids())
+        + size_of_val(task.deletes())
+        + task.key_metadata().map_or(0, <[u8]>::len);
+    for delete in task.deletes() {
+        bytes = bytes
+            .saturating_add(delete.file_path().len())
+            .saturating_add(delete.referenced_data_file().map_or(0, str::len))
+            .saturating_add(delete.equality_ids().map_or(0, size_of_val))
+            .saturating_add(delete.key_metadata().map_or(0, <[u8]>::len));
+    }
+    bytes
+}

crates/datafusion/src/read/mod.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+mod file_planning;
 pub(crate) mod predicate;
 pub(crate) mod scan;