This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3437-7ae9f4a44a4f8afae1ba6e7caa1456e4cde790e6 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit de0a6ac8471da43f63382d66fac27d7e56fd273e Author: Feng Zhang <[email protected]> AuthorDate: Tue Oct 27 21:41:05 2026 +0000 fix(arrow): read nested struct fields as null for files without ids or name mapping (#3437) * fix(arrow): read nested struct fields as null for files without ids or name mapping A Parquet file with no field ids and no name mapping gets fallback ids for its top-level columns only. RecordBatchTransformer then matched the id-less nested struct fields to the table's by position, so it returned the stored values. iceberg-java matches nested fields by id and reads them as absent. Id-less struct fields now take the absent-field path: identity partition constant, null, or an error for a required field. Map key and value match by position, as in iceberg-java's map readers. Closes #3436 * test(arrow): pin struct validity for id-less struct fields An id-less struct keeps the file's validity and reads its fields as null, as Spark's reader does. * Address review: map entry count error, drop redundant tests, changelog - Report a map entry field count mismatch with its own error instead of "expected struct-typed map entries". - Drop promote_idless_nested_struct_reads_null and promote_idless_map_keeps_data, covered by other tests. - Note in the changelog that tables migrated without field ids need a name mapping covering nested fields. --- CHANGELOG.md | 1 + crates/iceberg/src/arrow/reader/projection.rs | 23 +-- .../iceberg/src/arrow/record_batch_transformer.rs | 192 ++++++++------------- 3 files changed, 79 insertions(+), 137 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 728b057e7..b45216caa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). * AGS1 readers now require `StandardKeyMetadata::file_length` and reject missing or invalid lengths without falling back to a storage stat. AGS1-encrypted manifests, manifest lists, and Puffin files written by earlier development builds without this field must be rewritten using a build that can still read them before upgrading. This matches the Java client's read contract. * `iceberg_catalog_loader::BoxedCatalogBuilder` has a new required method, `with_runtime`. Types implementing `CatalogBuilder` get it through the blanket impl; types implementing `BoxedCatalogBuilder` directly (for example, a wrapper around another `Box<dyn BoxedCatalogBuilder>`) must implement it, typically by forwarding the runtime to the builder they wrap. * `Transaction::commit` now returns `ErrorKind::PreconditionFailed` for read-only tables (`Table::readonly()`, e.g. from `StaticTable::into_table` or `TableBuilder::readonly(true)`), including empty transactions, which previously returned the table unchanged. Java treats an empty transaction on a static table as a no-op. +* `ArrowReader` reads nested struct fields as null for Parquet files that have neither field ids nor a name mapping, as iceberg-java does. Earlier versions returned their stored values by position. Tables migrated without field ids need a name mapping (`schema.name-mapping.default`) that covers nested fields to read them. ## [v0.10.1] - 2026-07-28 diff --git a/crates/iceberg/src/arrow/reader/projection.rs b/crates/iceberg/src/arrow/reader/projection.rs index b76d60c85..3474b2162 100644 --- a/crates/iceberg/src/arrow/reader/projection.rs +++ b/crates/iceberg/src/arrow/reader/projection.rs @@ -1690,10 +1690,10 @@ message schema { } /// Test reading Parquet files without field IDs with nested types (struct). - /// Java's pruneColumnsFallback() projects entire top-level columns including nested content. - /// This test verifies that a top-level struct field is projected correctly with all its nested fields. + /// Java's pruneColumnsFallback() projects entire top-level columns including nested content, + /// but only top-level fields get fallback ids, so struct fields read as null. #[tokio::test] - async fn test_read_parquet_without_field_ids_with_struct() { + async fn test_read_parquet_without_field_ids_reads_struct_fields_as_null() { use arrow_array::{Int32Array, StructArray}; use arrow_schema::Fields; @@ -1706,13 +1706,13 @@ message schema { 2, "person", Type::Struct(StructType::new(vec![ - NestedField::required( + NestedField::optional( 3, "name", Type::Primitive(PrimitiveType::String), ) .into(), - NestedField::required(4, "age", Type::Primitive(PrimitiveType::Int)) + NestedField::optional(4, "age", Type::Primitive(PrimitiveType::Int)) .into(), ])), ) @@ -1807,16 +1807,9 @@ message schema { let person_array = batch.column(1).as_struct(); assert_eq!(person_array.num_columns(), 2); - - let name_array = person_array.column(0).as_string::<i32>(); - assert_eq!(name_array.value(0), "Alice"); - assert_eq!(name_array.value(1), "Bob"); - - let age_array = person_array - .column(1) - .as_primitive::<arrow_array::types::Int32Type>(); - assert_eq!(age_array.value(0), 30); - assert_eq!(age_array.value(1), 25); + assert_eq!(person_array.null_count(), 0); + assert_eq!(person_array.column(0).null_count(), 2); + assert_eq!(person_array.column(1).null_count(), 2); } /// Test reading Parquet files without field IDs with schema evolution - column added in the middle. diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs index d6d08fd80..108dd6ecc 100644 --- a/crates/iceberg/src/arrow/record_batch_transformer.rs +++ b/crates/iceberg/src/arrow/record_batch_transformer.rs @@ -270,19 +270,39 @@ impl PromotePlan { }), (DataType::Map(source_entries, _), DataType::Map(target_entries, sorted)) => { match (source_entries.data_type(), target_entries.data_type()) { - (DataType::Struct(source_fields), DataType::Struct(target_fields)) => { + // Key and value match by position, as in iceberg-java's map readers. + (DataType::Struct(source_fields), DataType::Struct(target_fields)) + if source_fields.len() == target_fields.len() => + { Ok(PromotePlan::Map { - entries: Self::build_struct_children( - source_fields, - target_fields, - snapshot_schema, - constant_fields, - )?, + entries: source_fields + .iter() + .zip(target_fields.iter()) + .enumerate() + .map(|(source_index, (source_field, target_field))| { + Ok(ChildPlan::FromSource { + source_index, + plan: Self::build( + source_field.data_type(), + target_field.data_type(), + snapshot_schema, + constant_fields, + )?, + }) + }) + .collect::<Result<_>>()?, entry_fields: target_fields.clone(), field: target_entries.clone(), sorted: *sorted, }) } + (DataType::Struct(source_fields), DataType::Struct(target_fields)) => { + Err(invalid_data!( + "map entries have {} source fields, expected {}", + source_fields.len(), + target_fields.len() + )) + } _ => Err(invalid_data!( "expected struct-typed map entries, got {source_entries:?} and {target_entries:?}" )), @@ -312,29 +332,9 @@ impl PromotePlan { return Err(invalid_data!("duplicate field id {id} in struct")); } } - // Name mapping only assigns top-level ids, so fully id-less children match - // by position when the counts line up, and `build` checks each pair. This is - // a compatibility fallback until recursive name mapping (#1845): a same-type - // reorder reads positionally. Any missing id errors instead of nulling that child. - if source_by_id.len() != source_fields.len() { - if source_by_id.is_empty() && source_fields.len() == target_fields.len() { - return source_fields - .iter() - .zip(target_fields.iter()) - .enumerate() - .map(|(source_index, (source_field, target_field))| { - Ok(ChildPlan::FromSource { - source_index, - plan: Self::build( - source_field.data_type(), - target_field.data_type(), - snapshot_schema, - constant_fields, - )?, - }) - }) - .collect(); - } + // Fully id-less children (no field ids and no name mapping) match no target + // field and read as absent, as in iceberg-java. A partial set of ids errors. + if !source_by_id.is_empty() && source_by_id.len() != source_fields.len() { return Err(invalid_data!( "cannot reconcile struct fields by id: source fields do not all have field ids" )); @@ -1427,7 +1427,7 @@ mod test { use arrow_array::types::Int32Type; use arrow_array::{ Array, ArrayRef, Date32Array, Float32Array, Float64Array, Int32Array, Int64Array, - LargeListArray, ListArray, MapArray, RecordBatch, StringArray, StructArray, + LargeListArray, ListArray, MapArray, RecordBatch, StringArray, StructArray, new_null_array, }; use arrow_buffer::{NullBuffer, OffsetBuffer}; use arrow_cast::cast; @@ -1702,16 +1702,28 @@ mod test { } #[test] - fn promote_struct_without_source_field_ids_errors() { + fn promote_struct_without_source_field_ids_reads_null() { + let nulls = NullBuffer::from(vec![true, true, false]); let source = Arc::new(StructArray::new( Fields::from(vec![Field::new("x", DataType::Int32, true)]), - vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef], - None, + vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef], + Some(nulls.clone()), )) as ArrayRef; - let err = promote(&source, &evolved_struct_type(), &empty_schema()).unwrap_err(); - assert_eq!(err.kind(), ErrorKind::DataInvalid); - assert!(err.to_string().contains("do not all have field ids")); + // Fields read as null and the struct keeps the file's validity, as Spark's reader does. + let out = promote(&source, &evolved_struct_type(), &empty_schema()).unwrap(); + let expected = StructArray::new( + Fields::from(vec![ + field_with_id("x", DataType::Int32, true, 5), + field_with_id("y", DataType::Int32, true, 6), + ]), + vec![ + new_null_array(&DataType::Int32, 3), + new_null_array(&DataType::Int32, 3), + ], + Some(nulls), + ); + assert_eq!(out.as_struct(), &expected); } #[test] @@ -1749,6 +1761,11 @@ mod test { map_of(DataType::Int64), "expected struct-typed map entries", ), + ( + map_of(unevolved_struct_type()), + map_of(evolved_struct_type()), + "map entries have 1 source fields, expected 2", + ), ] { let err = PromotePlan::build(&source, &target, &empty_schema(), &HashMap::new()).unwrap_err(); @@ -1855,64 +1872,20 @@ mod test { } #[test] - fn promote_idless_nested_struct_keeps_data() { - let inner = StructArray::new( - Fields::from(vec![Field::new("x", DataType::Int32, true)]), - vec![Arc::new(Int32Array::from(vec![10, 20, 30])) as ArrayRef], - None, - ); - let file = StructArray::new( - Fields::from(vec![Field::new("inner", inner.data_type().clone(), true)]), - vec![Arc::new(inner) as ArrayRef], - None, - ); - let out = transform_top_level( - NestedField::optional( - 2, - "s", - Type::Struct(StructType::new(vec![ - NestedField::optional( - 3, - "inner", - Type::Struct(StructType::new(vec![ - NestedField::optional(4, "x", Type::Primitive(PrimitiveType::Long)) - .into(), - ])), - ) - .into(), - ])), - ), - Arc::new(file), - ) - .unwrap(); - let inner_fields = Fields::from(vec![field_with_id("x", DataType::Int64, true, 4)]); - let expected = StructArray::new( - Fields::from(vec![field_with_id( - "inner", - DataType::Struct(inner_fields.clone()), - true, - 3, - )]), - vec![Arc::new(StructArray::new( - inner_fields, - vec![Arc::new(Int64Array::from(vec![10i64, 20, 30])) as ArrayRef], - None, - )) as ArrayRef], + fn promote_idless_map_struct_value_reads_null_fields() { + let value = StructArray::new( + Fields::from(vec![Field::new("v", DataType::Int32, true)]), + vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef], None, ); - assert_eq!(out.as_struct(), &expected); - } - - #[test] - fn promote_idless_map_keeps_data() { let entries = StructArray::new( Fields::from(vec![ Field::new("key", DataType::Utf8, false), - Field::new("value", DataType::Int32, true), + Field::new("value", value.data_type().clone(), true), ]), vec![ Arc::new(StringArray::from(vec!["k1", "k2"])) as ArrayRef, - Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef, + Arc::new(value) as ArrayRef, ], None, ); @@ -1931,20 +1904,27 @@ mod test { 3, Type::Primitive(PrimitiveType::String), 4, - Type::Primitive(PrimitiveType::Int), + Type::Struct(StructType::new(vec![ + NestedField::optional(5, "v", Type::Primitive(PrimitiveType::Int)).into(), + ])), )), ), Arc::new(map), ) .unwrap(); + let value_fields = Fields::from(vec![field_with_id("v", DataType::Int32, true, 5)]); let expected_entries = StructArray::new( Fields::from(vec![ field_with_id("key", DataType::Utf8, false, 3), - field_with_id("value", DataType::Int32, true, 4), + field_with_id("value", DataType::Struct(value_fields.clone()), true, 4), ]), vec![ Arc::new(StringArray::from(vec!["k1", "k2"])) as ArrayRef, - Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef, + Arc::new(StructArray::new( + value_fields, + vec![new_null_array(&DataType::Int32, 2)], + None, + )) as ArrayRef, ], None, ); @@ -1959,42 +1939,10 @@ mod test { None, false, ); - // Map array equality skips the entries field name. assert_eq!(out.data_type(), expected.data_type()); assert_eq!(out.as_map(), &expected); } - #[test] - fn promote_idless_same_type_reorder_stays_positional() { - // Without ids a same-type reorder is undetectable; #1845 replaces this fallback. - let source = Arc::new(StructArray::new( - Fields::from(vec![ - Field::new("a", DataType::Int32, true), - Field::new("b", DataType::Int32, true), - ]), - vec![ - Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef, - Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef, - ], - None, - )) as ArrayRef; - let fields = Fields::from(vec![ - field_with_id("b", DataType::Int32, true, 6), - field_with_id("a", DataType::Int32, true, 5), - ]); - - let out = promote(&source, &DataType::Struct(fields.clone()), &empty_schema()).unwrap(); - let expected = StructArray::new( - fields, - vec![ - Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef, - Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef, - ], - None, - ); - assert_eq!(out.as_struct(), &expected); - } - #[test] fn promote_large_list_file_column_to_list() { let element = Arc::new(field_with_id("element", DataType::Int32, true, 3));
