mbutrovich commented on code in PR #2647:
URL: https://github.com/apache/iceberg-rust/pull/2647#discussion_r4065630679
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -831,25 +848,26 @@ impl RecordBatchTransformer {
//
// At this point, all field IDs in the source schema are
trustworthy.
// No conflict detection needed - schema resolution happened
in reader.rs.
- let field_by_id =
field_id_to_source_schema_map.get(field_id).map(
- |(source_field, source_index)| {
- if
source_field.data_type().equals_datatype(target_type) {
- ColumnSource::PassThrough {
- source_index: *source_index,
- }
- } else {
- ColumnSource::Promote {
- target_type: target_type.clone(),
- source_index: *source_index,
- }
- }
- },
- );
-
+ //
// Apply spec's fallback steps for "not present" fields.
// Rule #1 (constants) is handled at the beginning of this
function
- let column_source = if let Some(source) = field_by_id {
- source
+ let column_source = if let Some((source_field, source_index)) =
+ field_id_to_source_schema_map.get(field_id)
+ {
+ if source_field.data_type().equals_datatype(target_type) {
Review Comment:
The spec allows reordering and renaming fields in nested structs, not just
top-level ones ([Schema
Evolution](https://iceberg.apache.org/spec/#schema-evolution): "Any struct,
including a top-level schema, can evolve through deleting fields, adding new
fields, renaming existing fields, reordering existing fields"). [Column
Projection](https://iceberg.apache.org/spec/#column-projection) then requires
projection by id because "the table schema's column names and order may change
after a data file is written".
`DataType::equals_datatype` compares struct children by position, type, and
nullability only. It ignores child names and metadata, so it ignores
`PARQUET:field_id` too ([arrow-schema 58.4.0
`datatype.rs`](https://github.com/apache/arrow-rs/blob/58.4.0/arrow-schema/src/datatype.rs#L683-L689)).
When a nested struct's children are reordered or renamed but their types still
line up, this check returns true and the column takes
`ColumnSource::PassThrough` without ever reaching `PromotePlan`. Could you add
a test for this case? Here's one I ran on the head commit:
```rust
#[test]
fn reorder_nested_struct_fields_with_identical_types() {
let snapshot_schema = Arc::new(
Schema::builder()
.with_schema_id(1)
.with_fields(vec![
NestedField::optional(
2,
"s",
Type::Struct(crate::spec::StructType::new(vec![
NestedField::optional(6, "b",
Type::Primitive(PrimitiveType::Int)).into(),
NestedField::optional(5, "a",
Type::Primitive(PrimitiveType::Int)).into(),
])),
)
.into(),
])
.build()
.unwrap(),
);
let mut transformer =
RecordBatchTransformerBuilder::new(snapshot_schema, &[2]).build();
let file_fields = Fields::from(vec![
field_with_id("a", DataType::Int32, true, 5),
field_with_id("b", DataType::Int32, true, 6),
]);
let file_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
"s",
DataType::Struct(file_fields.clone()),
true,
2,
)]));
let file_batch = RecordBatch::try_new(file_schema,
vec![Arc::new(StructArray::new(
file_fields,
vec![
Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
Arc::new(Int32Array::from(vec![100, 200])) as ArrayRef,
],
None,
))])
.unwrap();
let result = transformer.process_record_batch(file_batch).unwrap();
let s = result.column(0).as_struct();
assert_eq!(s.fields()[0].name(), "b");
assert_eq!(s.column(0).as_primitive::<Int32Type>().values(), &[100,
200]);
}
```
On the head commit, the batch schema says position 0 is `b` (id 6), but the
array there holds `a`'s values `[1, 2]`. The struct array also keeps the file's
field order, so its data type no longer matches the schema. A rename with no
type change fails the same way: the array keeps the old child name while the
schema has the new one. Your
`promote_evolved_nested_struct_via_process_record_batch` test only gets the
rename right because the int-to-long promotion forces the `Promote` path.
Can we route these columns through `PromotePlan::build` and let the plan
decide when to pass through? One constraint: files with no field ids rely on
today's positional pass-through
([`test_read_parquet_without_field_ids_with_struct`](https://github.com/apache/iceberg-rust/blob/832a4eba7a732baebbbd7f76bb6fd6225f1db1e9/crates/iceberg/src/arrow/reader/projection.rs#L1378)).
The reader assigns fallback ids to top-level fields only, so routing
everything through the plan trips the new "no source field carries a field id"
guard. This version keeps that path working:
```rust
// generate_transform_operations
match PromotePlan::build(source_field.data_type(), target_type,
snapshot_schema)? {
PromotePlan::PassThrough => ColumnSource::PassThrough {
source_index: *source_index,
},
plan => ColumnSource::Promote {
plan,
source_index: *source_index,
},
}
// PromotePlan::build, before the existing struct arm
(DataType::Struct(source_fields), DataType::Struct(_))
if source.equals_datatype(target)
&& source_fields
.iter()
.all(|f| !f.metadata().contains_key(PARQUET_FIELD_ID_META_KEY))
=>
{
Ok(PromotePlan::PassThrough)
}
```
With both changes, the test above and a rename-only variant pass, and so
does all of `cargo test -p iceberg --lib` (1482 tests). The per-batch cost for
a struct whose ids match but whose `Field`s aren't strictly equal (a `doc`
metadata entry, for example) is one `StructArray` rebuild from `Arc` clones.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -192,6 +154,272 @@ pub(crate) enum ColumnSource {
// post-processing step by using the projection mask.
}
+/// Reconciles a source array to the target type by matching nested fields on
+/// field id rather than position, mirroring iceberg-java's nested readers.
+///
+/// The plan is resolved once per file, when the `BatchTransform` is built, so
+/// applying it per batch is just index lookups and array assembly.
+#[derive(Debug)]
+pub(crate) enum PromotePlan {
+ PassThrough,
+ Cast(DataType),
+ Struct {
+ fields: Fields,
+ children: Vec<ChildPlan>,
+ },
+ List {
+ field: FieldRef,
+ element: Box<PromotePlan>,
+ },
+ LargeList {
+ field: FieldRef,
+ element: Box<PromotePlan>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Null {
+ target_type: DataType,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ ) -> Result<Self> {
+ if source == target {
+ return Ok(PromotePlan::PassThrough);
+ }
+ match (source, target) {
+ (DataType::Struct(source_fields), DataType::Struct(target_fields))
=> {
+ Ok(PromotePlan::Struct {
+ children: Self::build_struct_children(
+ source_fields,
+ target_fields,
+ snapshot_schema,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ (DataType::List(source_field), DataType::List(target_field)) =>
Ok(PromotePlan::List {
+ element: Box::new(Self::build(
+ source_field.data_type(),
+ target_field.data_type(),
+ snapshot_schema,
+ )?),
+ field: target_field.clone(),
+ }),
+ (DataType::LargeList(source_field),
DataType::LargeList(target_field)) => {
+ Ok(PromotePlan::LargeList {
+ element: Box::new(Self::build(
+ source_field.data_type(),
+ target_field.data_type(),
+ snapshot_schema,
+ )?),
+ field: target_field.clone(),
+ })
+ }
+ (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)) => {
+ Ok(PromotePlan::Map {
+ entries: Self::build_struct_children(
+ source_fields,
+ target_fields,
+ snapshot_schema,
+ )?,
+ entry_fields: target_fields.clone(),
+ field: target_entries.clone(),
+ sorted: *sorted,
+ })
+ }
+ _ => Err(Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "expected struct-typed map entries, got
{source_entries:?} and {target_entries:?}"
+ ),
+ )),
+ }
+ }
+ (_, DataType::Struct(_) | DataType::List(_) |
DataType::LargeList(_))
+ | (_, DataType::Map(_, _)) => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!("cannot promote {source:?} to {target:?}"),
+ )),
+ _ => Ok(PromotePlan::Cast(target.clone())),
+ }
+ }
+
+ fn build_struct_children(
+ source_fields: &Fields,
+ target_fields: &Fields,
+ snapshot_schema: &IcebergSchema,
+ ) -> Result<Vec<ChildPlan>> {
+ let mut source_by_id = HashMap::with_capacity(source_fields.len());
+ for (idx, field) in source_fields.iter().enumerate() {
+ if let Some(id) = try_get_field_id_from_metadata(field)? {
+ source_by_id.insert(id, idx);
+ }
+ }
+ // Reachable for id-less files with nested types because the reader
only
+ // applies name mapping to top-level fields; error rather than silently
+ // nulling every child.
+ if !source_fields.is_empty() && source_by_id.is_empty() {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ "cannot reconcile struct fields by id: no source field carries
a field id",
+ ));
+ }
+
+ target_fields
+ .iter()
+ .map(|target_field| {
+ let field_id = try_get_field_id_from_metadata(target_field)?;
+ match field_id.and_then(|id| source_by_id.get(&id).copied()) {
+ Some(source_index) => Ok(ChildPlan::FromSource {
+ plan: Self::build(
+ source_fields[source_index].data_type(),
+ target_field.data_type(),
+ snapshot_schema,
+ )?,
+ source_index,
+ }),
+ None => {
+ // Nested fields absent from the file only support
rule #4
+ // (null) of the spec's Column Projection rules; rule
#3
+ // (initial-default) is only wired up for top-level
columns
+ // via ColumnSource::Add.
+ let iceberg_field =
+ field_id.and_then(|id|
snapshot_schema.field_by_id(id));
+ if iceberg_field.is_some_and(|f|
f.initial_default.is_some()) {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ format!(
+ "initial-default of nested field {} is not
supported",
+ target_field.name()
+ ),
+ ));
Review Comment:
Returning `FeatureUnsupported` here is better than a silent null, and it
keeps the gap visible. The spec does define this case, though. [Default
values](https://iceberg.apache.org/spec/#default-values) covers sub-field
defaults in nested structs, and its table gives `{"x": 3}` with `point.y`
defaulting to `0` as `{"x": 3, "y": 0}`. On a v3 table with a defaulted nested
field, reads of older files will fail until this is implemented. There's no
issue for it yet. Could you open one and reference it in this comment and the
error message, so it doesn't get lost after merge?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
