From da7f30e7fc7899e26520a8b17246cf595e774a57 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 12:17:13 +0300 Subject: [PATCH 1/5] Add metadata-only Iceberg truncate snapshots --- iceberg-rust/src/table/transaction/mod.rs | 38 +++- .../src/table/transaction/operation.rs | 175 ++++++++++++++++++ 2 files changed, 212 insertions(+), 1 deletion(-) diff --git a/iceberg-rust/src/table/transaction/mod.rs b/iceberg-rust/src/table/transaction/mod.rs index 18705492..c9782d10 100644 --- a/iceberg-rust/src/table/transaction/mod.rs +++ b/iceberg-rust/src/table/transaction/mod.rs @@ -48,8 +48,9 @@ pub(crate) static UPDATE_PROPERTIES_INDEX: usize = 6; pub(crate) static SET_SNAPSHOT_REF_INDEX: usize = 7; pub(crate) static EXPIRE_SNAPSHOTS_INDEX: usize = 8; pub(crate) static REPLACE_SORT_ORDER_INDEX: usize = 9; +pub(crate) static TRUNCATE_INDEX: usize = 10; -pub(crate) static NUM_OPERATIONS: usize = 10; +pub(crate) static NUM_OPERATIONS: usize = 11; fn reject_spec_change_with_file_write( operations: &[Option], @@ -342,6 +343,16 @@ impl<'table> TableTransaction<'table> { } self } + + /// Replace the current table contents with an empty snapshot. The parent + /// snapshot and V3 row ID high-water mark are retained for change tracking. + #[must_use] + pub fn truncate(mut self) -> Self { + self.operations[TRUNCATE_INDEX] = Some(Operation::Truncate { + branch: self.branch.clone(), + }); + self + } /// Replaces all data files in the table with new ones /// /// This operation removes all existing data files and replaces them with the provided @@ -551,6 +562,21 @@ impl<'table> TableTransaction<'table> { &self.operations, self.table.metadata().default_spec_id, )?; + if self.operations[TRUNCATE_INDEX].is_some() + && [ + APPEND_INDEX, + APPEND_SEQUENCE_GROUPS_INDEX, + REPLACE_INDEX, + OVERWRITE_INDEX, + ] + .iter() + .any(|&index| self.operations[index].is_some()) + { + return Err(Error::NotSupported( + "TRUNCATE cannot be combined with another data write in one Iceberg transaction" + .to_owned(), + )); + } let catalog = self.table.catalog(); let identifier = self.table.identifier.clone(); let base_metadata = self.table.metadata(); @@ -567,6 +593,16 @@ impl<'table> TableTransaction<'table> { }], Vec::new(), ); + if self.operations[TRUNCATE_INDEX].is_some() { + // Snapshot sequence numbers are table-global, even when the target + // ref is not. Reject commits based on another ref's stale state. + requirements.extend(base_metadata.refs.iter().map(|(name, reference)| { + TableRequirement::AssertRefSnapshotId { + r#ref: name.clone(), + snapshot_id: Some(reference.snapshot_id), + } + })); + } for operation in self.operations.into_iter().flatten() { let metadata = staged_metadata.as_ref().unwrap_or(base_metadata); let (requirement, update) = operation diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index 0a85cde6..3e037dd1 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -103,6 +103,8 @@ pub enum Operation { files_to_overwrite: HashMap>, additional_summary: Option>, }, // /// Remove or replace rows in existing data files + /// Commit an empty snapshot without reading or rewriting existing manifests. + Truncate { branch: Option }, // NewRowDelta, // /// Delete files in the table and commit // NewDelete, @@ -140,6 +142,73 @@ impl Operation { } match self { + Operation::Truncate { branch } => { + let old_snapshot = table_metadata.current_snapshot(branch.as_deref())?; + let snapshot_id = generate_snapshot_id(); + let location = new_manifest_list_location( + &table_metadata.location, + snapshot_id, + 0, + &uuid::Uuid::new_v4().to_string(), + ); + let schema = match table_metadata.format_version { + FormatVersion::V1 => manifest_list_schema_v1(), + FormatVersion::V2 => manifest_list_schema_v2(), + FormatVersion::V3 => manifest_list_schema_v3(), + }; + let bytes = apache_avro::Writer::new(schema, Vec::new()).into_inner()?; + object_store + .put(&strip_prefix(&location).into(), bytes.into()) + .await?; + + let mut summary = update_snapshot_summary( + None, + std::iter::empty::<&DataFile>(), + std::iter::empty::<&DataFile>(), + ); + summary.insert( + "iceberg-rust.metadata-only-truncate".to_owned(), + "true".to_owned(), + ); + let mut builder = SnapshotBuilder::default(); + builder + .with_snapshot_id(snapshot_id) + .with_sequence_number(table_metadata.last_sequence_number + 1) + .with_schema_id(*table_metadata.current_schema()?.schema_id()) + .with_manifest_list(location) + .with_summary(Summary { + operation: SnapshotOperation::Overwrite, + other: summary, + }); + if let Some(snapshot) = old_snapshot { + builder.with_parent_snapshot_id(*snapshot.snapshot_id()); + } + if table_metadata.format_version == FormatVersion::V3 { + apply_v3_row_lineage( + &mut builder, + table_metadata, + Some(table_metadata.next_row_id), + )?; + } + let snapshot = builder.build()?; + let ref_name = branch.unwrap_or_else(|| "main".to_owned()); + Ok(( + Some(TableRequirement::AssertRefSnapshotId { + r#ref: ref_name.clone(), + snapshot_id: old_snapshot.map(|snapshot| *snapshot.snapshot_id()), + }), + vec![ + TableUpdate::AddSnapshot { snapshot }, + TableUpdate::SetSnapshotRef { + ref_name, + snapshot_reference: SnapshotReference { + snapshot_id, + retention: SnapshotRetention::default(), + }, + }, + ], + )) + } Operation::AppendSequenceGroups { branch, sequence_groups, @@ -1429,6 +1498,112 @@ mod tests { .unwrap() } + #[tokio::test] + async fn truncate_creates_snapshot_for_new_table() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.format_version = FormatVersion::V3; + let store = Arc::new(InMemory::new()); + let (requirement, updates) = Operation::Truncate { branch: None } + .execute(&metadata, store.clone()) + .await + .unwrap(); + assert_eq!( + requirement, + Some(TableRequirement::AssertRefSnapshotId { + r#ref: "main".to_owned(), + snapshot_id: None, + }) + ); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + assert_eq!(*snapshot.parent_snapshot_id(), None); + assert_eq!(*snapshot.first_row_id(), Some(0)); + assert_eq!(*snapshot.added_rows(), Some(0)); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!( + ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .count(), + 0 + ); + } + + #[tokio::test] + async fn truncate_writes_empty_manifest_list_for_v1_and_v2() { + for version in [FormatVersion::V1, FormatVersion::V2] { + let mut metadata = sample_metadata(&[(101, 1)], Some(101), &[("main", 101)]); + metadata.format_version = version; + let store = Arc::new(InMemory::new()); + let (_, updates) = Operation::Truncate { branch: None } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + assert_eq!(*snapshot.parent_snapshot_id(), Some(101)); + assert_eq!(*snapshot.first_row_id(), None); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!( + ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .count(), + 0 + ); + } + } + + #[tokio::test] + async fn truncate_preserves_v3_lineage_without_reading_old_manifests() { + let mut metadata = sample_metadata(&[(101, 1)], Some(101), &[("main", 101)]); + metadata.format_version = FormatVersion::V3; + metadata.next_row_id = 44; + let store = Arc::new(InMemory::new()); + // The old snapshot's manifest list is deliberately absent from the store. + let (requirement, updates) = Operation::Truncate { branch: None } + .execute(&metadata, store.clone()) + .await + .unwrap(); + assert_eq!( + requirement, + Some(TableRequirement::AssertRefSnapshotId { + r#ref: "main".to_owned(), + snapshot_id: Some(101), + }) + ); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + assert_eq!(*snapshot.parent_snapshot_id(), Some(101)); + assert_eq!(*snapshot.first_row_id(), Some(44)); + assert_eq!(*snapshot.added_rows(), Some(0)); + assert_eq!(metadata.next_row_id, 44); + assert_eq!(snapshot.summary().operation, SnapshotOperation::Overwrite); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!( + ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .count(), + 0 + ); + } + fn partitioned_data_file(path: &str, partition_value: Value) -> DataFile { partitioned_data_file_with_value(path, Some(partition_value)) } From fb02df59d70ef712a1687cc5c05eba1f1a2e6f92 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 12:39:11 +0300 Subject: [PATCH 2/5] Reject stale snapshot sequence numbers during commits --- iceberg-rust/src/catalog/commit.rs | 5 +++ .../src/table/transaction/operation.rs | 39 +++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/iceberg-rust/src/catalog/commit.rs b/iceberg-rust/src/catalog/commit.rs index 7d22c7a2..a2ae73b0 100644 --- a/iceberg-rust/src/catalog/commit.rs +++ b/iceberg-rust/src/catalog/commit.rs @@ -491,6 +491,11 @@ pub fn apply_table_updates( } } TableUpdate::AddSnapshot { snapshot } => { + if metadata.format_version != FormatVersion::V1 + && *snapshot.sequence_number() <= metadata.last_sequence_number + { + return Err(Error::CommitConflict(metadata.table_uuid.to_string())); + } if metadata.format_version == FormatVersion::V3 { let first_row_id = (*snapshot.first_row_id()).ok_or_else(|| { Error::InvalidFormat("v3 snapshot first-row-id".to_string()) diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index 3e037dd1..18f5a5eb 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -1604,6 +1604,45 @@ mod tests { ); } + #[tokio::test] + async fn stale_truncate_rejects_snapshot_added_on_new_branch() { + let base = sample_metadata(&[(101, 1)], Some(101), &[("main", 101)]); + let store = Arc::new(InMemory::new()); + let (requirement, updates) = Operation::Truncate { branch: None } + .execute(&base, store) + .await + .unwrap(); + + let mut current = base.clone(); + let audit = SnapshotBuilder::default() + .with_snapshot_id(202) + .with_sequence_number(2) + .with_manifest_list("audit-manifest.avro".to_owned()) + .with_summary(Summary { + operation: SnapshotOperation::Append, + other: HashMap::new(), + }) + .build() + .unwrap(); + current.snapshots.insert(202, audit); + current.refs.insert( + "audit".to_owned(), + SnapshotReference { + snapshot_id: 202, + retention: SnapshotRetention::default(), + }, + ); + current.last_sequence_number = 2; + assert!(crate::catalog::commit::check_table_requirements( + std::slice::from_ref(&requirement.unwrap()), + ¤t, + )); + assert!(matches!( + crate::catalog::commit::apply_table_updates(&mut current, updates), + Err(Error::CommitConflict(_)) + )); + } + fn partitioned_data_file(path: &str, partition_value: Value) -> DataFile { partitioned_data_file_with_value(path, Some(partition_value)) } From 361fc645b72da2c59636ce15e9f9926e486048a8 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 13:06:35 +0300 Subject: [PATCH 3/5] Align historical Iceberg struct fields by ID --- datafusion_iceberg/src/table/mod.rs | 27 ++- .../src/variant_schema_adapter.rs | 224 +++++++++++++++++- 2 files changed, 231 insertions(+), 20 deletions(-) diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index 67cb62f6..c0393de8 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -20,6 +20,7 @@ use iceberg_rust::file_format::parquet::{ }; use iceberg_rust::object_store::Bucket; use iceberg_rust::spec::partition::BoundPartitionField; +use iceberg_rust::spec::schema::Schema as IcebergSchema; use iceberg_rust::spec::table_metadata::{ self, TableMetadata, WRITE_METADATA_METRICS_DISTINCT_COUNTS_ENABLED, WRITE_OBJECT_STORAGE_ENABLED, @@ -309,6 +310,10 @@ pub struct DataFusionTableConfig { /// Intended for session-local overlays that replace uncommitted files. #[builder(default)] excluded_data_file_paths: Arc>, + /// Read a historical snapshot using this Iceberg schema, matching nested + /// fields by ID across schema evolution rather than casting by name. + #[builder(default)] + read_schema: Option, } impl DataFusionTable { @@ -329,9 +334,13 @@ impl DataFusionTable { ) -> Self { let schema = match &tabular { Tabular::Table(table) => { - let schema = end - .and_then(|snapshot_id| table.metadata().schema(snapshot_id).ok().cloned()) - .unwrap_or_else(|| table.current_schema().unwrap().clone()); + let schema = config + .as_ref() + .and_then(|config| config.read_schema.as_ref()) + .or_else(|| { + end.and_then(|snapshot_id| table.metadata().schema(snapshot_id).ok()) + }) + .unwrap_or_else(|| table.current_schema().unwrap()); let mut builder = SchemaBuilder::from(TryInto::::try_into(schema.fields()).unwrap()); if config @@ -640,10 +649,14 @@ async fn table_scan( filters: &[Expr], limit: Option, ) -> Result, DataFusionError> { - let schema = snapshot_range - .1 - .and_then(|snapshot_id| table.metadata().schema(snapshot_id).ok().cloned()) - .unwrap_or_else(|| table.current_schema().unwrap().clone()); + let schema = config + .and_then(|config| config.read_schema.as_ref()) + .or_else(|| { + snapshot_range + .1 + .and_then(|snapshot_id| table.metadata().schema(snapshot_id).ok()) + }) + .unwrap_or_else(|| table.current_schema().unwrap()); // Create a unique URI for this particular object store let object_store_url = fake_object_store_url(&table.metadata().location); diff --git a/datafusion_iceberg/src/variant_schema_adapter.rs b/datafusion_iceberg/src/variant_schema_adapter.rs index d6ae7167..8883af5b 100644 --- a/datafusion_iceberg/src/variant_schema_adapter.rs +++ b/datafusion_iceberg/src/variant_schema_adapter.rs @@ -3,12 +3,12 @@ use std::fmt::{self, Display}; use std::hash::{Hash, Hasher}; use std::sync::Arc; -use datafusion::arrow::array::{Array, ArrayRef, StructArray}; +use datafusion::arrow::array::{new_null_array, Array, ArrayRef, StructArray}; use datafusion::arrow::compute::cast; -use datafusion::arrow::datatypes::{DataType, FieldRef, Schema, SchemaRef}; +use datafusion::arrow::datatypes::{DataType, FieldRef, Fields, Schema, SchemaRef}; use datafusion::arrow::record_batch::RecordBatch; use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode}; -use datafusion::common::{internal_err, Result}; +use datafusion::common::{exec_err, internal_err, Result}; use datafusion::physical_expr::utils::collect_columns; use datafusion::physical_expr_adapter::{ DefaultPhysicalExprAdapter, PhysicalExprAdapter, PhysicalExprAdapterFactory, @@ -124,18 +124,28 @@ impl IcebergPhysicalExprAdapter { ))) } - fn rewrite_renamed_column(&self, column: &Column) -> Result>> { + fn rewrite_field_id_column(&self, column: &Column) -> Result>> { let Some((logical_field, physical_index, physical_field)) = self.field_id_mapping(column.name()) else { return Ok(None); }; - if logical_field.name() == physical_field.name() { + let nested_struct_change = matches!(logical_field.data_type(), DataType::Struct(_)) + && matches!(physical_field.data_type(), DataType::Struct(_)) + && logical_field.data_type() != physical_field.data_type(); + if logical_field.name() == physical_field.name() && !nested_struct_change { return Ok(None); } let physical_column: Arc = Arc::new(Column::new(physical_field.name(), physical_index)); + if nested_struct_change { + return Ok(Some(Arc::new(FieldIdColumnExpr::with_struct_projection( + physical_column, + logical_field, + physical_field.data_type(), + )?))); + } let input = if logical_field.data_type() == physical_field.data_type() { physical_column } else { @@ -203,7 +213,12 @@ impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { || Self::row_lineage_field_id(column.name()).is_some() || self .field_id_mapping(column.name()) - .is_some_and(|(logical, _, physical)| logical.name() != physical.name()) + .is_some_and(|(logical, _, physical)| { + logical.name() != physical.name() + || (matches!(logical.data_type(), DataType::Struct(_)) + && matches!(physical.data_type(), DataType::Struct(_)) + && logical.data_type() != physical.data_type()) + }) }); if !requires_iceberg_rewrite { return self.default.rewrite(expr); @@ -225,7 +240,7 @@ impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { return self.rewrite_variant_column(column).map(Transformed::yes); } - if let Some(physical) = self.rewrite_renamed_column(column)? { + if let Some(physical) = self.rewrite_field_id_column(column)? { return Ok(Transformed::yes(physical)); } @@ -239,6 +254,7 @@ impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { struct FieldIdColumnExpr { input: Arc, target_field: FieldRef, + struct_projection: Option, } impl FieldIdColumnExpr { @@ -246,13 +262,34 @@ impl FieldIdColumnExpr { Self { input, target_field, + struct_projection: None, } } + + fn with_struct_projection( + input: Arc, + target_field: FieldRef, + physical_type: &DataType, + ) -> Result { + let (DataType::Struct(source), DataType::Struct(target)) = + (physical_type, target_field.data_type()) + else { + return internal_err!("Iceberg nested field projection requires struct types"); + }; + let struct_projection = StructProjection::new(source, target)?; + Ok(Self { + input, + target_field, + struct_projection: Some(struct_projection), + }) + } } impl PartialEq for FieldIdColumnExpr { fn eq(&self, other: &Self) -> bool { - self.input.eq(&other.input) && self.target_field == other.target_field + self.input.eq(&other.input) + && self.target_field == other.target_field + && self.struct_projection == other.struct_projection } } @@ -260,6 +297,7 @@ impl Hash for FieldIdColumnExpr { fn hash(&self, state: &mut H) { self.input.hash(state); self.target_field.hash(state); + self.struct_projection.hash(state); } } @@ -279,7 +317,14 @@ impl PhysicalExpr for FieldIdColumnExpr { } fn evaluate(&self, batch: &RecordBatch) -> Result { - self.input.evaluate(batch) + let value = self.input.evaluate(batch)?; + let Some(projection) = &self.struct_projection else { + return Ok(value); + }; + let ColumnarValue::Array(value) = value else { + return internal_err!("Iceberg struct column unexpectedly evaluated to a scalar"); + }; + Ok(ColumnarValue::Array(projection.apply(&value)?)) } fn return_field(&self, _input_schema: &Schema) -> Result { @@ -297,10 +342,11 @@ impl PhysicalExpr for FieldIdColumnExpr { if children.len() != 1 { return internal_err!("FieldIdColumnExpr requires exactly one child"); } - Ok(Arc::new(Self::new( - Arc::clone(&children[0]), - Arc::clone(&self.target_field), - ))) + Ok(Arc::new(Self { + input: Arc::clone(&children[0]), + target_field: Arc::clone(&self.target_field), + struct_projection: self.struct_projection.clone(), + })) } fn fmt_sql(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -308,6 +354,91 @@ impl PhysicalExpr for FieldIdColumnExpr { } } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct StructProjection { + target_fields: Fields, + children: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct StructChildProjection { + source_index: Option, + target: FieldRef, + nested: Option>, +} + +impl StructProjection { + fn new(source: &Fields, target: &Fields) -> Result { + let source_has_ids = source.iter().any(|field| field_id(field).is_some()); + let children = target + .iter() + .map(|target_field| { + let source_index = match (source_has_ids, field_id(target_field)) { + (true, Some(id)) => source.iter().position(|field| field_id(field) == Some(id)), + _ => source + .iter() + .position(|field| field.name() == target_field.name()), + }; + if source_index.is_none() && !target_field.is_nullable() { + return exec_err!( + "Required Iceberg struct field '{}' is absent from the data file", + target_field.name() + ); + } + let nested = source_index + .and_then( + |index| match (source[index].data_type(), target_field.data_type()) { + (DataType::Struct(source), DataType::Struct(target)) + if source != target => + { + Some(Self::new(source, target).map(Box::new)) + } + _ => None, + }, + ) + .transpose()?; + Ok(StructChildProjection { + source_index, + target: Arc::clone(target_field), + nested, + }) + }) + .collect::>>()?; + Ok(Self { + target_fields: target.clone(), + children, + }) + } + + fn apply(&self, input: &ArrayRef) -> Result { + let Some(input) = input.as_any().downcast_ref::() else { + return internal_err!("Iceberg nested field projection expected a struct array"); + }; + let columns = self + .children + .iter() + .map(|child| { + let Some(index) = child.source_index else { + return Ok(new_null_array(child.target.data_type(), input.len())); + }; + let source = input.column(index); + if let Some(nested) = &child.nested { + nested.apply(source) + } else if source.data_type() == child.target.data_type() { + Ok(Arc::clone(source)) + } else { + Ok(cast(source.as_ref(), child.target.data_type())?) + } + }) + .collect::>>()?; + Ok(Arc::new(StructArray::try_new( + self.target_fields.clone(), + columns, + input.nulls().cloned(), + )?)) + } +} + fn is_variant_storage(data_type: &DataType) -> bool { let DataType::Struct(fields) = data_type else { return false; @@ -701,4 +832,71 @@ mod tests { assert!(Arc::ptr_eq(&output, &names)); Ok(()) } + + #[test] + fn maps_renamed_nested_iceberg_field_by_id() -> Result<()> { + use datafusion::arrow::array::Int32Array; + + let field_id = |id: i32| [(PARQUET_FIELD_ID_META_KEY.to_owned(), id.to_string())].into(); + let physical_nested = + Field::new("old_name", DataType::Int32, false).with_metadata(field_id(3)); + let physical_other = + Field::new("old_other", DataType::Int32, false).with_metadata(field_id(4)); + let logical_nested = + Field::new("new_name", DataType::Int32, false).with_metadata(field_id(3)); + let logical_other = + Field::new("new_other", DataType::Int32, false).with_metadata(field_id(4)); + let logical_added = Field::new("added", DataType::Int32, true).with_metadata(field_id(5)); + let physical_fields = Fields::from(vec![physical_nested, physical_other]); + let physical = StructArray::try_new( + physical_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![1, 2])), + Arc::new(Int32Array::from(vec![10, 20])), + ], + None, + )?; + let physical_schema = Arc::new(Schema::new(vec![Field::new( + "obj", + DataType::Struct(physical_fields), + false, + ) + .with_metadata(field_id(2))])); + let logical_schema = Arc::new(Schema::new(vec![Field::new( + "obj", + DataType::Struct(Fields::from(vec![ + logical_other, + logical_nested, + logical_added, + ])), + false, + ) + .with_metadata(field_id(2))])); + let batch = RecordBatch::try_new(Arc::clone(&physical_schema), vec![Arc::new(physical)])?; + let adapter = IcebergPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), Arc::clone(&physical_schema))?; + let expr = adapter.rewrite(Arc::new(Column::new("obj", 0)))?; + let ColumnarValue::Array(output) = expr.evaluate(&batch)? else { + return internal_err!("nested field unexpectedly evaluated to a scalar"); + }; + let output = output + .as_any() + .downcast_ref::() + .expect("struct output"); + assert_eq!(output.data_type(), logical_schema.field(0).data_type()); + let other = output + .column(0) + .as_any() + .downcast_ref::() + .expect("Int32 child"); + assert_eq!(other.values().as_ref(), &[10, 20]); + let values = output + .column(1) + .as_any() + .downcast_ref::() + .expect("Int32 child"); + assert_eq!(values.values().as_ref(), &[1, 2]); + assert_eq!(output.column(2).null_count(), 2); + Ok(()) + } } From cc397736506df7994abe742b2890acc25fe5cd60 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 13:18:40 +0300 Subject: [PATCH 4/5] Preserve Iceberg field identity through nested containers --- .../src/variant_schema_adapter.rs | 463 ++++++++++++++++-- 1 file changed, 419 insertions(+), 44 deletions(-) diff --git a/datafusion_iceberg/src/variant_schema_adapter.rs b/datafusion_iceberg/src/variant_schema_adapter.rs index 8883af5b..7b52722e 100644 --- a/datafusion_iceberg/src/variant_schema_adapter.rs +++ b/datafusion_iceberg/src/variant_schema_adapter.rs @@ -3,17 +3,20 @@ use std::fmt::{self, Display}; use std::hash::{Hash, Hasher}; use std::sync::Arc; -use datafusion::arrow::array::{new_null_array, Array, ArrayRef, StructArray}; +use datafusion::arrow::array::{ + new_null_array, Array, ArrayRef, FixedSizeListArray, LargeListArray, ListArray, MapArray, + StructArray, +}; use datafusion::arrow::compute::cast; use datafusion::arrow::datatypes::{DataType, FieldRef, Fields, Schema, SchemaRef}; use datafusion::arrow::record_batch::RecordBatch; use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode}; -use datafusion::common::{exec_err, internal_err, Result}; +use datafusion::common::{exec_err, internal_err, Result, ScalarValue}; use datafusion::physical_expr::utils::collect_columns; use datafusion::physical_expr_adapter::{ DefaultPhysicalExprAdapter, PhysicalExprAdapter, PhysicalExprAdapterFactory, }; -use datafusion::physical_plan::expressions::{CastExpr, Column}; +use datafusion::physical_plan::expressions::{CastExpr, Column, Literal}; use datafusion::physical_plan::PhysicalExpr; use datafusion_expr::ColumnarValue; use iceberg_rust::spec::arrow::schema::PARQUET_FIELD_ID_META_KEY; @@ -87,6 +90,16 @@ impl IcebergPhysicalExprAdapter { Some((Arc::new(logical.clone()), physical_index, physical)) } + fn reused_name_with_different_id(&self, name: &str) -> Option { + let logical = self.logical_file_schema.field_with_name(name).ok()?; + let logical_id = field_id(logical)?; + if self.physical_field_by_id(logical_id).is_some() { + return None; + } + let physical = self.physical_file_schema.field_with_name(name).ok()?; + (field_id(physical).is_some_and(|id| id != logical_id)).then(|| Arc::new(logical.clone())) + } + fn is_logical_variant_column(&self, name: &str) -> bool { self.logical_file_schema .field_with_name(name) @@ -130,17 +143,16 @@ impl IcebergPhysicalExprAdapter { else { return Ok(None); }; - let nested_struct_change = matches!(logical_field.data_type(), DataType::Struct(_)) - && matches!(physical_field.data_type(), DataType::Struct(_)) - && logical_field.data_type() != physical_field.data_type(); - if logical_field.name() == physical_field.name() && !nested_struct_change { + let nested_change = + needs_nested_projection(physical_field.data_type(), logical_field.data_type()); + if logical_field.name() == physical_field.name() && !nested_change { return Ok(None); } let physical_column: Arc = Arc::new(Column::new(physical_field.name(), physical_index)); - if nested_struct_change { - return Ok(Some(Arc::new(FieldIdColumnExpr::with_struct_projection( + if nested_change { + return Ok(Some(Arc::new(FieldIdColumnExpr::with_nested_projection( physical_column, logical_field, physical_field.data_type(), @@ -205,19 +217,30 @@ fn field_id(field: &datafusion::arrow::datatypes::Field) -> Option { .and_then(|id| id.parse().ok()) } +fn needs_nested_projection(source: &DataType, target: &DataType) -> bool { + source != target + && matches!( + (source, target), + (DataType::Struct(_), DataType::Struct(_)) + | (DataType::List(_), DataType::List(_)) + | (DataType::LargeList(_), DataType::LargeList(_)) + | (DataType::FixedSizeList(_, _), DataType::FixedSizeList(_, _)) + | (DataType::Map(_, _), DataType::Map(_, _)) + ) +} + impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { fn rewrite(&self, expr: Arc) -> Result> { let columns = collect_columns(&expr); let requires_iceberg_rewrite = columns.iter().any(|column| { self.is_logical_variant_column(column.name()) || Self::row_lineage_field_id(column.name()).is_some() + || self.reused_name_with_different_id(column.name()).is_some() || self .field_id_mapping(column.name()) .is_some_and(|(logical, _, physical)| { logical.name() != physical.name() - || (matches!(logical.data_type(), DataType::Struct(_)) - && matches!(physical.data_type(), DataType::Struct(_)) - && logical.data_type() != physical.data_type()) + || needs_nested_projection(physical.data_type(), logical.data_type()) }) }); if !requires_iceberg_rewrite { @@ -236,6 +259,22 @@ impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { return self.default.rewrite(expr).map(Transformed::yes); } + if let Some(field) = self.reused_name_with_different_id(column.name()) { + if !field.is_nullable() { + return exec_err!( + "Required Iceberg field '{}' is absent from the data file", + field.name() + ); + } + let value = ScalarValue::Null.cast_to(field.data_type())?; + return Ok(Transformed::yes(Arc::new(Literal::new_with_metadata( + value, + Some(datafusion::common::metadata::FieldMetadata::from( + field.as_ref(), + )), + )))); + } + if self.is_logical_variant_column(column.name()) { return self.rewrite_variant_column(column).map(Transformed::yes); } @@ -254,7 +293,7 @@ impl PhysicalExprAdapter for IcebergPhysicalExprAdapter { struct FieldIdColumnExpr { input: Arc, target_field: FieldRef, - struct_projection: Option, + nested_projection: Option, } impl FieldIdColumnExpr { @@ -262,25 +301,24 @@ impl FieldIdColumnExpr { Self { input, target_field, - struct_projection: None, + nested_projection: None, } } - fn with_struct_projection( + fn with_nested_projection( input: Arc, target_field: FieldRef, physical_type: &DataType, ) -> Result { - let (DataType::Struct(source), DataType::Struct(target)) = - (physical_type, target_field.data_type()) + let Some(nested_projection) = + NestedProjection::new(physical_type, target_field.data_type())? else { - return internal_err!("Iceberg nested field projection requires struct types"); + return internal_err!("Iceberg nested field projection requires container types"); }; - let struct_projection = StructProjection::new(source, target)?; Ok(Self { input, target_field, - struct_projection: Some(struct_projection), + nested_projection: Some(nested_projection), }) } } @@ -289,7 +327,7 @@ impl PartialEq for FieldIdColumnExpr { fn eq(&self, other: &Self) -> bool { self.input.eq(&other.input) && self.target_field == other.target_field - && self.struct_projection == other.struct_projection + && self.nested_projection == other.nested_projection } } @@ -297,7 +335,7 @@ impl Hash for FieldIdColumnExpr { fn hash(&self, state: &mut H) { self.input.hash(state); self.target_field.hash(state); - self.struct_projection.hash(state); + self.nested_projection.hash(state); } } @@ -318,11 +356,11 @@ impl PhysicalExpr for FieldIdColumnExpr { fn evaluate(&self, batch: &RecordBatch) -> Result { let value = self.input.evaluate(batch)?; - let Some(projection) = &self.struct_projection else { + let Some(projection) = &self.nested_projection else { return Ok(value); }; let ColumnarValue::Array(value) = value else { - return internal_err!("Iceberg struct column unexpectedly evaluated to a scalar"); + return internal_err!("Iceberg nested column unexpectedly evaluated to a scalar"); }; Ok(ColumnarValue::Array(projection.apply(&value)?)) } @@ -345,7 +383,7 @@ impl PhysicalExpr for FieldIdColumnExpr { Ok(Arc::new(Self { input: Arc::clone(&children[0]), target_field: Arc::clone(&self.target_field), - struct_projection: self.struct_projection.clone(), + nested_projection: self.nested_projection.clone(), })) } @@ -354,6 +392,177 @@ impl PhysicalExpr for FieldIdColumnExpr { } } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +enum NestedProjection { + Struct(StructProjection), + List { + field: FieldRef, + values: Option>, + }, + LargeList { + field: FieldRef, + values: Option>, + }, + FixedSizeList { + field: FieldRef, + size: i32, + values: Option>, + }, + Map { + field: FieldRef, + ordered: bool, + entries: Option>, + }, +} + +impl NestedProjection { + fn new(source: &DataType, target: &DataType) -> Result> { + if source == target { + return Ok(None); + } + let projection = match (source, target) { + (DataType::Struct(source), DataType::Struct(target)) => { + Self::Struct(StructProjection::new(source, target)?) + } + (DataType::List(source), DataType::List(target)) => { + verify_container_child_id(source, target)?; + Self::List { + field: Arc::clone(target), + values: Self::new(source.data_type(), target.data_type())?.map(Box::new), + } + } + (DataType::LargeList(source), DataType::LargeList(target)) => { + verify_container_child_id(source, target)?; + Self::LargeList { + field: Arc::clone(target), + values: Self::new(source.data_type(), target.data_type())?.map(Box::new), + } + } + ( + DataType::FixedSizeList(source, source_size), + DataType::FixedSizeList(target, size), + ) => { + if source_size != size { + return exec_err!("Iceberg fixed-size list width changed across schemas"); + } + verify_container_child_id(source, target)?; + Self::FixedSizeList { + field: Arc::clone(target), + size: *size, + values: Self::new(source.data_type(), target.data_type())?.map(Box::new), + } + } + (DataType::Map(source, _), DataType::Map(target, ordered)) => { + verify_container_child_id(source, target)?; + Self::Map { + field: Arc::clone(target), + ordered: *ordered, + entries: Self::new(source.data_type(), target.data_type())?.map(Box::new), + } + } + _ => return Ok(None), + }; + Ok(Some(projection)) + } + + fn apply(&self, input: &ArrayRef) -> Result { + match self { + Self::Struct(projection) => projection.apply(input), + Self::List { field, values } => { + let Some(input) = input.as_any().downcast_ref::() else { + return internal_err!("Iceberg nested projection expected a list array"); + }; + let values = project_or_cast(input.values(), field.data_type(), values.as_deref())?; + Ok(Arc::new(ListArray::try_new( + Arc::clone(field), + input.offsets().clone(), + values, + input.nulls().cloned(), + )?)) + } + Self::LargeList { field, values } => { + let Some(input) = input.as_any().downcast_ref::() else { + return internal_err!("Iceberg nested projection expected a large list array"); + }; + let values = project_or_cast(input.values(), field.data_type(), values.as_deref())?; + Ok(Arc::new(LargeListArray::try_new( + Arc::clone(field), + input.offsets().clone(), + values, + input.nulls().cloned(), + )?)) + } + Self::FixedSizeList { + field, + size, + values, + } => { + let Some(input) = input.as_any().downcast_ref::() else { + return internal_err!( + "Iceberg nested projection expected a fixed-size list array" + ); + }; + let values = project_or_cast(input.values(), field.data_type(), values.as_deref())?; + Ok(Arc::new(FixedSizeListArray::try_new_with_length( + Arc::clone(field), + *size, + values, + input.nulls().cloned(), + input.len(), + )?)) + } + Self::Map { + field, + ordered, + entries, + } => { + let Some(input) = input.as_any().downcast_ref::() else { + return internal_err!("Iceberg nested projection expected a map array"); + }; + let input_entries: ArrayRef = Arc::new(input.entries().clone()); + let projected = + project_or_cast(&input_entries, field.data_type(), entries.as_deref())?; + let Some(projected) = projected.as_any().downcast_ref::() else { + return internal_err!("Iceberg map entries must be a struct array"); + }; + Ok(Arc::new(MapArray::try_new( + Arc::clone(field), + input.offsets().clone(), + projected.clone(), + input.nulls().cloned(), + *ordered, + )?)) + } + } + } +} + +fn verify_container_child_id(source: &FieldRef, target: &FieldRef) -> Result<()> { + if let (Some(source_id), Some(target_id)) = (field_id(source), field_id(target)) { + if source_id != target_id { + return exec_err!( + "Iceberg container child '{}' changed field ID from {source_id} to {target_id}", + target.name() + ); + } + } + Ok(()) +} + +fn project_or_cast( + source: &ArrayRef, + target: &DataType, + projection: Option<&NestedProjection>, +) -> Result { + if let Some(projection) = projection { + projection.apply(source) + } else if source.data_type() == target { + Ok(Arc::clone(source)) + } else { + Ok(cast(source.as_ref(), target)?) + } +} + #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct StructProjection { target_fields: Fields, @@ -364,7 +573,7 @@ struct StructProjection { struct StructChildProjection { source_index: Option, target: FieldRef, - nested: Option>, + nested: Option>, } impl StructProjection { @@ -386,17 +595,12 @@ impl StructProjection { ); } let nested = source_index - .and_then( - |index| match (source[index].data_type(), target_field.data_type()) { - (DataType::Struct(source), DataType::Struct(target)) - if source != target => - { - Some(Self::new(source, target).map(Box::new)) - } - _ => None, - }, - ) - .transpose()?; + .map(|index| { + NestedProjection::new(source[index].data_type(), target_field.data_type()) + }) + .transpose()? + .flatten() + .map(Box::new); Ok(StructChildProjection { source_index, target: Arc::clone(target_field), @@ -422,13 +626,7 @@ impl StructProjection { return Ok(new_null_array(child.target.data_type(), input.len())); }; let source = input.column(index); - if let Some(nested) = &child.nested { - nested.apply(source) - } else if source.data_type() == child.target.data_type() { - Ok(Arc::clone(source)) - } else { - Ok(cast(source.as_ref(), child.target.data_type())?) - } + project_or_cast(source, child.target.data_type(), child.nested.as_deref()) }) .collect::>>()?; Ok(Arc::new(StructArray::try_new( @@ -833,6 +1031,27 @@ mod tests { Ok(()) } + #[test] + fn reused_iceberg_column_name_does_not_read_old_field_id() -> Result<()> { + let field_id = |id: i32| [(PARQUET_FIELD_ID_META_KEY.to_owned(), id.to_string())].into(); + let logical_schema = Arc::new(Schema::new(vec![ + Field::new("x", DataType::Int64, true).with_metadata(field_id(2)) + ])); + let physical_schema = Arc::new(Schema::new(vec![ + Field::new("x", DataType::Int64, true).with_metadata(field_id(1)) + ])); + let adapter = IcebergPhysicalExprAdapterFactory + .create(logical_schema, Arc::clone(&physical_schema))?; + let expr = adapter.rewrite(Arc::new(Column::new("x", 0)))?; + let batch = RecordBatch::try_new( + physical_schema, + vec![Arc::new(Int64Array::from(vec![Some(7), Some(8)]))], + )?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(result.null_count(), 2); + Ok(()) + } + #[test] fn maps_renamed_nested_iceberg_field_by_id() -> Result<()> { use datafusion::arrow::array::Int32Array; @@ -899,4 +1118,160 @@ mod tests { assert_eq!(output.column(2).null_count(), 2); Ok(()) } + + #[test] + fn maps_nested_iceberg_fields_inside_list_and_map_by_id() -> Result<()> { + use datafusion::arrow::array::Int32Array; + use datafusion::arrow::buffer::OffsetBuffer; + + let field_id = |id: i32| [(PARQUET_FIELD_ID_META_KEY.to_owned(), id.to_string())].into(); + let old_list_fields = Fields::from(vec![ + Field::new("old", DataType::Int32, false).with_metadata(field_id(4)), + Field::new("other", DataType::Int32, false).with_metadata(field_id(9)), + ]); + let new_list_fields = Fields::from(vec![ + Field::new("other_new", DataType::Int32, false).with_metadata(field_id(9)), + Field::new("new", DataType::Int32, false).with_metadata(field_id(4)), + ]); + let list_values = StructArray::try_new( + old_list_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![1, 2])), + Arc::new(Int32Array::from(vec![11, 22])), + ], + None, + )?; + let old_element = Arc::new( + Field::new("element", DataType::Struct(old_list_fields), false) + .with_metadata(field_id(3)), + ); + let new_element = Arc::new( + Field::new("element", DataType::Struct(new_list_fields), false) + .with_metadata(field_id(3)), + ); + let list = ListArray::try_new( + Arc::clone(&old_element), + OffsetBuffer::from_lengths([2]), + Arc::new(list_values), + None, + )?; + + let old_value_fields = Fields::from(vec![ + Field::new("old", DataType::Int32, false).with_metadata(field_id(8)), + Field::new("other", DataType::Int32, false).with_metadata(field_id(10)), + ]); + let new_value_fields = Fields::from(vec![ + Field::new("other_new", DataType::Int32, false).with_metadata(field_id(10)), + Field::new("new", DataType::Int32, false).with_metadata(field_id(8)), + ]); + let map_values = StructArray::try_new( + old_value_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![9])), + Arc::new(Int32Array::from(vec![19])), + ], + None, + )?; + let old_entries_fields = Fields::from(vec![ + Field::new("key", DataType::Utf8, false).with_metadata(field_id(6)), + Field::new("value", DataType::Struct(old_value_fields), false) + .with_metadata(field_id(7)), + ]); + let new_entries_fields = Fields::from(vec![ + Field::new("key", DataType::Utf8, false).with_metadata(field_id(6)), + Field::new("value", DataType::Struct(new_value_fields), false) + .with_metadata(field_id(7)), + ]); + let entries = StructArray::try_new( + old_entries_fields.clone(), + vec![Arc::new(StringArray::from(vec!["k"])), Arc::new(map_values)], + None, + )?; + let old_entries = Arc::new(Field::new( + "entries", + DataType::Struct(old_entries_fields), + false, + )); + let new_entries = Arc::new(Field::new( + "entries", + DataType::Struct(new_entries_fields), + false, + )); + let map = MapArray::try_new( + Arc::clone(&old_entries), + OffsetBuffer::from_lengths([1]), + entries, + None, + false, + )?; + + let physical_schema = Arc::new(Schema::new(vec![ + Field::new("items", DataType::List(old_element), false).with_metadata(field_id(2)), + Field::new("attrs", DataType::Map(old_entries, false), false) + .with_metadata(field_id(5)), + ])); + let logical_schema = Arc::new(Schema::new(vec![ + Field::new("items", DataType::List(new_element), false).with_metadata(field_id(2)), + Field::new("attrs", DataType::Map(new_entries, false), false) + .with_metadata(field_id(5)), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&physical_schema), + vec![Arc::new(list), Arc::new(map)], + )?; + let adapter = IcebergPhysicalExprAdapterFactory + .create(Arc::clone(&logical_schema), physical_schema)?; + + let list_expr = adapter.rewrite(Arc::new(Column::new("items", 0)))?; + let ColumnarValue::Array(list_output) = list_expr.evaluate(&batch)? else { + return internal_err!("list unexpectedly evaluated to a scalar"); + }; + let list_output = list_output.as_any().downcast_ref::().unwrap(); + assert_eq!(list_output.data_type(), logical_schema.field(0).data_type()); + let list_values = list_output + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(list_values.column_names(), vec!["other_new", "new"]); + let values = list_values + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), &[11, 22]); + let values = list_values + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), &[1, 2]); + + let map_expr = adapter.rewrite(Arc::new(Column::new("attrs", 1)))?; + let ColumnarValue::Array(map_output) = map_expr.evaluate(&batch)? else { + return internal_err!("map unexpectedly evaluated to a scalar"); + }; + let map_output = map_output.as_any().downcast_ref::().unwrap(); + assert_eq!(map_output.data_type(), logical_schema.field(1).data_type()); + let map_values = map_output + .entries() + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(map_values.column_names(), vec!["other_new", "new"]); + let values = map_values + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), &[19]); + let values = map_values + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), &[9]); + Ok(()) + } } From 96112c80ca1a8684cc868961f8760a155c983ac6 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 13:28:56 +0300 Subject: [PATCH 5/5] Keep borrowed scan schema free of redundant references --- datafusion_iceberg/src/table/mod.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index c0393de8..66c5cbc4 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -758,7 +758,7 @@ async fn table_scan( .metadata() .default_sort_order() .ok() - .and_then(|order| declared_output_ordering(order, &schema, &file_schema)); + .and_then(|order| declared_output_ordering(order, schema, &file_schema)); // If no projection was specified default to projecting all the fields let requested_projection = projection @@ -892,9 +892,9 @@ async fn table_scan( .try_build(physical_predicate)?; // After the first pruning stage the data_files are pruned again based on the pruning statistics in the manifest files. let files_to_prune = - pruning_predicate.prune(&PruneDataFiles::new(&schema, &arrow_schema, &data_files))?; + pruning_predicate.prune(&PruneDataFiles::new(schema, &arrow_schema, &data_files))?; - let mut statistics = statistics_from_datafiles(&schema, &data_files); + let mut statistics = statistics_from_datafiles(schema, &data_files); for _ in 0..usize::from(enable_row_id_column) + usize::from(enable_last_updated_sequence_number_column) { @@ -931,7 +931,7 @@ async fn table_scan( }); } - let mut statistics = statistics_from_datafiles(&schema, &data_files); + let mut statistics = statistics_from_datafiles(schema, &data_files); for _ in 0..usize::from(enable_row_id_column) + usize::from(enable_last_updated_sequence_number_column) { @@ -1469,7 +1469,7 @@ async fn table_scan( .as_ref() .is_some_and(|(order_id, _)| entry.data_file().sort_order_id() == &Some(*order_id)); let file = generate_partitioned_file( - &schema, + schema, &entry, last_updated_ms, include_partition_values,