mbutrovich commented on code in PR #3255:
URL: https://github.com/apache/iceberg-rust/pull/3255#discussion_r4138840644
##########
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:
What should this return for a nested field that is the source of an identity
partition? [Column
Projection](https://iceberg.apache.org/spec/#column-projection) resolves a
field id that's missing from a data file by first returning the partition value
when an [identity
transform](https://iceberg.apache.org/spec/#partition-transforms) exists for
it, and [Partitioning](https://iceberg.apache.org/spec/#partitioning) allows a
partition source column to be nested in a struct. `constants_map` already
stores nested source ids, because it [resolves the source with
`field_by_id`](https://github.com/apache/iceberg-rust/blob/608282436383fdcb092bc9dcf13d08dd9c492579/crates/iceberg/src/arrow/record_batch_transformer.rs#L90).
Only the [top-level
path](https://github.com/apache/iceberg-rust/blob/608282436383fdcb092bc9dcf13d08dd9c492579/crates/iceberg/src/arrow/record_batch_transformer.rs#L1016-L1050)
reads `constant_fields`, though.
I checked this with a table `s: struct<region: string (id 5), x: int (id
6)>` partitioned by `identity(s.region)` with partition value `"us"`, reading a
file whose `s` only has `x`. On this head, `process_record_batch` returns
`region` as `[null, null]`. On `main` the same read fails with `Unexpected =>
Arrow Schema Error`, so for this case the PR turns an error into wrong results.
Iceberg Java applies partition constants at every struct level in
[`ParquetValueReaders.structReader`](https://github.com/apache/iceberg/blob/5e7169168db3d34e29354c6f59ec4d6e420b8d2d/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java#L244-L260).
Could this arm check for a partition constant before it falls back to
`initial-default` or null, as the top-level path does for an absent column? One
way is to pass `constant_fields` into `PromotePlan::build`, add a `ChildPlan`
variant that holds the constant, and build its column with
`RecordBatchTransformer::create_column` in `apply_struct`. A test next to
`identity_partition_uses_constant_from_metadata` would cover it.
##########
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(),
+ }),
Review Comment:
Can the target type here be a `LargeList`? The target always comes from
[`schema_to_arrow_schema(snapshot_schema)`](https://github.com/apache/iceberg-rust/blob/608282436383fdcb092bc9dcf13d08dd9c492579/crates/iceberg/src/arrow/record_batch_transformer.rs#L809),
and
[`ToArrowSchemaConverter::list`](https://github.com/apache/iceberg-rust/blob/608282436383fdcb092bc9dcf13d08dd9c492579/crates/iceberg/src/arrow/schema.rs#L631-L642)
always builds `DataType::List`.
If there's no path to a `LargeList` target, could we drop
`PromotePlan::LargeList`, this arm, and the `List` to `LargeList` conversion at
lines 411-420, and let a `LargeList` target fall through to the `cannot
promote` error? That conversion has no test, and
`promote_large_list_element_struct_fills_added_field_by_id` reaches this arm
only by calling `PromotePlan::build` directly.
`promote_large_list_file_column_to_list` would still cover a `LargeList` file
column.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -1136,19 +1422,693 @@ 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 super::field_with_id;
- use crate::arrow::build_partition_constant;
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, StructType, Type,
};
- use crate::spec::{Literal, NestedField, PrimitiveType, Schema, Struct,
Type};
+
+ fn promote(source: &ArrayRef, target: &DataType, schema: &Schema) ->
crate::Result<ArrayRef> {
+ PromotePlan::build(source.data_type(), target, schema)?.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 assert_existing_field_kept(s: &StructArray, expected_existing: &[i32]) {
+ assert_eq!(
+ s.column(0).as_primitive::<Int32Type>().values(),
+ expected_existing
+ );
+ }
+
+ fn assert_added_field_null(s: &StructArray) {
+ assert_eq!(s.column(1).null_count(), s.len());
+ }
+
+ 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_middle_field_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 target = DataType::Struct(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),
+ ]));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let s = out.as_struct();
+ assert_eq!(s.num_columns(), 3);
+ assert_eq!(s.column(0).as_primitive::<Int32Type>().values(), &[1, 2]);
+ assert_eq!(s.column(1).null_count(), 2);
+ let cc = s.column(2).as_string::<i32>();
+ assert_eq!((cc.value(0), cc.value(1)), ("x", "y"));
+ }
+
+ #[test]
+ fn promote_struct_fills_appended_field_by_id() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![
+ field_with_id("a", DataType::Int32, true, 1),
+ field_with_id("b", DataType::Utf8, true, 2),
+ ]),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["x", "y"])) as ArrayRef,
+ ],
+ None,
+ )) as ArrayRef;
+ let target = DataType::Struct(Fields::from(vec![
+ field_with_id("a", DataType::Int32, true, 1),
+ field_with_id("b", DataType::Utf8, true, 2),
+ field_with_id("c", DataType::Int32, true, 3),
+ ]));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let s = out.as_struct();
+ assert_eq!(s.num_columns(), 3);
+ assert_eq!(s.column(0).as_primitive::<Int32Type>().values(), &[1, 2]);
+ let bb = s.column(1).as_string::<i32>();
+ assert_eq!((bb.value(0), bb.value(1)), ("x", "y"));
+ assert_eq!(s.column(2).null_count(), 2);
+ }
+
+ #[test]
+ fn promote_struct_missing_field_before_nested_list_struct() {
+ let elem_field = Arc::new(field_with_id("element",
unevolved_struct_type(), true, 4));
+ let list = Arc::new(ListArray::new(
+ elem_field.clone(),
+ OffsetBuffer::new(vec![0, 1, 2].into()),
+ unevolved_struct_data(vec![10, 20]),
+ None,
+ )) as ArrayRef;
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![
+ field_with_id("s", DataType::Utf8, true, 1),
+ field_with_id("ev", DataType::List(elem_field.clone()), true,
3),
+ ]),
+ vec![
+ Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
+ list,
+ ],
+ None,
+ )) as ArrayRef;
+ let target = DataType::Struct(Fields::from(vec![
+ field_with_id("s", DataType::Utf8, true, 1),
+ field_with_id("gap", DataType::Int32, true, 2),
+ field_with_id("ev", DataType::List(elem_field), true, 3),
+ ]));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let st = out.as_struct();
+ assert_eq!(st.num_columns(), 3);
+ assert_eq!(st.column(1).null_count(), 2);
+ let ev = st.column(2).as_list::<i32>();
+ assert_eq!(ev.len(), 2);
+ assert_eq!(
+ ev.value(0)
+ .as_struct()
+ .column(0)
+ .as_primitive::<Int32Type>()
+ .value(0),
+ 10
+ );
+ }
+
+ #[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 lst = out.as_list::<i32>();
+ assert_eq!(lst.len(), 2);
+ let elements = lst.values().as_struct();
+ assert_existing_field_kept(elements, &[10, 20]);
+ assert_added_field_null(elements);
+ }
+
+ #[test]
+ fn promote_map_value_struct_fills_added_field_by_id() {
+ let entries = StructArray::new(
+ Fields::from(vec![
+ field_with_id("key", DataType::Utf8, false, 7),
+ field_with_id("value", unevolved_struct_type(), true, 8),
+ ]),
+ vec![
+ Arc::new(StringArray::from(vec!["k1", "k2"])) as ArrayRef,
+ unevolved_struct_data(vec![100, 200]),
+ ],
+ None,
+ );
+ let source = Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", entries.data_type().clone(),
false)),
+ OffsetBuffer::new(vec![0, 1, 2].into()),
+ entries,
+ None,
+ false,
+ )) as ArrayRef;
+ let target_entries = DataType::Struct(Fields::from(vec![
+ field_with_id("key", DataType::Utf8, false, 7),
+ field_with_id("value", evolved_struct_type(), true, 8),
+ ]));
+ let target = DataType::Map(
+ Arc::new(Field::new("entries", target_entries, false)),
+ false,
+ );
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let m = out.as_map();
+ assert_eq!(m.len(), 2);
+ let entries = m.entries();
+ let ks = entries.column(0).as_string::<i32>();
+ assert_eq!((ks.value(0), ks.value(1)), ("k1", "k2"));
+ let values = entries.column(1).as_struct();
+ assert_existing_field_kept(values, &[100, 200]);
+ assert_added_field_null(values);
+ }
+
+ #[test]
+ fn promote_large_list_element_struct_fills_added_field_by_id() {
+ let source = Arc::new(LargeListArray::new(
+ Arc::new(field_with_id("element", unevolved_struct_type(), true,
4)),
+ OffsetBuffer::new(vec![0i64, 1, 2].into()),
+ unevolved_struct_data(vec![7, 8]),
+ None,
+ )) as ArrayRef;
+ let target = DataType::LargeList(Arc::new(field_with_id(
+ "element",
+ evolved_struct_type(),
+ true,
+ 4,
+ )));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let lst = out.as_list::<i64>();
+ assert_eq!(lst.len(), 2);
+ let elements = lst.values().as_struct();
+ assert_existing_field_kept(elements, &[7, 8]);
+ assert_added_field_null(elements);
+ }
+
+ #[test]
+ fn promote_struct_renames_field_by_id() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![field_with_id("x_old", DataType::Int32, true,
5)]),
+ vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef],
+ None,
+ )) as ArrayRef;
+ let target = DataType::Struct(Fields::from(vec![field_with_id(
+ "x",
+ DataType::Int32,
+ true,
+ 5,
+ )]));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let expected = StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef],
+ None,
+ );
+ assert_eq!(out.as_struct(), &expected);
+ }
+
+ #[test]
+ fn promote_struct_dropped_and_readded_same_name_nulls_by_id() {
+ let file = StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef],
+ None,
+ );
+ let out = transform_top_level(
+ NestedField::optional(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![
+ NestedField::optional(6, "x",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])),
+ ),
+ Arc::new(file),
+ )
+ .unwrap();
+ let expected = StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 6)]),
+ vec![Arc::new(Int32Array::from(vec![None, None])) as ArrayRef],
+ None,
+ );
+ assert_eq!(out.as_struct(), &expected);
+ }
+
+ #[test]
+ fn promote_struct_promotes_child_primitive() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef],
+ None,
+ )) as ArrayRef;
+ let target = DataType::Struct(Fields::from(vec![field_with_id(
+ "x",
+ DataType::Int64,
+ true,
+ 5,
+ )]));
+
+ let out = promote(&source, &target, &empty_schema()).unwrap();
+ let s = out.as_struct();
+ assert_eq!(s.column(0).as_primitive::<Int64Type>().values(), &[1, 2]);
+ }
+
+ #[test]
+ fn promote_struct_preserves_null_parent_rows() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![field_with_id("x", DataType::Int32, true, 5)]),
+ vec![Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef],
+ Some(NullBuffer::from(vec![true, false])),
+ )) as ArrayRef;
+
+ let out = promote(&source, &evolved_struct_type(),
&empty_schema()).unwrap();
+ let s = out.as_struct();
+ assert!(!s.is_null(0));
+ assert!(s.is_null(1));
+ assert_eq!(s.column(0).as_primitive::<Int32Type>().value(0), 10);
+ assert_added_field_null(s);
+ }
+
+ #[test]
+ fn promote_struct_without_source_field_ids_errors() {
+ let source = Arc::new(StructArray::new(
+ Fields::from(vec![Field::new("x", DataType::Int32, true)]),
+ vec![Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef],
+ None,
+ )) as ArrayRef;
+
+ let err = promote(&source, &evolved_struct_type(),
&empty_schema()).unwrap_err();
+ assert!(err.to_string().contains("do not all have field ids"));
+ }
Review Comment:
Could the error tests assert on `err.kind()` along with the message? The
split between `FeatureUnsupported` and `DataInvalid` is what callers act on.
`promote_nested_field_with_initial_default_errors` checks for the substring
`initial-default`, which the `DataInvalid` message at line 378 also contains,
so the check doesn't tell the two errors apart.
Could we also add a case where only some source children carry field ids?
The description lists it as an error path, and no test covers it yet.
--
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]
