laskoviymishka commented on code in PR #3629:
URL: https://github.com/apache/iceberg-rust/pull/3629#discussion_r4508052684
##########
crates/iceberg/src/arrow/reader/projection.rs:
##########
@@ -2482,4 +2544,535 @@ message schema {
.unwrap();
assert_eq!(result, vec![expected]);
}
+
+ #[test]
+ fn test_coerce_fixed_size_lists_rewrites_nested() {
Review Comment:
Could this build the expected schema and assert equality with it? With
`PARQUET:field_id` metadata and a non-nullable element on the input fields, it
would pin what name mapping and fallback IDs rely on. Right now it only checks
`matches!(…, DataType::List(_))`, so a rewrite that dropped either would still
pass. An `Arc::ptr_eq` on `plain` would cover the sharing the doc promises.
##########
crates/iceberg/src/arrow/reader/projection.rs:
##########
@@ -533,6 +533,68 @@ fn try_map_fields(
Ok(Fields::from(mapped))
}
+/// Rewrites file-side Arrow `FixedSizeList` hints to `List`.
+///
+/// parquet-rs picks its list reader from the embedded Arrow schema. Iceberg
has no
+/// fixed-size list type, so a `FixedSizeList` hint must become `List` before
the stream
+/// reader is built. That uses the 3-level LIST reader (no null-row padding, no
+/// `PromotePlan` conversion). Returns `None` when the schema has no
`FixedSizeList`.
+/// Unchanged fields are shared; nothing is allocated before the first change.
+///
+/// Same placement as the list-view rewrite in #3467: call this before name
mapping and
+/// fallback IDs so nested IDs land on the rewritten `List` fields.
Review Comment:
Could the doc stick to what the function does? "no null-row padding, no
`PromotePlan` conversion" describes the earlier version of this PR, and #3467
is still open, so the placement note points at code that isn't on `main`. The
[`pipeline.rs`
comment](https://github.com/apache/iceberg-rust/blob/aa525f1256e319aa3ddc0e7529123436070f5043/crates/iceberg/src/arrow/reader/pipeline.rs#L691-L692)
has the same #3467 reference.
```suggestion
/// reader is built. Returns `None` when the schema has no `FixedSizeList`.
/// Unchanged fields are shared; nothing is allocated before the first
change.
///
/// Call this before name mapping and fallback IDs so nested IDs land on the
rewritten
/// `List` fields.
```
##########
crates/iceberg/src/arrow/reader/projection.rs:
##########
@@ -2482,4 +2544,535 @@ message schema {
.unwrap();
assert_eq!(result, vec![expected]);
}
+
+ #[test]
+ fn test_coerce_fixed_size_lists_rewrites_nested() {
+ let item = |ty| Arc::new(Field::new("item", ty, true));
+ let schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("plain", DataType::Int32, true),
+ Field::new(
+ "fsl",
+ DataType::FixedSizeList(item(DataType::Int32), 2),
+ true,
+ ),
+ Field::new(
+ "nested",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "inner",
+ DataType::FixedSizeList(item(DataType::Int32), 3),
+ true,
+ )])),
+ true,
+ ),
+ Field::new(
+ "list_of_fsl",
+ DataType::List(Arc::new(Field::new(
+ "element",
+ DataType::FixedSizeList(item(DataType::Int32), 2),
+ true,
+ ))),
+ true,
+ ),
+ ]));
+
+ let coerced = super::coerce_fixed_size_lists(&schema).expect("should
rewrite");
+ assert!(matches!(
+ coerced.field_with_name("fsl").unwrap().data_type(),
+ DataType::List(_)
+ ));
+ let DataType::Struct(fields) =
coerced.field_with_name("nested").unwrap().data_type()
+ else {
+ panic!("nested should stay Struct");
+ };
+ assert!(matches!(fields[0].data_type(), DataType::List(_)));
+ let DataType::List(element) =
coerced.field_with_name("list_of_fsl").unwrap().data_type()
+ else {
+ panic!("list_of_fsl should stay List");
+ };
+ assert!(matches!(element.data_type(), DataType::List(_)));
+ assert!(super::coerce_fixed_size_lists(&coerced).is_none());
+ }
+
+ /// parquet-rs can return a LIST as `FixedSizeList` when the file's
embedded Arrow schema
+ /// says so. Rewrite the hint to `List` before reading so the scan never
sees FixedSizeList.
+ #[tokio::test]
+ async fn test_read_fixed_size_list_column_as_list() {
+ use arrow_array::{FixedSizeListArray, Int32Array};
+ use arrow_buffer::{NullBuffer, OffsetBuffer};
+
+ use crate::spec::ListType;
+
+ let with_id = |field: Field, id: i32| {
+ field.with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ id.to_string(),
+ )]))
+ };
+
+ // Row 0: [1, 2]; row 1: null; row 2: [5, 6].
+ let values = Arc::new(Int32Array::from(vec![
+ Some(1),
+ Some(2),
+ None,
+ None,
+ Some(5),
+ Some(6),
+ ])) as ArrayRef;
+ let row_nulls = NullBuffer::from(vec![true, false, true]);
+ let optional_element = Arc::new(with_id(Field::new("element",
DataType::Int32, true), 3));
+ let required_element = Arc::new(with_id(Field::new("element",
DataType::Int32, false), 5));
+ let optional_fsl = FixedSizeListArray::new(
+ Arc::clone(&optional_element),
+ 2,
+ Arc::clone(&values),
+ Some(row_nulls.clone()),
+ );
+ let required_fsl =
+ FixedSizeListArray::new(Arc::clone(&required_element), 2, values,
Some(row_nulls));
+ let file_schema = Arc::new(ArrowSchema::new(vec![
+ with_id(
+ Field::new(
+ "optional",
+ DataType::FixedSizeList(optional_element, 2),
+ true,
+ ),
+ 2,
+ ),
+ with_id(
+ Field::new(
+ "required",
+ DataType::FixedSizeList(required_element, 2),
+ true,
+ ),
+ 4,
+ ),
+ with_id(Field::new("id", DataType::Int32, true), 1),
+ ]));
+ let batch = RecordBatch::try_new(file_schema.clone(), vec![
+ Arc::new(optional_fsl) as ArrayRef,
+ Arc::new(required_fsl) as ArrayRef,
+ Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef,
+ ])
+ .unwrap();
+
+ let tmp_dir = TempDir::new().unwrap();
+ let path = format!("{}/fsl.parquet", tmp_dir.path().to_str().unwrap());
+ let mut writer =
+ ArrowWriter::try_new(File::create(&path).unwrap(), file_schema,
None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+
+ let table_schema = Arc::new(
+ Schema::builder()
+ .with_fields(vec![
+ NestedField::optional(
+ 2,
+ "optional",
+ Type::List(ListType::new(
+ NestedField::list_element(
+ 3,
+ Type::Primitive(PrimitiveType::Int),
+ false,
+ )
+ .into(),
+ )),
+ )
+ .into(),
+ NestedField::optional(
+ 4,
+ "required",
+ Type::List(ListType::new(
+ NestedField::list_element(5,
Type::Primitive(PrimitiveType::Int), true)
+ .into(),
+ )),
+ )
+ .into(),
+ NestedField::optional(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current()).build();
+ let tasks = Box::pin(futures::stream::iter(
+ vec![Ok(FileScanTask::builder()
+
.with_file_size_in_bytes(std::fs::metadata(&path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(table_schema)
+ .with_project_field_ids(vec![1, 2, 4])
+ .with_case_sensitive(false)
+ .build()
+ .unwrap())]
+ .into_iter(),
+ )) as FileScanTaskStream;
+
+ let result = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect::<Vec<RecordBatch>>()
+ .await
+ .unwrap();
+ assert_eq!(result.len(), 1);
+ let out = &result[0];
+ assert_eq!(out.num_rows(), 3);
+
+ let expected_offsets = OffsetBuffer::new(vec![0, 2, 2, 4].into());
+ let expected_values = Arc::new(Int32Array::from(vec![1, 2, 5, 6])) as
ArrayRef;
+ let expected_nulls = NullBuffer::from(vec![true, false, true]);
+ let schema = out.schema();
+ for (name, element_nullable) in [("optional", true), ("required",
false)] {
+ let col = schema.index_of(name).unwrap();
+ let list = out.column(col).as_list::<i32>();
+ assert_eq!(list.offsets(), &expected_offsets, "{name} offsets");
+ assert_eq!(
+ list.values().as_ref(),
+ expected_values.as_ref(),
+ "{name} values"
+ );
+ assert_eq!(list.nulls(), Some(&expected_nulls), "{name} nulls");
+ let DataType::List(element) = schema.field(col).data_type() else {
+ panic!("{name} should be List");
+ };
+ assert_eq!(
+ element.is_nullable(),
+ element_nullable,
+ "{name} element nullability"
+ );
+ }
+ let id_col = schema.index_of("id").unwrap();
+ assert_eq!(
+ out.column(id_col).as_ref(),
+ &Int32Array::from(vec![1, 2, 3]) as &dyn Array
+ );
+ }
+
+ /// Migrated files have no field IDs. Cover name mapping and fallback IDs
with a null row,
+ /// nested FixedSizeList shapes, and a required top-level element.
+ #[tokio::test]
+ async fn test_read_fixed_size_list_without_field_ids() {
+ use arrow_array::{FixedSizeListArray, Int32Array, ListArray, MapArray,
StructArray};
+ use arrow_buffer::{NullBuffer, OffsetBuffer};
+
+ use crate::spec::{ListType, MapType};
+
+ let item = |nullable| Arc::new(Field::new("item", DataType::Int32,
nullable));
+ let values = Arc::new(Int32Array::from(vec![
+ Some(1),
+ Some(2),
+ None,
+ None,
+ Some(5),
+ Some(6),
+ ])) as ArrayRef;
+ let row_nulls = NullBuffer::from(vec![true, false, true]);
+ let fsl = |nullable_element| {
+ FixedSizeListArray::new(
+ item(nullable_element),
+ 2,
+ Arc::clone(&values),
+ Some(row_nulls.clone()),
+ )
+ };
+
+ let top = fsl(false);
+ let in_struct = StructArray::new(
+ Fields::from(vec![Field::new(
+ "inner",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ )]),
+ vec![Arc::new(fsl(true)) as ArrayRef],
+ None,
+ );
+ let map_entries = StructArray::new(
+ Fields::from(vec![
+ Field::new("keys", DataType::Utf8, false),
+ Field::new("values", DataType::FixedSizeList(item(true), 2),
true),
+ ]),
+ vec![
+ Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
+ Arc::new(FixedSizeListArray::new(
+ item(true),
+ 2,
+ Arc::new(Int32Array::from(vec![
+ Some(10),
+ Some(11),
+ Some(20),
+ Some(21),
+ ])) as ArrayRef,
+ None,
+ )) as ArrayRef,
+ ],
+ None,
+ );
+ let as_map_value = MapArray::new(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(map_entries.fields().clone()),
+ false,
+ )),
+ OffsetBuffer::new(vec![0, 1, 1, 2].into()),
+ map_entries,
+ Some(row_nulls.clone()),
+ false,
+ );
+ let list_of_fsl = ListArray::new(
+ Arc::new(Field::new(
+ "element",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ )),
+ OffsetBuffer::new(vec![0, 1, 1, 2].into()),
+ Arc::new(FixedSizeListArray::new(
+ item(true),
+ 2,
+ Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(5),
Some(6)])) as ArrayRef,
+ None,
+ )) as ArrayRef,
+ Some(row_nulls.clone()),
+ );
+ let nested_fsl = FixedSizeListArray::new(
+ Arc::new(Field::new(
+ "item",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ )),
+ 1,
+ Arc::new(FixedSizeListArray::new(
+ item(true),
+ 2,
+ Arc::clone(&values),
+ None,
+ )) as ArrayRef,
+ Some(row_nulls),
+ );
+
+ let file_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("top", DataType::FixedSizeList(item(false), 2), true),
+ Field::new(
+ "s",
+ DataType::Struct(Fields::from(vec![Field::new(
+ "inner",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ )])),
+ true,
+ ),
+ Field::new(
+ "m",
+ DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(Fields::from(vec![
+ Field::new("keys", DataType::Utf8, false),
+ Field::new("values",
DataType::FixedSizeList(item(true), 2), true),
+ ])),
+ false,
+ )),
+ false,
+ ),
+ true,
+ ),
+ Field::new(
+ "list_of_fsl",
+ DataType::List(Arc::new(Field::new(
+ "element",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ ))),
+ true,
+ ),
+ Field::new(
+ "nested_fsl",
+ DataType::FixedSizeList(
+ Arc::new(Field::new(
+ "item",
+ DataType::FixedSizeList(item(true), 2),
+ true,
+ )),
+ 1,
+ ),
+ true,
+ ),
+ ]));
+ let batch = RecordBatch::try_new(file_schema.clone(), vec![
+ Arc::new(top) as ArrayRef,
+ Arc::new(in_struct) as ArrayRef,
+ Arc::new(as_map_value) as ArrayRef,
+ Arc::new(list_of_fsl) as ArrayRef,
+ Arc::new(nested_fsl) as ArrayRef,
+ ])
+ .unwrap();
+
+ let tmp_dir = TempDir::new().unwrap();
+ let path = format!("{}/fsl_no_ids.parquet",
tmp_dir.path().to_str().unwrap());
+ let mut writer =
+ ArrowWriter::try_new(File::create(&path).unwrap(), file_schema,
None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ let file_size = std::fs::metadata(&path).unwrap().len();
+
+ let int_ty = Type::Primitive(PrimitiveType::Int);
+ let list_of = |id, required| {
+ Type::List(ListType::new(
+ NestedField::list_element(id, int_ty.clone(), required).into(),
+ ))
+ };
+ let table_schema = Arc::new(
+ Schema::builder()
+ .with_fields(vec![
+ NestedField::optional(1, "top", list_of(6, true)).into(),
+ NestedField::optional(
+ 2,
+ "s",
+ Type::Struct(StructType::new(vec![
+ NestedField::optional(7, "inner", list_of(8,
false)).into(),
+ ])),
+ )
+ .into(),
+ NestedField::optional(
+ 3,
+ "m",
+ Type::Map(MapType::optional(
+ 9,
+ Type::Primitive(PrimitiveType::String),
+ 10,
+ list_of(11, false),
+ )),
+ )
+ .into(),
+ NestedField::optional(
+ 4,
+ "list_of_fsl",
+ Type::List(ListType::new(
+ NestedField::list_element(12, list_of(13, false),
false).into(),
+ )),
+ )
+ .into(),
+ NestedField::optional(
+ 5,
+ "nested_fsl",
+ Type::List(ListType::new(
+ NestedField::list_element(14, list_of(15, false),
false).into(),
+ )),
+ )
+ .into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let name_mapping = Arc::new(NameMapping::new(vec![
+ MappedField::new(Some(1), vec!["top".to_string()],
vec![MappedField::new(
+ Some(6),
+ vec!["element".to_string()],
+ vec![],
+ )]),
+ MappedField::new(Some(2), vec!["s".to_string()],
vec![MappedField::new(
+ Some(7),
+ vec!["inner".to_string()],
+ vec![MappedField::new(
+ Some(8),
+ vec!["element".to_string()],
+ vec![],
+ )],
+ )]),
+ MappedField::new(Some(3), vec!["m".to_string()], vec![
+ MappedField::new(Some(9), vec!["key".to_string()], vec![]),
+ MappedField::new(Some(10), vec!["value".to_string()],
vec![MappedField::new(
+ Some(11),
+ vec!["element".to_string()],
+ vec![],
+ )]),
+ ]),
+ MappedField::new(Some(4), vec!["list_of_fsl".to_string()], vec![
+ MappedField::new(Some(12), vec!["element".to_string()], vec![
+ MappedField::new(Some(13), vec!["element".to_string()],
vec![]),
+ ]),
+ ]),
+ MappedField::new(Some(5), vec!["nested_fsl".to_string()], vec![
+ MappedField::new(Some(14), vec!["element".to_string()], vec![
+ MappedField::new(Some(15), vec!["element".to_string()],
vec![]),
+ ]),
+ ]),
+ ]));
+
+ let assert_top_list = |out: &RecordBatch| {
+ assert_eq!(out.num_rows(), 3);
+ let top = out
+ .column(out.schema().index_of("top").unwrap())
+ .as_list::<i32>();
+ assert_eq!(
+ top.nulls(),
+ Some(&NullBuffer::from(vec![true, false, true]))
+ );
+ assert_eq!(
+ top.values().as_ref(),
+ &Int32Array::from(vec![1, 2, 5, 6]) as &dyn Array
+ );
+ let schema = out.schema();
+ for name in ["s", "m", "list_of_fsl", "nested_fsl"] {
Review Comment:
Could `assert_top_list` compare `s`, `m`, `list_of_fsl` and `nested_fsl`
against expected arrays, like `top`? The "not `FixedSizeList`" check can't fail
for `s` and `m`. With a debug print added, the fallback pass reads `s.inner` as
`[null, null, null]`, since id-less struct children [read as
absent](https://github.com/apache/iceberg-rust/blob/aa525f1256e319aa3ddc0e7529123436070f5043/crates/iceberg/src/arrow/record_batch_transformer.rs#L335-L341),
and the test still passes. Expecting that null under fallback and the values
under name mapping would pin both.
--
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]
