laskoviymishka commented on code in PR #3255:
URL: https://github.com/apache/iceberg-rust/pull/3255#discussion_r4232840902
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> Result<Vec<ChildPlan>> {
+ let mut source_by_id = HashMap::with_capacity(source_fields.len());
Review Comment:
`HashMap::insert` silently overwrites on a duplicate id, so a struct with
two children sharing an id collapses the count and then trips the `do not all
have field ids` error below — which points at the wrong cause. I'd catch the
collision where it happens:
```rust
if source_by_id.insert(id, idx).is_some() {
return Err(invalid_data!("duplicate field id {id} in struct"));
}
```
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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:?}"
Review Comment:
Non-struct map entries are a bad input, not an internal invariant, so I'd
use `invalid_data!` here to match the sibling `cannot promote` / `required
nested field` arms rather than `Unexpected`. Minor, but it keeps the kind
meaningful for callers that match on it (`promote_err` below has the same
smell).
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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:?}"))
Review Comment:
This guard only rejects an illegal source when the target is nested, so a
nested source against a primitive target (struct→Int32) falls through to `Cast`
and only errors when the first batch arrives — and never at all for a zero-row
file. Since the win of resolving the plan once is surfacing structural
mismatches at plan time, I'd add the symmetric arm: error when the source is
`Struct`/`List`/`LargeList`/`Map` and the target isn't.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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_by_id.len() != source_fields.len() {
+ if source_by_id.is_empty() && source_fields.len() ==
target_fields.len() {
Review Comment:
This positional branch is the one spot that can still silently misassign: a
same-type reorder (or a drop-plus-add that keeps the count and types) reads the
old child's data under the new child's id, with no error. It's documented and
no worse than the old positional cast, so I'm not blocking on it — but since
the whole point of this change is to kill positional matching, I'd at least
require the source and target child names to match here (as a multiset) and
error otherwise, or pin the current behavior with a test and cross-link #1845
so nobody later mistakes this for a supported path.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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_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,
+ constant_fields,
+ )?,
+ })
+ })
+ .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,
+ constant_fields,
+ )?,
+ source_index,
+ }),
+ None => {
+ // Identity partition constants apply inside structs
too. A missing
+ // child uses that value before initial-default
(#3261) or null.
+ if let Some(ColumnConstant::Scalar(datum)) =
+ field_id.and_then(|id| constant_fields.get(&id))
+ {
+ return Ok(ChildPlan::Add {
+ target_type: target_field.data_type().clone(),
+ value: Some(datum.literal().clone()),
+ });
+ }
+ // 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()) {
Review Comment:
Erroring loudly here beats silently null-filling, so no objection to
shipping it as-is. That said, this looks cheap to just support:
`ChildPlan::Add` already carries `Option<PrimitiveLiteral>`, and the value is
`iceberg_field.initial_default` fed through `create_column` — the exact path
the constant branch just above already uses. If it folds in cleanly I'd rather
close #3261 than leave a nested column with a default unreadable for every
pre-evolution file; otherwise keep the issue open with a TODO right here.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -1147,22 +1422,546 @@ 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_buffer::{NullBuffer, OffsetBuffer};
use arrow_cast::cast;
- use arrow_schema::{DataType, Field, Schema as ArrowSchema};
+ use arrow_schema::{DataType, Field, Fields, Schema as ArrowSchema};
use super::{field_with_id, schema_to_arrow_schema};
- use crate::arrow::build_partition_constant;
+ use crate::ErrorKind;
use crate::arrow::record_batch_transformer::{
- RecordBatchTransformer, RecordBatchTransformerBuilder,
+ PromotePlan, RecordBatchTransformer, RecordBatchTransformerBuilder,
};
+ use crate::arrow::{DEFAULT_MAP_FIELD_NAME, build_partition_constant};
use crate::spec::{
- ListType, Literal, MapType, NestedField, PrimitiveType, Schema,
Struct, Type,
+ ListType, Literal, MapType, NestedField, PrimitiveType, Schema,
Struct, StructType, Type,
};
+ fn promote(source: &ArrayRef, target: &DataType, schema: &Schema) ->
crate::Result<ArrayRef> {
+ PromotePlan::build(source.data_type(), target, schema,
&HashMap::new())?.apply(source)
+ }
+
+ fn empty_schema() -> Schema {
+ Schema::builder().build().unwrap()
+ }
+
+ fn unevolved_struct_type() -> DataType {
+ DataType::Struct(Fields::from(vec![field_with_id(
+ "x",
+ DataType::Int32,
+ true,
+ 5,
+ )]))
+ }
+
+ fn evolved_struct_type() -> DataType {
+ DataType::Struct(Fields::from(vec![
+ field_with_id("x", DataType::Int32, true, 5),
+ field_with_id("y", DataType::Int32, true, 6),
+ ]))
+ }
+
+ fn unevolved_struct_data(x_values: Vec<i32>) -> Arc<StructArray> {
+ Arc::new(StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(x_values)) as ArrayRef],
+ None,
+ ))
+ }
+
+ fn transform_top_level(column: NestedField, file_column: ArrayRef) ->
crate::Result<ArrayRef> {
Review Comment:
+1 to @comphead's reader-level test suggestion. Every fixture here
hand-stamps `field_with_id`, so nothing actually pins that parquet-rs writes
`PARQUET:field_id` onto struct children, list `element`, and map key/value. If
any of those stamps is absent in practice, `build_struct_children` quietly
falls back to the positional / `DataInvalid` path this change was written to
eliminate, and all of these tests still pass green. One end-to-end test in
`reader/projection.rs` that writes an old nested schema and reads it back
through an evolved one would guard the real contract — and as @comphead noted,
it can stand in for several of these unit tests. While you're in there, a
sliced list/map input and a null list/map slot would pin the offset reuse too.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +195,284 @@ 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>,
+ },
+ Map {
+ field: FieldRef,
+ entry_fields: Fields,
+ entries: Vec<ChildPlan>,
+ sorted: bool,
+ },
+}
+
+#[derive(Debug)]
+pub(crate) enum ChildPlan {
+ FromSource {
+ source_index: usize,
+ plan: PromotePlan,
+ },
+ Add {
+ target_type: DataType,
+ value: Option<PrimitiveLiteral>,
+ },
+}
+
+impl PromotePlan {
+ fn build(
+ source: &DataType,
+ target: &DataType,
+ snapshot_schema: &IcebergSchema,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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,
+ constant_fields,
+ )?,
+ fields: target_fields.clone(),
+ })
+ }
+ // A file's Arrow schema hint can give either offset width;
`apply` converts it.
+ // Table schemas are always `List`
(`ToArrowSchemaConverter::list`).
+ (
+ 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,
+ constant_fields,
+ )?),
+ 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,
+ constant_fields,
+ )?,
+ 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,
+ constant_fields: &HashMap<i32, ColumnConstant>,
+ ) -> 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_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,
+ constant_fields,
+ )?,
+ })
+ })
+ .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,
+ constant_fields,
+ )?,
+ source_index,
+ }),
+ None => {
+ // Identity partition constants apply inside structs
too. A missing
+ // child uses that value before initial-default
(#3261) or null.
+ if let Some(ColumnConstant::Scalar(datum)) =
+ field_id.and_then(|id| constant_fields.get(&id))
+ {
+ return Ok(ChildPlan::Add {
+ target_type: target_field.data_type().clone(),
+ value: Some(datum.literal().clone()),
+ });
+ }
+ // 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) {
Review Comment:
This absent-field policy (initial-default → required → null) restates the
top-level `Add` logic in `generate_transform_operations`, down to
`map_or(!is_nullable)` here versus `required` there — close enough to drift
apart later. Not blocking, but a shared `resolve_absent_field(field,
target_type, constants)` would keep the two in lockstep.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -1147,22 +1422,546 @@ 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_buffer::{NullBuffer, OffsetBuffer};
use arrow_cast::cast;
- use arrow_schema::{DataType, Field, Schema as ArrowSchema};
+ use arrow_schema::{DataType, Field, Fields, Schema as ArrowSchema};
use super::{field_with_id, schema_to_arrow_schema};
- use crate::arrow::build_partition_constant;
+ use crate::ErrorKind;
use crate::arrow::record_batch_transformer::{
- RecordBatchTransformer, RecordBatchTransformerBuilder,
+ PromotePlan, RecordBatchTransformer, RecordBatchTransformerBuilder,
};
+ use crate::arrow::{DEFAULT_MAP_FIELD_NAME, build_partition_constant};
use crate::spec::{
- ListType, Literal, MapType, NestedField, PrimitiveType, Schema,
Struct, Type,
+ ListType, Literal, MapType, NestedField, PrimitiveType, Schema,
Struct, StructType, Type,
};
+ fn promote(source: &ArrayRef, target: &DataType, schema: &Schema) ->
crate::Result<ArrayRef> {
+ PromotePlan::build(source.data_type(), target, schema,
&HashMap::new())?.apply(source)
+ }
+
+ fn empty_schema() -> Schema {
+ Schema::builder().build().unwrap()
+ }
+
+ fn unevolved_struct_type() -> DataType {
+ DataType::Struct(Fields::from(vec![field_with_id(
+ "x",
+ DataType::Int32,
+ true,
+ 5,
+ )]))
+ }
+
+ fn evolved_struct_type() -> DataType {
+ DataType::Struct(Fields::from(vec![
+ field_with_id("x", DataType::Int32, true, 5),
+ field_with_id("y", DataType::Int32, true, 6),
+ ]))
+ }
+
+ fn unevolved_struct_data(x_values: Vec<i32>) -> Arc<StructArray> {
+ Arc::new(StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(x_values)) as ArrayRef],
+ None,
+ ))
+ }
+
+ fn transform_top_level(column: NestedField, file_column: ArrayRef) ->
crate::Result<ArrayRef> {
+ let name = column.name.clone();
+ let snapshot_schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ column.into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let mut transformer =
RecordBatchTransformerBuilder::new(snapshot_schema, &[1, 2]).build();
+ let file_schema = Arc::new(ArrowSchema::new(vec![
+ field_with_id("id", DataType::Int32, false, 1),
+ field_with_id(name, file_column.data_type().clone(), true, 2),
+ ]));
+ let batch = RecordBatch::try_new(file_schema, vec![
+ Arc::new(Int32Array::from(vec![1; file_column.len()])) as ArrayRef,
+ file_column,
+ ])
+ .unwrap();
+ Ok(transformer.process_record_batch(batch)?.column(1).clone())
+ }
+
+ #[test]
+ fn promote_struct_fills_added_fields_by_id() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![
+ field_with_id("a", DataType::Int32, true, 1),
+ field_with_id("c", DataType::Utf8, true, 3),
+ ]),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["x", "y"])) as ArrayRef,
+ ],
+ None,
+ )) as ArrayRef;
+ let fields = Fields::from(vec![
+ field_with_id("a", DataType::Int32, true, 1),
+ field_with_id("b", DataType::Int32, true, 2),
+ field_with_id("c", DataType::Utf8, true, 3),
+ field_with_id("d", DataType::Int32, true, 4),
+ ]);
+ let target = DataType::Struct(fields.clone());
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let expected = StructArray::new(
+ fields,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(Int32Array::from(vec![None, None])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["x", "y"])) as ArrayRef,
+ Arc::new(Int32Array::from(vec![None, None])) as ArrayRef,
+ ],
+ None,
+ );
+ assert_eq!(out.as_struct(), &expected);
+ }
+
+ #[test]
+ fn promote_list_element_struct_fills_added_field_by_id() {
+ let source = Arc::new(ListArray::new(
+ Arc::new(field_with_id("element", unevolved_struct_type(), true,
4)),
+ OffsetBuffer::new(vec![0, 1, 2].into()),
+ unevolved_struct_data(vec![10, 20]),
+ None,
+ )) as ArrayRef;
+ let target = DataType::List(Arc::new(field_with_id(
+ "element",
+ evolved_struct_type(),
+ true,
+ 4,
+ )));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let values = StructArray::new(
+ Fields::from(vec![
+ field_with_id("x", DataType::Int32, true, 5),
+ field_with_id("y", DataType::Int32, true, 6),
+ ]),
+ vec![
+ Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef,
+ Arc::new(Int32Array::from(vec![None, None])) as ArrayRef,
+ ],
+ None,
+ );
+ let expected = ListArray::new(
+ Arc::new(field_with_id("element", evolved_struct_type(), true, 4)),
+ OffsetBuffer::new(vec![0, 1, 2].into()),
+ Arc::new(values),
+ None,
+ );
+ assert_eq!(out.as_list::<i32>(), &expected);
Review Comment:
If I'm reading arrow's List/Map equality right,
`assert_eq!(out.as_list::<i32>(), &expected)` compares through `ArrayData` and
doesn't check the `element`/`entries` field's own name or `PARQUET:field_id` —
which is exactly what `PromotePlan` installs. So a regression that kept the
source element's stale id/name would still pass this test. I'd also assert on
the type, where `Field` equality does include metadata:
`assert_eq!(out.data_type(), expected.data_type())`, and the same at the
equivalent spots in the map tests.
--
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]
