You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "toutane (via GitHub)" <gi...@apache.org> on 2026/06/18 16:24:41 UTC
[PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
toutane opened a new pull request, #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671
## Which issue does this PR close?
- Closes #2220
- Related to
- #1604
- #128
- #2298
## What changes are included in this PR?
This PR adds an opt-in eager scan planning path for the DataFusion integration.
When `iceberg.enable_eager_scan_planning` is enabled, `IcebergTableProvider::scan()` plans `FileScanTask`s during physical planning, groups them across DataFusion `target_partitions`, and exposes the resulting output partition count as `UnknownPartitioning(N)`. Each `execute(partition)` call then reads only the task group assigned to that output partition through `ArrowReaderBuilder`.
The default behavior is unchanged: eager planning is disabled by default, so scans keep the existing lazy single-partition planning path unless explicitly enabled.
This is a narrower split from #2298, focused on file-level eager planning and output partition count reporting. Some of the broader design discussions and review feedback happened in #2298; this PR keeps only the first scoped step and intentionally does not include hash partitioning, size-aware bin-packing, or row-group/sub-file planning.
#### Trade-offs:
- Eager planning does catalog/metadata work during `TableProvider::scan()`, so it is kept opt-in
- Task grouping is round-robin and count-based, not size-aware. This is simple and deterministic, but can be imbalanced when file sizes vary
- Parallelism is at the `FileScanTask` level only. A table with one large file will not benefit from this change
- The scan reports `UnknownPartitioning(N)`, not hash partitioning. This exposes the number of output partitions without claiming stronger partitioning semantics
#### Follow-up work:
- Add size-aware task bin-packing using `file_size_in_bytes` (#128)
- Add sub-file / row-group level parallelism as the longer-term direction (#1604)
- Cache planned tasks for repeated scans with the same snapshot, projection, filter, and target partition count
- Investigate stronger output partitioning declarations for safe identity-partitioned cases
- Add benchmarks for multi-file tables and skewed file-size distributions
## Are these changes tested?
Yes. Integration tests cover:
- eager scan planning disabled by default
- enabling eager scan planning through `iceberg.enable_eager_scan_planning`
- exposing multiple scan output partitions when eager planning is enabled
- preserving query results between single-partition and multi-partition scans
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5093429458
Hello @mbutrovich, thanks for this second pass! I addressed your comments in https://github.com/apache/iceberg-rust/pull/2671/commits/1b0504d93faa3b585fc80f1a9b3977e4ffe05e26.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "NoahKusaba (via GitHub)" <gi...@apache.org>.
NoahKusaba commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5819042406
Hey @toutane re-open your PR in the new datafusion-iceberg repo.
https://github.com/apache/datafusion-iceberg/pulls
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3720153002
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +112,531 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_multi_row_group_table_context(
+ namespace_name: &str,
+ table_name: &str,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ let table = iceberg_catalog.create_table(&namespace, creation).await?;
+ let arrow_schema: Arc<ArrowSchema> = Arc::new(
+ table
+ .metadata()
+ .current_schema()
+ .as_ref()
+ .try_into()
+ .unwrap(),
+ );
+
+ let file_rows = [
+ vec![
+ (1, "row-1"),
+ (2, "row-2"),
+ (100, "row-100"),
+ (101, "row-101"),
+ ],
+ vec![
+ (50, "row-50"),
+ (150, "row-150"),
+ (151, "row-151"),
+ (152, "row-152"),
+ ],
+ vec![
+ (99, "row-99"),
+ (1000, "row-1000"),
+ (1001, "row-1001"),
+ (1002, "row-1002"),
+ ],
+ ];
+
+ let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
+ let writer_properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut data_files = Vec::with_capacity(file_rows.len());
+
+ for (file_idx, rows) in file_rows.into_iter().enumerate() {
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ writer_properties.clone(),
+ table.metadata().current_schema().clone(),
+ );
+ let file_name_generator = DefaultFileNameGenerator::new(
+ format!("multi-row-group-{file_idx}"),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+ let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size(
+ parquet_writer_builder,
+ table.file_io().clone(),
+ location_generator.clone(),
+ file_name_generator,
+ );
+ let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
+ let mut data_file_writer = data_file_writer_builder.build(None).await?;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(
+ rows.iter().map(|(foo1, _)| *foo1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ Arc::new(StringArray::from(
+ rows.iter().map(|(_, foo2)| *foo2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ])?;
+
+ data_file_writer.write(batch).await?;
+ let file_data_files = data_file_writer.close().await?;
+ assert_eq!(file_data_files.len(), 1);
+ data_files.extend(file_data_files);
+ }
+
+ for data_file in &data_files {
+ let file_path = data_file
+ .file_path()
+ .strip_prefix("file://")
+ .unwrap_or(data_file.file_path());
+ let reader = SerializedFileReader::new(File::open(file_path)?)?;
+ assert!(
+ reader.metadata().num_row_groups() > 1,
+ "expected multiple row groups for {}",
+ data_file.file_path()
+ );
+ }
+
+ let tx = Transaction::new(&table);
+ let action = tx.fast_append().add_data_files(data_files);
+ let tx = action.apply(tx)?;
+ tx.commit(iceberg_catalog.as_ref()).await?;
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_plan(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> Arc<dyn ExecutionPlan> {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ table.scan(&state, None, &[], None).await.unwrap()
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let plan = scan_plan(ctx, namespace, table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+fn find_iceberg_scan(plan: &dyn ExecutionPlan) -> Option<&IcebergTableScan> {
+ if let Some(scan) = plan.downcast_ref::<IcebergTableScan>() {
+ return Some(scan);
+ }
+
+ plan.children()
+ .into_iter()
+ .find_map(|child| find_iceberg_scan(child.as_ref()))
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let plan = scan_plan(&ctx, &namespace, &table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ let actual_partition_count = plan.properties().output_partitioning().partition_count();
+ assert_eq!(actual_partition_count, data_file_count);
+
+ // Pins the eager plan's task-group accounting. The reader settings it derives from
+ // the retained TableScan are not introspectable, so their equivalence with the lazy
+ // path is covered behaviorally by test_multi_partition_scan_matches_single_partition_results.
+ let display = datafusion::physical_plan::displayable(plan.as_ref())
+ .one_line()
+ .to_string();
+ assert!(
+ display.contains("task_groups:[3] tasks:[3]"),
+ "unexpected eager scan display: {display}"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query =
+ format!("SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} LIMIT {limit}");
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ // The physical optimizer absorbs the initial GlobalLimitExec into
+ // CoalescePartitionsExec; its fetch enforces the global bound in the final plan.
+ let global_limit_coalescer = plan
+ .downcast_ref::<CoalescePartitionsExec>()
Review Comment:
Done in ab4e8ff7823084243e6fb3231d9582461b4b200d
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3721537069
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,234 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
Review Comment:
Done: https://github.com/apache/iceberg-rust/issues/2962
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3720202522
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +134,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .eager_plan
+ {
+ Some(eager_plan) => {
+ let Some(file_task_group) = eager_plan.task_group(partition) else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ eager_plan.partition_count()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = eager_plan
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
Review Comment:
Is there an open issue for this? If so we should link here so future developers can track this TODO status.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,234 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
Review Comment:
Recommend opening an issue and linking here.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,234 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
Review Comment:
Recommend opening an issue and linking here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "alexanderbianchi (via GitHub)" <gi...@apache.org>.
alexanderbianchi commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-6116801448
Clarification, ArrowReaderBuilder and Table::reader_builder() are already publicn [Example](https://github.com/alexanderbianchi/datafusion-distributed/blob/40212b7dc7a621efc739fca0fffb01e9b53d913c/iceberg/src/data_source.rs#L261). Exposing the builder type isn't the issue. The missing piece is a way to retain the reader settings chosen by a TableScan without retaining its manifest-planning state or binding the eventual reader to that scan's FileIO and runtime.
This PR is being referenced in datafusion-iceberg now, and should largely be moved there.
I can see the case for a lightweight reader-options value that travels with the scan plan. Ideally iceberg-rust would own the settings-to-builder mapping for both to_arrow() and external consumers, so the paths cannot drift.
The eager DataFusion path could retain FileScanTasks and reader options, then drop TableScan. At execution it could create a reader from local resources and apply those options. The same boundary also works for distributed engines: DFD already plans tasks separately and builds readers on workers with worker-local FileIO and runtime (datafusion-contrib/datafusion-distributed#773).
```text
Planning:
scan.plan_files() → FileScanTasks
scan.reader_options() → lightweight reader settings
drop scan
Execution, locally or on a worker:
builder = ArrowReaderBuilder::new(local_file_io, local_runtime)
reader = reader_options.apply_to(builder).build()
reader.read(assigned_tasks)
```
For distributed execution, decoupling plan_files() from reading is an important supported use case: tasks are planned centrally, then workers construct readers with local storage/runtime resources and runtime predicate. Ideally we wouldn't need a `TableScan::arrow_reader_builder()`, and reader's wouldn't need to be transported.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3681162309
##########
crates/integrations/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::config::ConfigExtension;
+use datafusion::common::extensions_options;
+
+extensions_options! {
+ /// Configuration options for Iceberg's DataFusion integration.
+ pub struct IcebergDataFusionConfig {
Review Comment:
Good point. I documented on `IcebergDataFusionConfig` that applications must register the extension with `SessionConfig::with_option_extension` before using `SET iceberg.<option> = <value>`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r4153186316
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -498,8 +498,8 @@ impl TableScan {
Ok(file_scan_task_rx.boxed())
}
- /// Returns an [`ArrowRecordBatchStream`].
- pub async fn to_arrow(&self) -> Result<ArrowRecordBatchStream> {
+ /// Returns an [`ArrowReaderBuilder`] configured for this table scan.
+ pub fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
Review Comment:
Since `crates/iceberg` is `publish = true`, this pins `ArrowReaderBuilder` — an impl type — into the crate's stable API the moment it ships to crates.io, and we'd be doing it purely to feed the DataFusion integration. Once it's out, reshaping or removing it is a breaking change, and every future reader knob on `TableScan` has to be mirrored here or external callers who captured the builder silently miss it.
I'd rather not expose the builder itself. Could we hand back a small value type — a `TableScanReaderConfig` via something like `TableScan::reader_config()` — that the DataFusion crate turns into its own `ArrowReaderBuilder`? That keeps the stable surface a plain config that can't be misused as a decoupled reader, and it also solves the `Arc<TableScan>` retention point I left on `scan_planning.rs`. If we do keep the builder public, the doc needs to spell out the `plan_files()` + `build().read()` pairing, since `to_arrow()` is the intended path for everyone else.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,235 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
Review Comment:
I'd make `IcebergScanConfig::new` fallible and drop this `unwrap()`. `Schema::project` returns a `Result`, and an out-of-bounds projection index turns into a process panic here with nowhere to recover — the kind of thing we've been stamping out across the crate. Both call sites in `table/mod.rs` are already inside `async fn -> DFResult<…>`, so returning `DFResult<Self>` and propagating with `.map_err(|e| DataFusionError::ArrowError(e, None))?` is a clean, local change. DataFusion validating indices upstream makes it unreachable today, but I'd rather not leave a panic on the read path betting on that staying true.
##########
crates/integrations/datafusion/src/config.rs:
##########
@@ -0,0 +1,34 @@
+// 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::config::ConfigExtension;
+use datafusion::common::extensions_options;
+
+extensions_options! {
+ /// Configuration options for Iceberg's DataFusion integration.
+ ///
+ /// Register this extension with [`SessionConfig::with_option_extension`](datafusion::prelude::SessionConfig::with_option_extension)
+ /// before creating a session to make `SET iceberg.<option> = <value>` available.
+ pub struct IcebergDataFusionConfig {
+ /// Plan Iceberg file scan tasks during TableProvider::scan().
+ pub enable_eager_scan_planning: bool, default = false
+ }
+}
+
+impl ConfigExtension for IcebergDataFusionConfig {
+ const PREFIX: &'static str = "iceberg";
Review Comment:
`PREFIX = "iceberg"` claims the whole `iceberg.*` option namespace for this one extension. DataFusion wants a unique prefix per registered `ExtensionOptions`, so the next Iceberg DF extension — write config, catalog auth, whatever — that also picks `iceberg` collides at registration. I'd scope it now, `iceberg.scan` or `iceberg.datafusion`, while nothing depends on the bare prefix yet.
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +134,61 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .eager_plan
+ {
+ Some(eager_plan) => {
+ let Some(file_task_group) = eager_plan.task_group(partition) else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ eager_plan.partition_count()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = eager_plan
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
Review Comment:
`build()` constructs a fresh `CachingDeleteFileLoader` per `execute(partition)`, so each output partition gets its own cache. A delete file referenced by tasks in several partitions then gets fetched from object storage once per partition — with `target_partitions = 16` and equality deletes that's roughly 16x the delete I/O the lazy path does with its single loader. Results stay correct, but it's a surprising cost behind a flag people flip on for speed. I'd build the reader once in `plan_eager_scan()` and share it across partitions (`ArrowReader` is `Clone`), or if that's follow-up material, call it out in the PR description and a tracking issue alongside the other TODOs here.
While we're on this chain: `with_data_file_concurrency_limit(1)` silently overrides whatever the user configured (the builder just picked that up inside `arrow_reader_builder()`). It's a deliberate choice per the comment, but it should be documented on the flag so nobody's surprised their concurrency setting does nothing in eager mode.
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +134,61 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .eager_plan
+ {
+ Some(eager_plan) => {
+ let Some(file_task_group) = eager_plan.task_group(partition) else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ eager_plan.partition_count()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = eager_plan
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ // Tracked in https://github.com/apache/iceberg-rust/issues/2964.
+ .read(tasks)
+ .map_err(to_datafusion_error)?
+ .stream()
+ .map_err(to_datafusion_error);
+
+ Box::pin(stream)
+ }
+ None => {
Review Comment:
The eager arm guards `partition` (the `else` return a few lines up), but this lazy arm ignores it and always returns the full-scan stream. `IcebergTableScan` is public, so a caller holding an `Arc` and calling `execute(1)` directly gets a valid stream for a partition that doesn't exist — duplicate rows instead of an error. Normal DataFusion flow keeps `partition_count` at 1 here so it never fires, but the asymmetry is a latent trap; I'd add the same `partition != 0` guard returning `Internal` at the top of this arm.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,235 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
Review Comment:
Holding `Arc<TableScan>` here keeps its `PlanContext` / `ObjectCache` alive for the whole physical-plan lifetime — that's a moka cache sized to `DEFAULT_CACHE_SIZE_BYTES` (32MB), populated during `plan_files()` and never read again after planning. A query joining N eager-scanned tables pins up to N×32MB for its full duration; the lazy path avoids this since its `TableScan` is built and dropped inside `execute()`. All `execute()` actually needs off the scan is the reader settings, so I'd pull those into a small struct at planning time and store that instead, letting the `TableScan` drop once `plan_eager_scan()` returns. This is the same reader-config value type that would let us keep `arrow_reader_builder` off the public API — one change covers both.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,235 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // TODO: Cache eager scan planning results across equivalent provider scan calls with a
+ // precise cache key. Catalog-backed providers must still reload table metadata before
+ // cache lookup so that a new snapshot naturally causes a cache miss.
+ // Tracked in https://github.com/apache/iceberg-rust/issues/2963.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+// Tracked in https://github.com/apache/iceberg-rust/issues/2962.
+fn group_file_scan_tasks_round_robin(
+ tasks: Vec<FileScanTask>,
+ target_partitions: usize,
+) -> Vec<Vec<FileScanTask>> {
+ if tasks.is_empty() {
+ return vec![vec![]];
+ }
+
+ let target_partitions = target_partitions.max(1).min(tasks.len());
+
+ let mut groups: Vec<Vec<FileScanTask>> = vec![Vec::new(); target_partitions];
+ for (i, task) in tasks.into_iter().enumerate() {
+ groups[i % target_partitions].push(task);
+ }
+
+ groups
+}
+
+pub(crate) fn build_table_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+) -> DFResult<TableScan> {
+ let builder = match scan_config.snapshot_id {
+ Some(id) => table.scan().snapshot_id(id),
+ None => table.scan(),
+ };
+ let mut builder = match scan_config.column_names.clone() {
+ Some(names) => builder.select(names),
+ None => builder.select_all(),
+ };
+ if let Some(pred) = scan_config.predicates.clone() {
+ builder = builder.with_filter(pred);
+ }
+ builder.build().map_err(to_datafusion_error)
+}
+
+#[cfg(test)]
+mod tests {
+ use iceberg::spec::{DataFileFormat, Schema};
+
+ use super::*;
+
+ #[test]
+ fn test_group_file_scan_tasks_round_robin_uneven_split() {
Review Comment:
The one unit test covers the 5-tasks / 3-partitions uneven case, but the load-bearing line is the `.max(1).min(tasks.len())` clamp — it's what stops us handing DataFusion empty partitions. I'd add direct cases for it: empty input (one empty group), `target_partitions = 0` (clamped to 1), and `target_partitions > tasks.len()` (e.g. 3 tasks / 10 partitions producing 3 groups, not 10). That last one is the case an integration test won't reliably catch, since it leans on the real file count.
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +112,532 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_multi_row_group_table_context(
+ namespace_name: &str,
+ table_name: &str,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ let table = iceberg_catalog.create_table(&namespace, creation).await?;
+ let arrow_schema: Arc<ArrowSchema> = Arc::new(
+ table
+ .metadata()
+ .current_schema()
+ .as_ref()
+ .try_into()
+ .unwrap(),
+ );
+
+ let file_rows = [
+ vec![
+ (1, "row-1"),
+ (2, "row-2"),
+ (100, "row-100"),
+ (101, "row-101"),
+ ],
+ vec![
+ (50, "row-50"),
+ (150, "row-150"),
+ (151, "row-151"),
+ (152, "row-152"),
+ ],
+ vec![
+ (99, "row-99"),
+ (1000, "row-1000"),
+ (1001, "row-1001"),
+ (1002, "row-1002"),
+ ],
+ ];
+
+ let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
+ let writer_properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut data_files = Vec::with_capacity(file_rows.len());
+
+ for (file_idx, rows) in file_rows.into_iter().enumerate() {
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ writer_properties.clone(),
+ table.metadata().current_schema().clone(),
+ );
+ let file_name_generator = DefaultFileNameGenerator::new(
+ format!("multi-row-group-{file_idx}"),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+ let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size(
+ parquet_writer_builder,
+ table.file_io().clone(),
+ location_generator.clone(),
+ file_name_generator,
+ );
+ let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
+ let mut data_file_writer = data_file_writer_builder.build(None).await?;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(
+ rows.iter().map(|(foo1, _)| *foo1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ Arc::new(StringArray::from(
+ rows.iter().map(|(_, foo2)| *foo2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ])?;
+
+ data_file_writer.write(batch).await?;
+ let file_data_files = data_file_writer.close().await?;
+ assert_eq!(file_data_files.len(), 1);
+ data_files.extend(file_data_files);
+ }
+
+ for data_file in &data_files {
+ let file_path = data_file
+ .file_path()
+ .strip_prefix("file://")
+ .unwrap_or(data_file.file_path());
+ let reader = SerializedFileReader::new(File::open(file_path)?)?;
+ assert!(
+ reader.metadata().num_row_groups() > 1,
+ "expected multiple row groups for {}",
+ data_file.file_path()
+ );
+ }
+
+ let tx = Transaction::new(&table);
+ let action = tx.fast_append().add_data_files(data_files);
+ let tx = action.apply(tx)?;
+ tx.commit(iceberg_catalog.as_ref()).await?;
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_plan(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> Arc<dyn ExecutionPlan> {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ table.scan(&state, None, &[], None).await.unwrap()
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let plan = scan_plan(ctx, namespace, table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+fn find_iceberg_scan(plan: &dyn ExecutionPlan) -> Option<&IcebergTableScan> {
+ if let Some(scan) = plan.downcast_ref::<IcebergTableScan>() {
+ return Some(scan);
+ }
+
+ plan.children()
+ .into_iter()
+ .find_map(|child| find_iceberg_scan(child.as_ref()))
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let plan = scan_plan(&ctx, &namespace, &table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ let actual_partition_count = plan.properties().output_partitioning().partition_count();
+ assert_eq!(actual_partition_count, data_file_count);
+
+ // Pins the eager plan's task-group accounting. The reader settings it derives from
+ // the retained TableScan are not introspectable, so their equivalence with the lazy
+ // path is covered behaviorally by test_multi_partition_scan_matches_single_partition_results.
+ let display = datafusion::physical_plan::displayable(plan.as_ref())
+ .one_line()
+ .to_string();
+ assert!(
+ display.contains("task_groups:[3] tasks:[3]"),
+ "unexpected eager scan display: {display}"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query =
+ format!("SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} LIMIT {limit}");
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
Review Comment:
This calls `create_physical_plan()` to inspect the plan and then `collect()` on the dataframe, which re-plans — so `plan_eager_scan()` runs twice and the plan you assert on isn't the object that executes; the shape assertions don't actually constrain what ran. I'd plan once and execute that same plan. Separately, the `downcast_ref::<CoalescePartitionsExec>()` pins the DF55 plan shape — a future bump that stacks a `GlobalLimitExec` on top would panic the downcast rather than fail readably — so I'd lean on the summed-row-count `== limit` assertion as the real invariant and treat the shape check as best-effort.
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +112,532 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_multi_row_group_table_context(
+ namespace_name: &str,
+ table_name: &str,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ let table = iceberg_catalog.create_table(&namespace, creation).await?;
+ let arrow_schema: Arc<ArrowSchema> = Arc::new(
+ table
+ .metadata()
+ .current_schema()
+ .as_ref()
+ .try_into()
+ .unwrap(),
+ );
+
+ let file_rows = [
+ vec![
+ (1, "row-1"),
+ (2, "row-2"),
+ (100, "row-100"),
+ (101, "row-101"),
+ ],
+ vec![
+ (50, "row-50"),
+ (150, "row-150"),
+ (151, "row-151"),
+ (152, "row-152"),
+ ],
+ vec![
+ (99, "row-99"),
+ (1000, "row-1000"),
+ (1001, "row-1001"),
+ (1002, "row-1002"),
+ ],
+ ];
+
+ let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
+ let writer_properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut data_files = Vec::with_capacity(file_rows.len());
+
+ for (file_idx, rows) in file_rows.into_iter().enumerate() {
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ writer_properties.clone(),
+ table.metadata().current_schema().clone(),
+ );
+ let file_name_generator = DefaultFileNameGenerator::new(
+ format!("multi-row-group-{file_idx}"),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+ let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size(
+ parquet_writer_builder,
+ table.file_io().clone(),
+ location_generator.clone(),
+ file_name_generator,
+ );
+ let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
+ let mut data_file_writer = data_file_writer_builder.build(None).await?;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(
+ rows.iter().map(|(foo1, _)| *foo1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ Arc::new(StringArray::from(
+ rows.iter().map(|(_, foo2)| *foo2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ])?;
+
+ data_file_writer.write(batch).await?;
+ let file_data_files = data_file_writer.close().await?;
+ assert_eq!(file_data_files.len(), 1);
+ data_files.extend(file_data_files);
+ }
+
+ for data_file in &data_files {
+ let file_path = data_file
+ .file_path()
+ .strip_prefix("file://")
+ .unwrap_or(data_file.file_path());
+ let reader = SerializedFileReader::new(File::open(file_path)?)?;
+ assert!(
+ reader.metadata().num_row_groups() > 1,
+ "expected multiple row groups for {}",
+ data_file.file_path()
+ );
+ }
+
+ let tx = Transaction::new(&table);
+ let action = tx.fast_append().add_data_files(data_files);
+ let tx = action.apply(tx)?;
+ tx.commit(iceberg_catalog.as_ref()).await?;
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_plan(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> Arc<dyn ExecutionPlan> {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ table.scan(&state, None, &[], None).await.unwrap()
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let plan = scan_plan(ctx, namespace, table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+fn find_iceberg_scan(plan: &dyn ExecutionPlan) -> Option<&IcebergTableScan> {
+ if let Some(scan) = plan.downcast_ref::<IcebergTableScan>() {
+ return Some(scan);
+ }
+
+ plan.children()
+ .into_iter()
+ .find_map(|child| find_iceberg_scan(child.as_ref()))
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let plan = scan_plan(&ctx, &namespace, &table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ let actual_partition_count = plan.properties().output_partitioning().partition_count();
+ assert_eq!(actual_partition_count, data_file_count);
+
+ // Pins the eager plan's task-group accounting. The reader settings it derives from
+ // the retained TableScan are not introspectable, so their equivalence with the lazy
+ // path is covered behaviorally by test_multi_partition_scan_matches_single_partition_results.
+ let display = datafusion::physical_plan::displayable(plan.as_ref())
+ .one_line()
+ .to_string();
+ assert!(
+ display.contains("task_groups:[3] tasks:[3]"),
+ "unexpected eager scan display: {display}"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query =
+ format!("SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} LIMIT {limit}");
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ // The physical optimizer absorbs the initial GlobalLimitExec into
+ // CoalescePartitionsExec; its fetch enforces the global bound in the final plan.
+ // After a DataFusion bump, failure here likely means the plan shape changed, not eager scanning.
+ let global_limit_coalescer = plan
+ .downcast_ref::<CoalescePartitionsExec>()
+ .expect("Expected a globally limited CoalescePartitionsExec");
+ assert_eq!(global_limit_coalescer.fetch(), Some(limit));
+
+ let scan = find_iceberg_scan(global_limit_coalescer.input().as_ref())
+ .expect("Expected IcebergTableScan below the global limit");
+ assert_eq!(scan.limit(), Some(limit));
+ assert_eq!(
+ scan.properties().output_partitioning().partition_count(),
+ data_file_count
+ );
+
+ let batches = dataframe.collect().await.unwrap();
+ let mut seen_ids = std::collections::HashSet::new();
+ for batch in &batches {
+ let foo1 = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let foo2 = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+
+ for row in 0..batch.num_rows() {
+ let id = foo1.value(row);
+ assert!((1..=data_file_count as i32).contains(&id));
+ assert_eq!(foo2.value(row), format!("row-{id}"));
+ assert!(seen_ids.insert(id), "duplicate row {id}");
+ }
+ }
+
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ limit
+ );
+ assert_eq!(seen_ids.len(), limit);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_ordered_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_ordered_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query = format!(
+ "SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} ORDER BY foo1 LIMIT {limit}"
+ );
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ let scan = find_iceberg_scan(plan.as_ref()).expect("Expected IcebergTableScan in ordered plan");
Review Comment:
This test can't actually catch the risk it's guarding. Each file holds one row, so per-partition truncation to `LIMIT 2` can never drop a globally-smaller row — a regression where the limit got pushed into the scan would still pass. The per-partition bound is only safe because DataFusion currently doesn't push a limit through the sort, which leaves `self.limit` as `None` for `ORDER BY … LIMIT`. I'd assert that assumption right here with `assert_eq!(scan.limit(), None)`, and switch to a multi-row-per-file table with `ORDER BY foo1 LIMIT K` (K below a single file's row count) so per-partition truncation would produce wrong rows if it ever kicked in.
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +112,532 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_multi_row_group_table_context(
+ namespace_name: &str,
+ table_name: &str,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ let table = iceberg_catalog.create_table(&namespace, creation).await?;
+ let arrow_schema: Arc<ArrowSchema> = Arc::new(
+ table
+ .metadata()
+ .current_schema()
+ .as_ref()
+ .try_into()
+ .unwrap(),
+ );
+
+ let file_rows = [
+ vec![
+ (1, "row-1"),
+ (2, "row-2"),
+ (100, "row-100"),
+ (101, "row-101"),
+ ],
+ vec![
+ (50, "row-50"),
+ (150, "row-150"),
+ (151, "row-151"),
+ (152, "row-152"),
+ ],
+ vec![
+ (99, "row-99"),
+ (1000, "row-1000"),
+ (1001, "row-1001"),
+ (1002, "row-1002"),
+ ],
+ ];
+
+ let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
+ let writer_properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut data_files = Vec::with_capacity(file_rows.len());
+
+ for (file_idx, rows) in file_rows.into_iter().enumerate() {
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ writer_properties.clone(),
+ table.metadata().current_schema().clone(),
+ );
+ let file_name_generator = DefaultFileNameGenerator::new(
+ format!("multi-row-group-{file_idx}"),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+ let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size(
+ parquet_writer_builder,
+ table.file_io().clone(),
+ location_generator.clone(),
+ file_name_generator,
+ );
+ let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
+ let mut data_file_writer = data_file_writer_builder.build(None).await?;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(
+ rows.iter().map(|(foo1, _)| *foo1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ Arc::new(StringArray::from(
+ rows.iter().map(|(_, foo2)| *foo2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ])?;
+
+ data_file_writer.write(batch).await?;
+ let file_data_files = data_file_writer.close().await?;
+ assert_eq!(file_data_files.len(), 1);
+ data_files.extend(file_data_files);
+ }
+
+ for data_file in &data_files {
+ let file_path = data_file
+ .file_path()
+ .strip_prefix("file://")
+ .unwrap_or(data_file.file_path());
+ let reader = SerializedFileReader::new(File::open(file_path)?)?;
+ assert!(
+ reader.metadata().num_row_groups() > 1,
+ "expected multiple row groups for {}",
+ data_file.file_path()
+ );
+ }
+
+ let tx = Transaction::new(&table);
+ let action = tx.fast_append().add_data_files(data_files);
+ let tx = action.apply(tx)?;
+ tx.commit(iceberg_catalog.as_ref()).await?;
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_plan(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> Arc<dyn ExecutionPlan> {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ table.scan(&state, None, &[], None).await.unwrap()
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let plan = scan_plan(ctx, namespace, table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+fn find_iceberg_scan(plan: &dyn ExecutionPlan) -> Option<&IcebergTableScan> {
+ if let Some(scan) = plan.downcast_ref::<IcebergTableScan>() {
+ return Some(scan);
+ }
+
+ plan.children()
+ .into_iter()
+ .find_map(|child| find_iceberg_scan(child.as_ref()))
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let plan = scan_plan(&ctx, &namespace, &table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ let actual_partition_count = plan.properties().output_partitioning().partition_count();
+ assert_eq!(actual_partition_count, data_file_count);
+
+ // Pins the eager plan's task-group accounting. The reader settings it derives from
+ // the retained TableScan are not introspectable, so their equivalence with the lazy
+ // path is covered behaviorally by test_multi_partition_scan_matches_single_partition_results.
+ let display = datafusion::physical_plan::displayable(plan.as_ref())
+ .one_line()
+ .to_string();
+ assert!(
+ display.contains("task_groups:[3] tasks:[3]"),
+ "unexpected eager scan display: {display}"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query =
+ format!("SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} LIMIT {limit}");
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ // The physical optimizer absorbs the initial GlobalLimitExec into
+ // CoalescePartitionsExec; its fetch enforces the global bound in the final plan.
+ // After a DataFusion bump, failure here likely means the plan shape changed, not eager scanning.
+ let global_limit_coalescer = plan
+ .downcast_ref::<CoalescePartitionsExec>()
+ .expect("Expected a globally limited CoalescePartitionsExec");
+ assert_eq!(global_limit_coalescer.fetch(), Some(limit));
+
+ let scan = find_iceberg_scan(global_limit_coalescer.input().as_ref())
+ .expect("Expected IcebergTableScan below the global limit");
+ assert_eq!(scan.limit(), Some(limit));
+ assert_eq!(
+ scan.properties().output_partitioning().partition_count(),
+ data_file_count
+ );
+
+ let batches = dataframe.collect().await.unwrap();
+ let mut seen_ids = std::collections::HashSet::new();
+ for batch in &batches {
+ let foo1 = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let foo2 = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+
+ for row in 0..batch.num_rows() {
+ let id = foo1.value(row);
+ assert!((1..=data_file_count as i32).contains(&id));
+ assert_eq!(foo2.value(row), format!("row-{id}"));
+ assert!(seen_ids.insert(id), "duplicate row {id}");
+ }
+ }
+
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ limit
+ );
+ assert_eq!(seen_ids.len(), limit);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_ordered_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_ordered_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query = format!(
+ "SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} ORDER BY foo1 LIMIT {limit}"
+ );
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ let scan = find_iceberg_scan(plan.as_ref()).expect("Expected IcebergTableScan in ordered plan");
+ assert_eq!(
+ scan.properties().output_partitioning().partition_count(),
+ data_file_count
+ );
+
+ let batches = dataframe.collect().await.unwrap();
+ let rows = batches
+ .iter()
+ .flat_map(|batch| {
+ let foo1 = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let foo2 = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+
+ (0..batch.num_rows()).map(|row| (foo1.value(row), foo2.value(row).to_string()))
+ })
+ .collect::<Vec<_>>();
+
+ assert_eq!(rows, vec![
+ (1, "row-1".to_string()),
+ (2, "row-2".to_string())
+ ]);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_matches_single_partition_results() -> Result<()> {
Review Comment:
None of the new eager tests exercise delete files — every table here is built without them. The association looks correct by construction (each task carries its own `deletes`, and round-robin grouping keeps that intact per task), but for a read-path change this is exactly where I'd want a test rather than reasoning. I'd add a positional-delete table scanned in eager mode across multiple partitions and assert the deleted rows are absent, mirroring the parity check this test does — equality deletes too if it's cheap.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "alexanderbianchi (via GitHub)" <gi...@apache.org>.
alexanderbianchi commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-6348795093
@laskoviymishka I'm going to pick up this work from @toutane. (we both work at datadog) we can carry on with https://github.com/apache/datafusion-iceberg/pull/67 for the DF components and i'll open up a iceberg_rust PR now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5193387421
> I think you've done a great job pushing through on this topic. I know there have been many iterations and I think it's paid off because this looks like a very clean PR. I don't have committer access, so I can only provide an extra "approve" that doesn't carry weight.
>
> The DataFusion parts specifically look very good to me. I am unable to test at scale myself, so I will have to rely on others or on the included tests.
>
> The comments I have are all documentation, not code issues.
@timsaucer, thanks a lot for your message, really appreciate it!
For testing, we're going to shadow real traffic in our infra. It won't be a standard benchmark, but it should at least tell us whether things are improving on real-world queries.
That said, I think having a standard way to benchmark this kind of optimization would be really valuable for the `iceberg-datafusion` crate. Not sure if that's already been discussed somewhere?
Thanks again for the comments - I opened three issues to track the TODOs.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5191152915
> Thanks for staying with this through several rounds, this is close. Two small suggestions inline while we wait for a committer to review, both about test coverage rather than correctness of the feature itself.
Thank you for continuing to review this code, @mbutrovich.
Do you know of any contributors who might be able to take this on?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3667578012
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +137,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = build_table_scan(&self.table, &self.scan_config)?
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ .read(tasks)
+ .map_err(to_datafusion_error)?
+ .stream()
+ .map_err(to_datafusion_error);
+
+ Box::pin(stream)
+ }
+ None => {
+ let table = self.table.clone();
+ let scan_config = self.scan_config.clone();
+ let fut = async move {
+ let table_scan = build_table_scan(&table, &scan_config)?;
+ let stream = table_scan
+ .to_arrow()
+ .await
+ .map_err(to_datafusion_error)?
+ .map_err(to_datafusion_error);
+ Ok::<_, datafusion::common::DataFusionError>(stream)
+ };
+
+ Box::pin(futures::stream::once(fut).try_flatten())
+ }
+ };
+
+ // Apply a scan-partition bound if specified. In eager planning this is only
Review Comment:
This explains why the per-partition limit looseness is safe (a `GlobalLimitExec` above the scan enforces the real bound), which is a fair resolution of the earlier comment. But nothing exercises it: no test runs a `SELECT ... LIMIT k` query against an eager multi-partition scan and checks the row count and rows. Since this PR is what introduces the N-way over-read in the first place (the single-partition lazy path couldn't over-read), a targeted test here would confirm the `GlobalLimitExec` assumption holds end to end rather than resting on the comment alone.
##########
crates/integrations/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::config::ConfigExtension;
+use datafusion::common::extensions_options;
+
+extensions_options! {
+ /// Configuration options for Iceberg's DataFusion integration.
+ pub struct IcebergDataFusionConfig {
Review Comment:
`SET iceberg.enable_eager_scan_planning = true` will not work on a plain `SessionContext`. Nothing in this crate registers `IcebergDataFusionConfig` as an extension anywhere except the test helper (`read_session_config` in `integration_datafusion_test.rs:287-297`, via `.with_option_extension(...)`).
Checked `datafusion-common` 54.1.0 (pinned in this repo's `Cargo.lock`): `ConfigOptions::set` looks up the prefix in `self.extensions.0`, and if nothing has registered `"iceberg"`, it returns `Could not find config namespace "iceberg"` rather than setting the option. `test_set_enable_eager_scan_planning` (`integration_datafusion_test.rs:359`) only passes because it pre-registers the extension with `Some(false)` before running the `SET` statement, so it doesn't cover the case of a user who never touches the extension in Rust first.
So today, using this feature via SQL `SET` requires the embedding application to already know to call `.with_option_extension(IcebergDataFusionConfig::default())` before `SET` can target it at all, and that's not documented anywhere. Worth a line on `IcebergDataFusionConfig`'s doc comment spelling out the registration requirement, or registering it somewhere reachable (e.g. `IcebergCatalogProvider::try_new`) so `SET` works without that precondition.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5128991193
Thank for this pass @mbutrovich!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3681266502
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +137,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = build_table_scan(&self.table, &self.scan_config)?
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ .read(tasks)
+ .map_err(to_datafusion_error)?
+ .stream()
+ .map_err(to_datafusion_error);
+
+ Box::pin(stream)
+ }
+ None => {
+ let table = self.table.clone();
+ let scan_config = self.scan_config.clone();
+ let fut = async move {
+ let table_scan = build_table_scan(&table, &scan_config)?;
+ let stream = table_scan
+ .to_arrow()
+ .await
+ .map_err(to_datafusion_error)?
+ .map_err(to_datafusion_error);
+ Ok::<_, datafusion::common::DataFusionError>(stream)
+ };
+
+ Box::pin(futures::stream::once(fut).try_flatten())
+ }
+ };
+
+ // Apply a scan-partition bound if specified. In eager planning this is only
Review Comment:
- 5ae5544b7f12c1399bad9a855085349ca8c99b90: add a targeted eager multi-partition `SELECT ... LIMIT k` integration test that verifies the scan has multiple partitions, the optimized plan keeps a global bound above the scan, and the collected result contains exactly `k` valid rows .
- 1082c50de3bf9ef01b765d4422b9e7e0ef0f1c6d: add a separate `ORDER BY foo1 LIMIT k` test to cover the deterministic ordered case end to end.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3720129529
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,200 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+fn group_file_scan_tasks_round_robin(
Review Comment:
Address in f66c9b9461c945edb634153a21d20740981985d3 via`test_group_file_scan_tasks_round_robin_uneven_split`
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,200 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+fn group_file_scan_tasks_round_robin(
Review Comment:
Addressed in f66c9b9461c945edb634153a21d20740981985d3 via`test_group_file_scan_tasks_round_robin_uneven_split`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3721533882
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +134,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .eager_plan
+ {
+ Some(eager_plan) => {
+ let Some(file_task_group) = eager_plan.task_group(partition) else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ eager_plan.partition_count()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = eager_plan
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
Review Comment:
Done: https://github.com/apache/iceberg-rust/issues/2964
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,234 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
Review Comment:
Done: https://github.com/apache/iceberg-rust/issues/2963
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3558635085
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -146,13 +139,55 @@ impl ExecutionPlan for IcebergTableScan {
_partition: usize,
Review Comment:
`execute`'s `partition` argument is read at line 146 (`file_task_groups.get(_partition)`) and interpolated into the error message at line 148 (`partition {_partition} does not exist`). The leading underscore signals "unused" to readers and lint, and it leaks into user-facing error text. Rename to `partition`. (`_context` at line 140 is still legitimately unused; leave it.)
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -146,13 +139,55 @@ impl ExecutionPlan for IcebergTableScan {
_partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(_partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {_partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let task_count = file_task_group.len();
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..task_count).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
Review Comment:
`task_count` is used once; inline it as `(0..file_task_group.len())`. Heads-up for anyone tempted to go further: `file_task_group.iter().cloned().map(Ok)` does not compile here, because the stream must be `'static` and `iter()` would borrow the local `Arc<[FileScanTask]>`. The move-plus-index form is the allocation-free way to clone lazily out of the `Arc`, so keep that shape; just remove the extra binding.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,150 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+pub(crate) async fn plan_file_task_groups(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<Vec<Vec<FileScanTask>>> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let tasks: Vec<FileScanTask> = build_table_scan(table, scan_config)?
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ Ok(group_file_scan_tasks_round_robin(tasks, target_partitions))
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. `target_partitions` is clamped to a minimum of 1.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+fn group_file_scan_tasks_round_robin(
+ tasks: Vec<FileScanTask>,
+ target_partitions: usize,
+) -> Vec<Vec<FileScanTask>> {
+ if tasks.is_empty() {
+ return vec![vec![]];
+ }
+
+ let target_partitions = target_partitions.max(1);
+
+ let mut groups: Vec<Vec<FileScanTask>> = vec![Vec::new(); target_partitions];
+ for (i, task) in tasks.into_iter().enumerate() {
+ groups[i % target_partitions].push(task);
+ }
+
+ groups.retain(|group| !group.is_empty());
+ groups
Review Comment:
When `target_partitions > tasks.len()` (which the tests deliberately hit with `data_file_count + 1`), this allocates extra empty vecs and then walks the whole vec to remove them. The result is always `min(target_partitions, tasks.len())` non-empty groups, so clamp up front and the `retain` disappears:
```rust
// tasks is non-empty here (early return above handles the empty case)
let target_partitions = target_partitions.max(1).min(tasks.len());
let mut groups: Vec<Vec<FileScanTask>> = vec![Vec::new(); target_partitions];
for (i, task) in tasks.into_iter().enumerate() {
groups[i % target_partitions].push(task);
}
groups
```
##########
crates/integrations/datafusion/src/table/mod.rs:
##########
Review Comment:
Both `IcebergTableProvider::scan` (`:158`, used at `:178`) and `IcebergStaticTableProvider::scan` (`:343`, used at `:352`) pass `_state` into `create_scan_plan`, which reads it via `enable_eager_scan_planning(state)`. Rename both to `state`. Note: `_state` at `:365` (`insert_into` is genuinely unused; leave it.)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-4989845540
@mbutrovich Could you please give this another look?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3703807458
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +137,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = build_table_scan(&self.table, &self.scan_config)?
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ .read(tasks)
Review Comment:
Hey @mbutrovich, thanks for the thorough review and for staying with this PR.
Commit 9fe0442 addresses it by threading the planning `TableScan` through to `execute()`.
`scan_planning.rs` gains an `EagerScanPlan` struct holding the `Arc<TableScan>` that planned the tasks, alongside those tasks grouped per output partition. `plan_file_task_groups` is renamed `plan_eager_scan` and returns that struct instead of a bare `Vec<Vec<FileScanTask>>`. `TableScan` isn't `Clone`, but `arrow_reader_builder(&self)` and `plan_files(&self)` both take `&self`, so an `Arc` is enough and no change to the `iceberg` crate is needed.
`IcebergTableScan` stores it as `eager_plan: Option<EagerScanPlan>`, replacing the previous `Option<Vec<Arc<[FileScanTask]>>>`. A single `Option` keeps the "tasks planned = the scan that planned them" invariant in the type instead of spread across two fields; `None` still selects the lazy path.
To answer your question directly: it is the `TableScan` instance that is reused, not the `ArrowReaderBuilder`. Each `execute(partition)` still calls `arrow_reader_builder()` on that shared scan and builds its own reader - that part is only a handful of `Arc` clones.
On the gap you flagged, both paths still go through `build_table_scan` and `TableScan::arrow_reader_builder()`. `build_table_scan` has two callers - `plan_eager_scan` at plan time, and the lazy arm of `execute()` at execute time - so it stays the single source for snapshot, projection, predicate and reader settings. The eager reader is now derived from the very instance that planned the tasks, which makes that equivalence structural rather than a calling convention. `test_multi_partition_scan_matches_single_partition_results` still pins it behaviorally across lazy, eager with 1 partition, and eager with 4 partitions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3691811085
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +137,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = build_table_scan(&self.table, &self.scan_config)?
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ .read(tasks)
Review Comment:
The eager arm rebuilds a `TableScan` via `build_table_scan` on every `execute(partition)` call just to read `.arrow_reader_builder()` off it. `build_table_scan` (`scan_planning.rs:133-148`) does a snapshot lookup, per-column schema validation, field id resolution, and `PlanContext` construction. `arrow_reader_builder()` (`crates/iceberg/src/scan/mod.rs:484-496`) does not read any of that: it only copies `file_io`, `runtime`, `concurrency_limit_data_files`, `row_group_filtering_enabled`, `row_selection_enabled`, and `batch_size` off the `TableScan`. So this runs once per output partition per scan, and none of the snapshot/schema/predicate work it does is used by the one thing this call site wants from it.
This isn't an argument for dropping `TableScan` from this path. Routing both the eager arm and `to_arrow()` through the same `arrow_reader_builder()` is what closed the "two independently-hardcoded default sources with nothing pinning their equivalence" gap from the earlier round on this PR (https://github.com/apache/iceberg-rust/pull/2671#discussion_r3605675108). Building `ArrowReaderBuilder::new(...)` directly here instead would reopen that gap.
`plan_file_task_groups` (`scan_planning.rs:80-113`) already builds a `TableScan` once, to call `.plan_files()`. Could that same instance, or the `ArrowReaderBuilder` it produces, be threaded through to `IcebergTableScan` and reused by every `execute(partition)` call, instead of rebuilding it per partition here?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5667978957
@mbutrovich @toutane - Are there any pending items in this PR? This is a prerequisite for my work in https://github.com/apache/iceberg-rust/issues/3126.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3715740272
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +112,531 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_multi_row_group_table_context(
+ namespace_name: &str,
+ table_name: &str,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ let table = iceberg_catalog.create_table(&namespace, creation).await?;
+ let arrow_schema: Arc<ArrowSchema> = Arc::new(
+ table
+ .metadata()
+ .current_schema()
+ .as_ref()
+ .try_into()
+ .unwrap(),
+ );
+
+ let file_rows = [
+ vec![
+ (1, "row-1"),
+ (2, "row-2"),
+ (100, "row-100"),
+ (101, "row-101"),
+ ],
+ vec![
+ (50, "row-50"),
+ (150, "row-150"),
+ (151, "row-151"),
+ (152, "row-152"),
+ ],
+ vec![
+ (99, "row-99"),
+ (1000, "row-1000"),
+ (1001, "row-1001"),
+ (1002, "row-1002"),
+ ],
+ ];
+
+ let location_generator = DefaultLocationGenerator::new(table.metadata()).unwrap();
+ let writer_properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(2))
+ .build();
+ let mut data_files = Vec::with_capacity(file_rows.len());
+
+ for (file_idx, rows) in file_rows.into_iter().enumerate() {
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ writer_properties.clone(),
+ table.metadata().current_schema().clone(),
+ );
+ let file_name_generator = DefaultFileNameGenerator::new(
+ format!("multi-row-group-{file_idx}"),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+ let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size(
+ parquet_writer_builder,
+ table.file_io().clone(),
+ location_generator.clone(),
+ file_name_generator,
+ );
+ let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);
+ let mut data_file_writer = data_file_writer_builder.build(None).await?;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(
+ rows.iter().map(|(foo1, _)| *foo1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ Arc::new(StringArray::from(
+ rows.iter().map(|(_, foo2)| *foo2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ])?;
+
+ data_file_writer.write(batch).await?;
+ let file_data_files = data_file_writer.close().await?;
+ assert_eq!(file_data_files.len(), 1);
+ data_files.extend(file_data_files);
+ }
+
+ for data_file in &data_files {
+ let file_path = data_file
+ .file_path()
+ .strip_prefix("file://")
+ .unwrap_or(data_file.file_path());
+ let reader = SerializedFileReader::new(File::open(file_path)?)?;
+ assert!(
+ reader.metadata().num_row_groups() > 1,
+ "expected multiple row groups for {}",
+ data_file.file_path()
+ );
+ }
+
+ let tx = Transaction::new(&table);
+ let action = tx.fast_append().add_data_files(data_files);
+ let tx = action.apply(tx)?;
+ tx.commit(iceberg_catalog.as_ref()).await?;
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_plan(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> Arc<dyn ExecutionPlan> {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ table.scan(&state, None, &[], None).await.unwrap()
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let plan = scan_plan(ctx, namespace, table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+fn find_iceberg_scan(plan: &dyn ExecutionPlan) -> Option<&IcebergTableScan> {
+ if let Some(scan) = plan.downcast_ref::<IcebergTableScan>() {
+ return Some(scan);
+ }
+
+ plan.children()
+ .into_iter()
+ .find_map(|child| find_iceberg_scan(child.as_ref()))
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let plan = scan_plan(&ctx, &namespace, &table_name).await;
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+
+ let actual_partition_count = plan.properties().output_partitioning().partition_count();
+ assert_eq!(actual_partition_count, data_file_count);
+
+ // Pins the eager plan's task-group accounting. The reader settings it derives from
+ // the retained TableScan are not introspectable, so their equivalence with the lazy
+ // path is covered behaviorally by test_multi_partition_scan_matches_single_partition_results.
+ let display = datafusion::physical_plan::displayable(plan.as_ref())
+ .one_line()
+ .to_string();
+ assert!(
+ display.contains("task_groups:[3] tasks:[3]"),
+ "unexpected eager scan display: {display}"
+ );
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_enforces_global_limit() -> Result<()> {
+ let data_file_count = 3;
+ let limit = 2;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_partition_scan_limit",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let namespace_name = &namespace[0];
+ let query =
+ format!("SELECT foo1, foo2 FROM catalog.{namespace_name}.{table_name} LIMIT {limit}");
+ let dataframe = ctx.sql(&query).await.unwrap();
+ let plan = dataframe.create_physical_plan().await.unwrap();
+
+ // The physical optimizer absorbs the initial GlobalLimitExec into
+ // CoalescePartitionsExec; its fetch enforces the global bound in the final plan.
+ let global_limit_coalescer = plan
+ .downcast_ref::<CoalescePartitionsExec>()
Review Comment:
This is a good end-to-end check that the global limit is actually enforced. One thought: it pins today's DataFusion optimizer output shape (`GlobalLimitExec` absorbed into `CoalescePartitionsExec::fetch`). If a future DataFusion bump changes how that composes, this assertion could fail with no regression in this crate's code. Might be worth a short comment here noting that a failure after a `datafusion` version bump likely means the plan shape changed, not that eager scanning broke, so whoever hits it later doesn't have to rediscover that.
##########
crates/integrations/datafusion/src/physical_plan/scan_planning.rs:
##########
@@ -0,0 +1,200 @@
+// 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 std::sync::Arc;
+
+use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef;
+use datafusion::error::Result as DFResult;
+use datafusion::prelude::Expr;
+use futures::TryStreamExt;
+use iceberg::arrow::ArrowReaderBuilder;
+use iceberg::expr::Predicate;
+use iceberg::scan::{FileScanTask, TableScan};
+use iceberg::table::Table;
+
+use super::expr_to_predicate::convert_filters_to_predicate;
+use crate::to_datafusion_error;
+
+#[derive(Debug, Clone)]
+pub(crate) struct IcebergScanConfig {
+ /// Snapshot of the table to scan.
+ snapshot_id: Option<i64>,
+ /// Output schema after projection.
+ output_schema: ArrowSchemaRef,
+ /// Projection column names, None means all columns.
+ column_names: Option<Vec<String>>,
+ /// Filters to apply to the table scan.
+ predicates: Option<Predicate>,
+}
+
+impl IcebergScanConfig {
+ pub(crate) fn new(
+ schema: ArrowSchemaRef,
+ snapshot_id: Option<i64>,
+ projection: Option<&Vec<usize>>,
+ filters: &[Expr],
+ ) -> Self {
+ let output_schema = match projection {
+ None => schema.clone(),
+ Some(projection) => Arc::new(schema.project(projection).unwrap()),
+ };
+
+ Self {
+ snapshot_id,
+ output_schema,
+ column_names: get_column_names(schema, projection),
+ predicates: convert_filters_to_predicate(filters),
+ }
+ }
+
+ pub(crate) fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+
+ pub(crate) fn output_schema(&self) -> ArrowSchemaRef {
+ self.output_schema.clone()
+ }
+
+ pub(crate) fn column_names(&self) -> Option<&[String]> {
+ self.column_names.as_deref()
+ }
+
+ pub(crate) fn predicates(&self) -> Option<&Predicate> {
+ self.predicates.as_ref()
+ }
+}
+
+/// Result of eager scan planning: the [`TableScan`] that planned the file scan
+/// tasks, alongside those tasks grouped per output partition.
+#[derive(Debug)]
+pub(crate) struct EagerScanPlan {
+ /// The [`TableScan`] used to plan `task_groups`. Retained so that every output
+ /// partition builds its reader from this same scan, instead of rebuilding a
+ /// throwaway `TableScan` on each `execute()` call.
+ table_scan: Arc<TableScan>,
+ /// Planned file scan tasks, one group per output partition.
+ task_groups: Vec<Arc<[FileScanTask]>>,
+}
+
+impl EagerScanPlan {
+ /// Number of output partitions, i.e. the number of task groups.
+ pub(crate) fn partition_count(&self) -> usize {
+ self.task_groups.len()
+ }
+
+ /// Total number of planned file scan tasks across all partitions.
+ pub(crate) fn task_count(&self) -> usize {
+ self.task_groups.iter().map(|group| group.len()).sum()
+ }
+
+ /// Returns the task group assigned to `partition`, or `None` if out of range.
+ pub(crate) fn task_group(&self, partition: usize) -> Option<Arc<[FileScanTask]>> {
+ self.task_groups.get(partition).cloned()
+ }
+
+ /// Returns an [`ArrowReaderBuilder`] configured for this scan.
+ ///
+ /// This deliberately routes through [`TableScan::arrow_reader_builder`] rather
+ /// than constructing an [`ArrowReaderBuilder`] directly: it keeps the reader
+ /// settings (batch size, row group filtering, row selection) sourced from the
+ /// same place as the lazy path's `TableScan::to_arrow`, so the two scan paths
+ /// cannot silently drift apart.
+ pub(crate) fn arrow_reader_builder(&self) -> ArrowReaderBuilder {
+ self.table_scan.arrow_reader_builder()
+ }
+}
+
+pub(crate) async fn plan_eager_scan(
+ table: &Table,
+ scan_config: &IcebergScanConfig,
+ target_partitions: usize,
+) -> DFResult<EagerScanPlan> {
+ // Do not cache planned FileScanTasks in the provider in v1. They are query-specific
+ // because projection, predicate binding, snapshot schema, and delete planning can differ
+ // between scans. Catalog-backed providers also need fresh metadata on each scan.
+ // TODO: Revisit provider-level caching for static tables with a precise cache key.
+ let table_scan = Arc::new(build_table_scan(table, scan_config)?);
+
+ let tasks: Vec<FileScanTask> = table_scan
+ .plan_files()
+ .await
+ .map_err(to_datafusion_error)?
+ .try_collect::<Vec<_>>()
+ .await
+ .map_err(to_datafusion_error)?;
+
+ let task_groups = group_file_scan_tasks_round_robin(tasks, target_partitions)
+ .into_iter()
+ .map(Arc::<[FileScanTask]>::from)
+ .collect();
+
+ Ok(EagerScanPlan {
+ table_scan,
+ task_groups,
+ })
+}
+
+fn get_column_names(
+ schema: ArrowSchemaRef,
+ projection: Option<&Vec<usize>>,
+) -> Option<Vec<String>> {
+ projection.map(|v| {
+ v.iter()
+ .map(|p| schema.field(*p).name().clone())
+ .collect::<Vec<String>>()
+ })
+}
+
+/// Groups file scan tasks into `target_partitions` groups using a naive
+/// round-robin assignment. Non-empty groups are bounded by `tasks.len()`.
+// TODO: Replace this naive round-robin grouping with size-based grouping once the
+// first parallel scan path is stable. Keep this v1 simple and deterministic.
+fn group_file_scan_tasks_round_robin(
Review Comment:
Every test that exercises this function uses a task count equal to (or clamped down to) the partition count, so each group ends up with exactly one task. Would it be worth adding a direct `#[test]` here for an uneven split, e.g. 5 tasks over 3 partitions landing as `[2, 2, 1]`? It's a small pure function with no I/O, so a unit test seems cheap, and right now nothing would catch a change that preserves total task/row counts but breaks the round-robin balance itself.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3605669905
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
Review Comment:
`IcebergTableScan` is a leaf node (`children()` returns `vec![]` at `:117-119`), so any non-empty `children` here is a caller error. Right now it is dropped instead of surfaced. This is the same issue @timsaucer raised on the new scan node in #2298, which you fixed there with a `DataFusionError::Internal` matching `IcebergCommitExec`, and you noted `IcebergTableScan::with_new_children` had the same latent bug. Since this PR is the one that keeps and extends this node, it seems like the right place to close it:
```rust
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn ExecutionPlan>>,
) -> DFResult<Arc<dyn ExecutionPlan>> {
if !children.is_empty() {
return Err(DataFusionError::Internal(
"IcebergTableScan is a leaf node and cannot have children".to_string(),
));
}
Ok(self)
}
```
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
Review Comment:
Each `execute(partition)` call gets its own `remaining = limit`, so with N output partitions the scan can emit up to `N * limit` rows before DataFusion's top-level limit trims to `limit`. Results stay correct because a `GlobalLimitExec` sits above the scan, but each partition reads up to `limit` rows independently, so this is looser than the single-partition lazy path. Either clamp the per-partition read, or add a one-line comment noting the top-level limit makes the loose per-partition bound acceptable, so the intent is explicit.
##########
crates/integrations/datafusion/tests/integration_datafusion_test.rs:
##########
@@ -95,6 +99,237 @@ fn get_table_creation(
Ok(creation)
}
+async fn get_multi_file_table_context(
+ namespace_name: &str,
+ table_name: &str,
+ data_file_count: usize,
+) -> Result<(Arc<MemoryCatalog>, NamespaceIdent, String)> {
+ let iceberg_catalog = Arc::new(get_iceberg_catalog().await);
+ let namespace = NamespaceIdent::new(namespace_name.to_string());
+ set_test_namespace(&iceberg_catalog, &namespace).await?;
+
+ let creation = get_table_creation(temp_path(), table_name, None)?;
+ iceberg_catalog.create_table(&namespace, creation).await?;
+
+ let write_ctx = SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(data_file_count),
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("foo1", DataType::Int32, false),
+ Field::new("foo2", DataType::Utf8, false),
+ ]));
+
+ let batches: Vec<RecordBatch> = (1..=data_file_count as i32)
+ .map(|idx| {
+ RecordBatch::try_new(arrow_schema.clone(), vec![
+ Arc::new(Int32Array::from(vec![idx])) as ArrayRef,
+ Arc::new(StringArray::from(vec![format!("row-{idx}")])) as ArrayRef,
+ ])
+ })
+ .collect::<std::result::Result<_, _>>()?;
+
+ let partitions = batches.into_iter().map(|batch| vec![batch]).collect();
+ let source_table = Arc::new(MemTable::try_new(arrow_schema, partitions).unwrap());
+ write_ctx
+ .register_table("source_table", source_table)
+ .unwrap();
+
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(iceberg_catalog.clone()).await?);
+ write_ctx.register_catalog("catalog", catalog);
+
+ let insert_sql =
+ format!("INSERT INTO catalog.{namespace_name}.{table_name} SELECT * FROM source_table");
+ let batches = write_ctx
+ .sql(&insert_sql)
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+
+ let rows_inserted = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .unwrap();
+ assert_eq!(rows_inserted.value(0), data_file_count as u64);
+
+ Ok((iceberg_catalog, namespace, table_name.to_string()))
+}
+
+async fn get_read_context(
+ catalog: Arc<MemoryCatalog>,
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> Result<SessionContext> {
+ let ctx = SessionContext::new_with_config(read_session_config(
+ target_partitions,
+ enable_eager_scan_planning,
+ ));
+ let catalog = Arc::new(IcebergCatalogProvider::try_new(catalog).await?);
+ ctx.register_catalog("catalog", catalog);
+ Ok(ctx)
+}
+
+fn read_session_config(
+ target_partitions: usize,
+ enable_eager_scan_planning: Option<bool>,
+) -> SessionConfig {
+ let config = SessionConfig::new().with_target_partitions(target_partitions);
+
+ match enable_eager_scan_planning {
+ Some(enabled) => {
+ let mut iceberg_config = IcebergDataFusionConfig::default();
+ iceberg_config.enable_eager_scan_planning = enabled;
+ config.with_option_extension(iceberg_config)
+ }
+ None => config,
+ }
+}
+
+async fn scan_partition_count(
+ ctx: &SessionContext,
+ namespace: &NamespaceIdent,
+ table_name: &str,
+) -> usize {
+ let provider = ctx.catalog("catalog").unwrap();
+ let namespace_name = &namespace[0];
+ let schema = provider.schema(namespace_name).unwrap();
+ let table = schema.table(table_name).await.unwrap().unwrap();
+
+ let state = ctx.state();
+ let plan = table.scan(&state, None, &[], None).await.unwrap();
+ plan.downcast_ref::<IcebergTableScan>()
+ .expect("Expected IcebergTableScan");
+ plan.properties().output_partitioning().partition_count()
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_produces_multiple_partitions() -> Result<()> {
+ let data_file_count = 3;
+ // Ask for more partitions than files to verify scan planning does not expose empty partitions.
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_partitions",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(true)).await?;
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_file_scan_defaults_to_single_lazy_partition() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) = get_multi_file_table_context(
+ "test_multi_file_scan_default_lazy",
+ "my_table",
+ data_file_count,
+ )
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, None).await?;
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, 1);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_set_enable_eager_scan_planning() -> Result<()> {
+ let data_file_count = 3;
+ let target_partitions = data_file_count + 1;
+ let (iceberg_catalog, namespace, table_name) =
+ get_multi_file_table_context("test_set_eager_scan_planning", "my_table", data_file_count)
+ .await?;
+ let ctx = get_read_context(iceberg_catalog, target_partitions, Some(false)).await?;
+
+ ctx.sql("SET iceberg.enable_eager_scan_planning = true")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let actual_partition_count = scan_partition_count(&ctx, &namespace, &table_name).await;
+
+ assert_eq!(actual_partition_count, data_file_count);
+
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_multi_partition_scan_matches_single_partition_results() -> Result<()> {
Review Comment:
`test_multi_partition_scan_matches_single_partition_results` is a good check that single- and multi-partition scans return identical rows. But `get_multi_file_table_context` writes one row per file (`:125`, `Int32Array::from(vec![idx])`), so every file is a single row group with no filtering opportunity. A divergence in `row_group_filtering_enabled` or `batch_size` between the eager and lazy paths (comment 2) would not change the result here, so the test would still pass. A file with multiple row groups (and ideally a predicate that prunes some of them) would turn this into a real guard on the eager/lazy read equivalence.
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,16 +131,57 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>> = match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) = file_task_groups.get(partition).cloned() else {
+ return Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx| Ok(file_task_group[idx].clone())),
+ ));
+ let stream = self
+ .table
+ .reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency via output
+ // partitions. Match DataFusion's FileStream model, where each
+ // output partition owns one ScanState; keep one data file in
+ // flight per output partition here.
+ // https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader can accept shared tasks.
+ .read(tasks)
+ .map_err(to_datafusion_error)?
+ .stream()
+ .map_err(to_datafusion_error);
Review Comment:
The eager arm builds its reader from `self.table.reader_builder()` and overrides only `.with_data_file_concurrency_limit(1)`:
```rust
let stream = self
.table
.reader_builder()
.with_data_file_concurrency_limit(1)
.build()
.read(tasks)
...
```
That inherits `ArrowReaderBuilder::new` defaults (`crates/iceberg/src/arrow/reader/mod.rs:65-73`): `row_group_filtering_enabled = true`, `row_selection_enabled = false`, `batch_size = None`. The lazy arm at `:174-178` reaches the reader through `TableScan::to_arrow`, which sources those same three fields from `TableScanBuilder::new` defaults (`crates/iceberg/src/scan/mod.rs:77-81`) and applies them at `crates/iceberg/src/scan/mod.rs:487-492`.
So one conceptual "how to read Iceberg data files" decision is now fed by two default sources in two different crates, with nothing pinning their equivalence. They match today, but if a reader default is flipped in one struct, the eager and lazy paths diverge silently in pushdown/read behavior. This split is also what introduced the divergence: #2298's eager path went through `TableScan::read` and shared one config source, and dropping that (consistent with the #2298 discussion about not making `read` public) moved the eager arm onto `ArrowReaderBuilder` defaults directly.
Two options, in preference order:
- Thread the scan's own reader config into the eager arm so `concurrency = 1` is the only intentional eager/lazy difference. This keeps a single source of truth for the read settings.
- If that is out of scope, at minimum add a comment here stating the eager path intentionally mirrors `to_arrow`'s reader defaults, and back it with an equivalence test over a multi-row-group file.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-6373428606
@alexanderbianchi / @toutane -- shall we close this PR then?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-4845184529
👋 cc @mbutrovich @timsaucer
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-4935531560
Thanks @mbutrovich, addressed all four comments in https://github.com/apache/iceberg-rust/pull/2671/commits/ef774e837431bedccd3644f934485af29b881bc3
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5688717997
Hi @toutane, thanks for this work — I've been building on top of it for a follow-up ([sort-order reporting](https://github.com/apache/iceberg-rust/issues/3126)), so I rebased this branch onto current `main` to test it and found two things worth flagging before merge, both caused by `main` moving forward since your last push rather than anything in this PR's own design:
1. `main` picked up #3091 (`FileScanTask` fields went private behind a serde adapter, `build()` now returns `Result<FileScanTask>`). Your round-robin test in `scan_planning.rs` builds a `FileScanTask` with `.build()` and no `.unwrap()`, which no longer compiles.
2. `main` also picked up a DataFusion bump to 55.0, which deprecated `ExecutionPlan::with_new_children` in favor of `replace_children`/`ReplaceChildrenOptions`. `test_iceberg_table_scan_rejects_non_empty_children` calls `.with_new_children()` directly, which now trips `clippy -D warnings`.
Both are one-line fixes, but you'll hit them as soon as you rebase onto current `main`. Happy to share the exact diffs if useful.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "anuragmantri (via GitHub)" <gi...@apache.org>.
anuragmantri commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5486558618
Thanks for this work @toutane. Eager planning is what I needed for my work on using the sort orders in https://github.com/apache/iceberg-rust/issues/3126
I did a review and it looks good to my untrained eyes (I'm new to Rust and this repo).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "toutane (via GitHub)" <gi...@apache.org>.
toutane commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-5906978591
> Hi @toutane, I've been building on top of it for a follow-up ([sort-order reporting](https://github.com/apache/iceberg-rust/issues/3126)), so I rebased this branch onto current `main` to test it and found two things.
>
> 1. `main` picked up [feat(scan): make FileScanTask serializable #3091](https://github.com/apache/iceberg-rust/pull/3091) (`FileScanTask` fields went private behind a serde adapter, `build()` now returns `Result<FileScanTask>`). Your round-robin test in `scan_planning.rs` builds a `FileScanTask` with `.build()` and no `.unwrap()`, which no longer compiles.
> 2. `main` also picked up a DataFusion bump to 55.0, which deprecated `ExecutionPlan::with_new_children` in favor of `replace_children`/`ReplaceChildrenOptions`. `test_iceberg_table_scan_rejects_non_empty_children` calls `.with_new_children()` directly, which now trips `clippy -D warnings`.
>
> Both are one-line fixes, but you'll hit them as soon as you rebase onto current `main`.
Hey @anuragmantri, thanks for your comment and for your suggestions! It's so nice to see that you built something on top of it.
I will rebase this branch on top of `main` and apply the fixes you mentioned - I will probably do it directly into the new repository.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "alexanderbianchi (via GitHub)" <gi...@apache.org>.
alexanderbianchi commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-6129421513
https://github.com/apache/datafusion-iceberg/pull/67 FYI i'm giving this a crack in this draft PR, with the public API's
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(datafusion): Add opt-in eager file scan planning with output partitioning [iceberg-rust]
Posted by "alexanderbianchi (via GitHub)" <gi...@apache.org>.
alexanderbianchi commented on PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#issuecomment-6373609037
Yes, i'd say so
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org