mbutrovich commented on code in PR #3255:
URL: https://github.com/apache/iceberg-rust/pull/3255#discussion_r4067028857
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -192,6 +194,272 @@ 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(),
+ })
+ }
+ (DataType::List(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::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(Error::new(
+ ErrorKind::DataInvalid,
+ format!("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);
+ }
+ }
+ // Reachable for id-less files with nested types because the reader
only
+ // applies name mapping to top-level fields; error rather than silently
+ // nulling every child.
+ if !source_fields.is_empty() && source_by_id.is_empty() {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ "cannot reconcile struct fields by id: no source field carries
a field id",
+ ));
+ }
Review Comment:
This guard is right for structs whose types differ, but files without field
ids that previously passed through positionally still need to work once more
columns reach `PromotePlan::build` (see the comment at line 1108). A map from
an id-less file is the case that breaks: the `Map` arm calls
`build_struct_children` directly, and `key`/`value` carry no ids because the
reader assigns fallback ids to [top-level fields
only](https://github.com/apache/iceberg-rust/blob/0054dd3cc78f2d2c9971f800b5bfa845fdf3c53e/crates/iceberg/src/arrow/reader/projection.rs#L469-L471).
With the line 1108 change alone, reading an id-less `map<string, int>` column
fails with `DataInvalid => cannot reconcile struct fields by id: no source
field carries a field id`. On this head it passes.
Could we keep positional matching when there are no ids and the child types
still line up? Doing it here covers the struct and map arms in one place, and
recursing through `build` handles a struct nested in an id-less struct:
```suggestion
// Reachable for id-less files with nested types because the reader
only
// applies name mapping to top-level fields. Keep positional
matching when
// the child types still line up, and error otherwise rather than
silently
// nulling every child.
if !source_fields.is_empty() && source_by_id.is_empty() {
if source_fields.len() == target_fields.len()
&& source_fields
.iter()
.zip(target_fields.iter())
.all(|(s, t)|
s.data_type().equals_datatype(t.data_type()))
{
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(Error::new(
ErrorKind::DataInvalid,
"cannot reconcile struct fields by id: no source field
carries a field id",
));
}
```
With this and the line 1108 change, tests for an id-less `map<string, int>`
and an id-less `struct<inner: struct<x: int>>` both pass through
`process_record_batch`, and so does
`test_read_parquet_without_field_ids_with_struct` in `projection.rs`.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -192,6 +194,272 @@ 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(),
+ })
+ }
+ (DataType::List(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::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(Error::new(
+ ErrorKind::DataInvalid,
+ format!("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);
+ }
+ }
+ // Reachable for id-less files with nested types because the reader
only
+ // applies name mapping to top-level fields; error rather than silently
+ // nulling every child.
+ if !source_fields.is_empty() && source_by_id.is_empty() {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ "cannot reconcile struct fields by id: no source field carries
a field id",
+ ));
+ }
+
+ 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 => {
+ // Nested fields absent from the file only support
rule #4
+ // (null) of the spec's Column Projection rules; rule
#3
+ // (initial-default) is only wired up for top-level
columns
+ // via ColumnSource::Add.
+ 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",
+ target_field.name()
Review Comment:
The `FeatureUnsupported` error is better than a silent null. The spec does
define this case though. [Default
values](https://iceberg.apache.org/spec/#default-values) says sub-field
defaults are tracked in the sub-field's metadata, and its table gives `{"x":
3}` with a `point.y` default of `0` as `{"x": 3, "y": 0}`. Reads of older files
on a v3 table with a defaulted nested field will fail until this is
implemented. I couldn't find an issue for it. Could you open one and reference
it in this comment and in the error message, so the gap is tracked after merge?
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -831,25 +1099,26 @@ impl RecordBatchTransformer {
//
// At this point, all field IDs in the source schema are
trustworthy.
// No conflict detection needed - schema resolution happened
in reader.rs.
- let field_by_id =
field_id_to_source_schema_map.get(field_id).map(
- |(source_field, source_index)| {
- if
source_field.data_type().equals_datatype(target_type) {
- ColumnSource::PassThrough {
- source_index: *source_index,
- }
- } else {
- ColumnSource::Promote {
- target_type: target_type.clone(),
- source_index: *source_index,
- }
- }
- },
- );
-
+ //
// Apply spec's fallback steps for "not present" fields.
// Rule #1 (constants) is handled at the beginning of this
function
- let column_source = if let Some(source) = field_by_id {
- source
+ let column_source = if let Some((source_field, source_index)) =
+ field_id_to_source_schema_map.get(field_id)
+ {
+ if source_field.data_type().equals_datatype(target_type) {
+ ColumnSource::PassThrough {
+ source_index: *source_index,
+ }
+ } else {
+ ColumnSource::Promote {
+ plan: PromotePlan::build(
+ source_field.data_type(),
+ target_type,
+ snapshot_schema,
+ )?,
+ source_index: *source_index,
+ }
+ }
Review Comment:
[Schema Evolution](https://iceberg.apache.org/spec/#schema-evolution) allows
reordering and renaming inside any struct, not just the top-level schema, and
[Column Projection](https://iceberg.apache.org/spec/#column-projection)
requires projecting by id because "the table schema's column names and order
may change after a data file is written".
`DataType::equals_datatype` compares struct children by position, type, and
nullability. It ignores child names and metadata, including `PARQUET:field_id`.
So when a nested struct's children are reordered or renamed but the types still
line up, this check returns true and the column takes
`ColumnSource::PassThrough` without reaching `PromotePlan`. What happens when
the table has `s: struct<b: int (id 6), a: int (id 5)>` and the file has `s:
struct<a (id 5), b (id 6)>` with `a = [1, 2]` and `b = [100, 200]`? On this
head, `process_record_batch` returns a batch whose schema says position 0 is
`b`, but the array there holds `[1, 2]`. The struct array also keeps the file's
child order, so its data type no longer matches the batch schema. A rename with
no type change fails the same way: the array keeps the old child name.
`promote_evolved_nested_struct_via_process_record_batch` only gets the rename
right because the int-to-long promotion forces the `Promote` path.
Could we let the plan decide when to pass through?
```suggestion
match PromotePlan::build(
source_field.data_type(),
target_type,
snapshot_schema,
)? {
PromotePlan::PassThrough =>
ColumnSource::PassThrough {
source_index: *source_index,
},
plan => ColumnSource::Promote {
plan,
source_index: *source_index,
},
}
```
This needs the change at lines 320-325, or id-less maps start failing. With
both, reorder and rename tests pass, and all 1773 tests in `cargo test -p
iceberg --lib` pass. For a struct whose ids match but whose `Field`s aren't
strictly equal (a `doc` metadata entry, for example), the per-batch cost is one
`StructArray` rebuild from `Arc` clones.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -1144,20 +1404,418 @@ mod test {
use std::collections::HashMap;
use std::sync::Arc;
+ use arrow_array::cast::AsArray;
+ use arrow_array::types::{Int32Type, Int64Type};
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 parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use super::field_with_id;
use crate::arrow::build_partition_constant;
use crate::arrow::record_batch_transformer::{
- RecordBatchTransformer, RecordBatchTransformerBuilder,
+ PromotePlan, RecordBatchTransformer, RecordBatchTransformerBuilder,
};
use crate::spec::{Literal, NestedField, PrimitiveType, Schema, Struct,
Type};
+ fn simple_field(name: &str, ty: DataType, nullable: bool, value: &str) ->
Field {
+ Field::new(name, ty, nullable).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ value.to_string(),
+ )]))
+ }
Review Comment:
#2668 replaced this helper with the module-level `field_with_id(name,
data_type, nullable, field_id: i32)`, which this test module already imports at
line 1418. Could the new tests use `field_with_id` and drop `simple_field`, so
there's one helper for field-id metadata again?
--
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]
