gabotechs commented on code in PR #67:
URL: https://github.com/apache/datafusion-iceberg/pull/67#discussion_r4374639082
##########
crates/datafusion/src/test_utils/harness.rs:
##########
@@ -53,22 +54,46 @@ pub struct IcebergTestHarness {
impl IcebergTestHarness {
pub async fn new() -> Result<Self> {
- let state = SessionStateBuilder::new()
+
Self::with_config(SessionConfig::new().with_target_partitions(16)).await
+ }
+
+ pub async fn with_config(config: SessionConfig) -> Result<Self> {
Review Comment:
Rather than adding a bunch of `with_x`, `with_x_and_y`, `with_x_and_y_and_z`
constructor methods, a more scalable approach could be what you did here:
https://github.com/datafusion-contrib/datafusion-distributed/blob/2783b8f59b9821a53cd222950c299cf7a4ffc07c/iceberg/src/test_utils/harness.rs#L49-L49
Where a builder pattern is used for constructing different permutations of
the `IcebergTestHarness`
##########
crates/datafusion/src/read/scan.rs:
##########
@@ -231,11 +392,19 @@ impl DataSource for IcebergDataSource {
if let Some(fetch) = self.fetch {
write!(f, ", fetch={fetch}")?;
}
+ if let Some(planned) = self.planned_files() {
+ write!(f, ", planned_files={}", planned.len())?;
+ }
Ok(())
}
fn output_partitioning(&self) -> Partitioning {
- Partitioning::UnknownPartitioning(1)
+ // Partitioned lazy discovery:
https://github.com/apache/datafusion-iceberg/issues/84
Review Comment:
I'd add a `TODO` prefix here, just to mark this as something we want to do.
##########
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.
Review Comment:
🤔 I don't think I know what "Iceberg's planning order" or "Assign each once
per execution" mean.
##########
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.
Review Comment:
I wouldn’t mention DataFusion Distributed here, to keep this project neutral
with respect to its consumers.
##########
crates/datafusion/src/read/scan.rs:
##########
@@ -145,28 +221,88 @@ impl IcebergDataSourceBuilder {
self
}
- /// Builds the source, which reads the columns of its projected schema by
name.
+ /// Select files during [`Self::build`] when enabled. Defaults to `false`.
+ pub fn with_eager_planning(mut self, enabled: bool) -> Self {
+ self.eager_planning = enabled;
+ self
+ }
+
+ /// Builds the source and its shared reader. When eager planning is
enabled,
+ /// discovers files using `context` for partition count and memory
accounting.
+ /// Lazy builds perform no file-discovery I/O.
///
/// # Errors
///
- /// Returns an error if the projection is out of bounds for the schema.
- pub fn build(self) -> Result<IcebergDataSource> {
- let schema = match &self.projection {
- None => self.schema,
- Some(projection) => Arc::new(self.schema.project(projection)?),
+ /// Returns an error if the projection is invalid, file planning fails, or
+ /// retained tasks exceed the query memory pool.
+ pub async fn build(self, context: Arc<TaskContext>) ->
Result<IcebergDataSource> {
Review Comment:
Actually, I'm looking at how this is used, and the current state feels
actually more ergonomic, so +1 if you just want to keep it like it is. I would
probably name the method `try_build()` though, as builder patterns typically
return infallible results in the `build()` method.
##########
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";
+}
Review Comment:
I think it's a good time to adopt the config structure proposed here:
https://github.com/apache/datafusion-iceberg/pull/55
This was a suggestion given by @mbutrovich in
https://github.com/apache/datafusion-iceberg/pull/55#discussion_r4298481201 and
I think it was on point, so maybe we can ship it here.
##########
crates/datafusion/src/read/scan.rs:
##########
@@ -145,28 +221,88 @@ impl IcebergDataSourceBuilder {
self
}
- /// Builds the source, which reads the columns of its projected schema by
name.
+ /// Select files during [`Self::build`] when enabled. Defaults to `false`.
+ pub fn with_eager_planning(mut self, enabled: bool) -> Self {
+ self.eager_planning = enabled;
+ self
+ }
+
+ /// Builds the source and its shared reader. When eager planning is
enabled,
+ /// discovers files using `context` for partition count and memory
accounting.
+ /// Lazy builds perform no file-discovery I/O.
///
/// # Errors
///
- /// Returns an error if the projection is out of bounds for the schema.
- pub fn build(self) -> Result<IcebergDataSource> {
- let schema = match &self.projection {
- None => self.schema,
- Some(projection) => Arc::new(self.schema.project(projection)?),
+ /// Returns an error if the projection is invalid, file planning fails, or
+ /// retained tasks exceed the query memory pool.
+ pub async fn build(self, context: Arc<TaskContext>) ->
Result<IcebergDataSource> {
Review Comment:
An alternative API design that comes to mind is having two separate methods
for the eager/lazy file task discovery:
```rust
pub fn build_lazy(self) -> Result<IcebergDataSource>;
```
And
```rust
pub async fn build_eager(self, context: Arc<TaskContext>) ->
Result<IcebergDataSource> {
```
So that people just wanting to build the "lazy" version do not pay the
ergonomics price of passing a `TaskContext` and having the method be `async`.
Also, prefixing the method with `try_*` might be a good idea as it's
fallible, following rust conventions.
These are just suggestions though, I think the API is already in an
acceptable state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
