jordepic commented on code in PR #2647:
URL: https://github.com/apache/iceberg-rust/pull/2647#discussion_r3650480324
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -619,7 +621,113 @@ impl RecordBatchTransformer {
})
.collect()
}
+}
+
+/// Look up an Iceberg field id from an Arrow field's `PARQUET:field_id`
metadata.
+fn arrow_field_id(field: &Field) -> Option<i32> {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .and_then(|s| s.parse::<i32>().ok())
+}
+
+/// Mirrors iceberg-java logic, resolve by field ID instead of position.
+fn cast_schema_to_target(array: &ArrayRef, target_type: &DataType) ->
Result<ArrayRef> {
+ match target_type {
+ DataType::Struct(target_children) => {
+ let source = array.as_struct_opt().ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "expected a struct array to promote to
{target_type:?}, got {:?}",
+ array.data_type()
+ ),
+ )
+ })?;
+ let mut source_by_id: HashMap<i32, usize> = HashMap::new();
+ for (idx, field) in source.fields().iter().enumerate() {
+ if let Some(id) = arrow_field_id(field) {
+ source_by_id.insert(id, idx);
+ }
+ }
+
+ let len = source.len();
+ let mut new_columns: Vec<ArrayRef> =
Vec::with_capacity(target_children.len());
+ for target_child in target_children.iter() {
+ let matched = arrow_field_id(target_child)
+ .and_then(|id| source_by_id.get(&id))
+ .copied();
+ match matched {
+ Some(src_idx) => new_columns.push(cast_schema_to_target(
+ source.column(src_idx),
+ target_child.data_type(),
+ )?),
+ None =>
new_columns.push(new_null_array(target_child.data_type(), len)),
Review Comment:
Agreed it shouldn't diverge silently. Since the plan is now built once per
file with the snapshot schema in hand, `build_struct_children` enforces the
policy instead of nulling unconditionally: an absent required child without a
default errors (matching the top-level path), an absent child with an
`initial-default` returns an explicit `FeatureUnsupported` error rather than a
silent null (fully wiring nested defaults through is the larger change you
suspected, and the error keeps the gap visible), and only
optional-without-default null-fills. There's a comment at that spot documenting
the limitation. Tests:
`promote_required_nested_field_absent_without_default_errors`,
`promote_nested_field_with_initial_default_errors`.
Neither error is a regression — main's positional cast also fails on those
files, just less legibly.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -619,7 +621,113 @@ impl RecordBatchTransformer {
})
.collect()
}
+}
+
+/// Look up an Iceberg field id from an Arrow field's `PARQUET:field_id`
metadata.
+fn arrow_field_id(field: &Field) -> Option<i32> {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .and_then(|s| s.parse::<i32>().ok())
+}
+
+/// Mirrors iceberg-java logic, resolve by field ID instead of position.
+fn cast_schema_to_target(array: &ArrayRef, target_type: &DataType) ->
Result<ArrayRef> {
+ match target_type {
+ DataType::Struct(target_children) => {
+ let source = array.as_struct_opt().ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "expected a struct array to promote to
{target_type:?}, got {:?}",
+ array.data_type()
+ ),
+ )
+ })?;
+ let mut source_by_id: HashMap<i32, usize> = HashMap::new();
+ for (idx, field) in source.fields().iter().enumerate() {
+ if let Some(id) = arrow_field_id(field) {
+ source_by_id.insert(id, idx);
+ }
+ }
+
+ let len = source.len();
+ let mut new_columns: Vec<ArrayRef> =
Vec::with_capacity(target_children.len());
+ for target_child in target_children.iter() {
+ let matched = arrow_field_id(target_child)
+ .and_then(|id| source_by_id.get(&id))
+ .copied();
+ match matched {
+ Some(src_idx) => new_columns.push(cast_schema_to_target(
+ source.column(src_idx),
+ target_child.data_type(),
+ )?),
+ None =>
new_columns.push(new_null_array(target_child.data_type(), len)),
+ }
+ }
+ Ok(Arc::new(StructArray::new(
Review Comment:
Switched — `StructArray`, both list variants (via `GenericListArray<O>`),
and `MapArray` all use `try_new` now, propagating as `iceberg::Error` through
the existing `From<ArrowError>` impl.
--
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]
