ewoodbury commented on code in PR #3255:
URL: https://github.com/apache/iceberg-rust/pull/3255#discussion_r4142624396
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,296 @@ 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(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ (
+ DataType::List(source_field) |
DataType::LargeList(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::List(source_field) |
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(invalid_data!("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);
+ }
+ }
+ // Name mapping only assigns top-level ids (#1845). Fully id-less
children
+ // match by position when the counts line up, and `build` checks each
pair.
+ // A same-type reorder cannot be detected without ids. Any missing id
+ // errors instead of nulling that child.
+ if !source_fields.is_empty() && 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,
+ )?,
+ })
+ })
+ .collect();
+ }
+ return Err(invalid_data!(
+ "cannot reconcile struct fields by id: source fields do not
all have field ids"
+ ));
+ }
+
+ 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 => {
+ // Missing nested fields are null-filled. Applying an
+ // initial-default here is not implemented yet (#3261).
+ 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, see https://github.com/apache/iceberg-rust/issues/3261";,
+ target_field.name()
+ ),
+ ));
+ }
+ if iceberg_field.map_or(!target_field.is_nullable(),
|f| f.required) {
+ return Err(invalid_data!(
+ "required nested field {} is absent from the
data file and has no initial-default",
+ target_field.name()
+ ));
+ }
+ Ok(ChildPlan::Null {
+ target_type: target_field.data_type().clone(),
+ })
+ }
Review Comment:
Ah, this is a great catch, pretty subtle. I updated this so the
missing-child arm checks constant_fields before initial-default or null.
`PromotePlan::build` takes the map, and `apply_struct` builds the column with
create_column.
I covered this case you ran with new test
`identity_partition_fills_missing_nested_field`, so the `s: struct<region, x>`
partitioned by `identity(s.region) = "us"`, file only has x, and region comes
back `["us", "us"]`. A child that is in the file still comes from the file
--
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]
