jordepic commented on code in PR #2647:
URL: https://github.com/apache/iceberg-rust/pull/2647#discussion_r3650481192
##########
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(
+ target_children.clone(),
+ new_columns,
+ source.nulls().cloned(),
+ )))
+ }
+ DataType::List(target_element) => {
+ let source = array
+ .as_list_opt::<i32>()
Review Comment:
Done — the two arms collapse into `PromotePlan::apply_list<O:
OffsetSizeTrait>` over `GenericListArray<O>`, so the recursion/offset/nulls
logic exists once.
##########
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(
+ target_children.clone(),
+ new_columns,
+ source.nulls().cloned(),
+ )))
+ }
+ DataType::List(target_element) => {
+ let source = array
+ .as_list_opt::<i32>()
+ .ok_or_else(|| list_err(array, target_type))?;
+ let values = cast_schema_to_target(source.values(),
target_element.data_type())?;
+ Ok(Arc::new(ListArray::new(
+ target_element.clone(),
+ source.offsets().clone(),
+ values,
+ source.nulls().cloned(),
+ )))
+ }
+ DataType::LargeList(target_element) => {
+ let source = array
+ .as_list_opt::<i64>()
+ .ok_or_else(|| list_err(array, target_type))?;
+ let values = cast_schema_to_target(source.values(),
target_element.data_type())?;
+ Ok(Arc::new(LargeListArray::new(
+ target_element.clone(),
+ source.offsets().clone(),
+ values,
+ source.nulls().cloned(),
+ )))
+ }
+ DataType::Map(target_entries, sorted) => {
+ let source = array
+ .as_map_opt()
+ .ok_or_else(|| list_err(array, target_type))?;
+ let entries = cast_schema_to_target(
+ &(Arc::new(source.entries().clone()) as ArrayRef),
Review Comment:
Done — the map plan now stores the entry `Fields` plus per-child plans
directly and calls `apply_struct(source.entries(), ...)`, which returns a
`StructArray`, so the `ArrayRef` round trip and both clones are gone. This fell
out of the plan restructuring from your other comment.
##########
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> {
Review Comment:
Renamed — the function became `PromotePlan::build`/`PromotePlan::apply`, and
the doc comment on `PromotePlan` uses your suggested sentence. Updated the PR
description to match the new shape.
--
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]
