jordepic commented on code in PR #2647:
URL: https://github.com/apache/iceberg-rust/pull/2647#discussion_r3650481929
##########
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),
+ target_entries.data_type(),
+ )?;
+ let entries_struct = entries.as_struct().clone();
+ Ok(Arc::new(MapArray::new(
+ target_entries.clone(),
+ source.offsets().clone(),
+ entries_struct,
+ source.nulls().cloned(),
+ *sorted,
+ )))
+ }
+ _ => Ok(cast(array.as_ref(), target_type)?),
+ }
+}
+
+fn list_err(array: &ArrayRef, target_type: &DataType) -> Error {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "expected a list/map array to promote to {target_type:?}, got
{:?}",
+ array.data_type()
+ ),
+ )
+}
+
+impl RecordBatchTransformer {
Review Comment:
Fixed — the free fns are gone; `PromotePlan`/`ChildPlan` and their impl sit
with the other enums right after `ColumnSource`, and the
`RecordBatchTransformer` impl is a single block again (`create_column` merged
back in).
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -663,18 +771,209 @@ mod test {
use std::collections::HashMap;
use std::sync::Arc;
+ use arrow_array::cast::AsArray;
+ use arrow_array::types::Int32Type;
use arrow_array::{
- Array, Date32Array, Float32Array, Float64Array, Int32Array,
Int64Array, RecordBatch,
- StringArray,
+ Array, ArrayRef, Date32Array, Float32Array, Float64Array, Int32Array,
Int64Array,
+ LargeListArray, ListArray, MapArray, RecordBatch, StringArray,
StructArray,
};
- use arrow_schema::{DataType, Field, Schema as ArrowSchema};
+ use arrow_buffer::OffsetBuffer;
+ use arrow_schema::{DataType, Field, Fields, Schema as ArrowSchema};
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use crate::arrow::record_batch_transformer::{
- RecordBatchTransformer, RecordBatchTransformerBuilder,
+ RecordBatchTransformer, RecordBatchTransformerBuilder,
cast_schema_to_target,
};
use crate::spec::{Literal, NestedField, PrimitiveType, Schema, Struct,
Type};
+ fn with_id(field: Field, id: i32) -> Field {
Review Comment:
Done — dropped `with_id`/`field_with_id`; the tests use the existing
`simple_field`.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -610,7 +612,7 @@ impl RecordBatchTransformer {
ColumnSource::Promote {
target_type,
source_index,
- } => cast(&*columns[*source_index], target_type)?,
+ } => cast_schema_to_target(&columns[*source_index],
target_type)?,
Review Comment:
Added the high-value ones:
- rename-by-id: `promote_struct_renames_field_by_id`
- nested primitive promotion (int → long in a struct child):
`promote_struct_promotes_child_primitive`
- null parent row (`source.nulls()` survives):
`promote_struct_preserves_null_parent_rows`
- end-to-end via `process_record_batch`:
`promote_evolved_nested_struct_via_process_record_batch`, which drives rename +
child promotion + added-field null fill + null parent rows through one nested
struct column
The name-mapping guard and the required/`initial-default` corners are pinned
by the tests noted on the other threads. Reorder falls out of by-id matching
(the added-middle-field test already exercises a shifted child), and deletes
are handled upstream by the projection mask. I'm not aware of coverage for
these paths on the Comet side, so having them here seems right.
--
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]
