comphead opened a new issue, #3393:
URL: https://github.com/apache/iceberg-rust/issues/3393
### Apache Iceberg Rust version
main @ `b2e90754b`
### Describe the bug
parquet-rs reads a Parquet LIST as `ListView` or `LargeListView` when the
file's embedded Arrow schema asks for it, for example in a file that arrow-rs
wrote from a `ListViewArray`. The reader builds its projection mask by walking
the Arrow schema with arrow-schema's `Fields::filter_leaves`, which treats
`ListView` and `LargeListView` as leaves. Even when the file has field IDs at
every level:
- The list's element is never matched by its field ID, so the column reads
as NULL whenever another column is projected.
- A `ListView` of structs counts as one leaf where Parquet has one per
struct field, so the leaf index of every later column is off. A plain column
after it is read from the wrong Parquet leaf and comes back NULL.
### To Reproduce
Added to the tests in `crates/iceberg/src/arrow/reader/projection.rs`:
```rust
#[tokio::test]
async fn list_view_shifts_projected_leaves() {
use arrow_array::types::Int32Type;
use arrow_array::{Int32Array, ListViewArray, StructArray};
use arrow_buffer::ScalarBuffer;
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())]))
};
let int = |id, name: &str| NestedField::optional(id, name,
Type::Primitive(PrimitiveType::Int)).into();
// 2: lv list_view<3: struct<4: a int, 5: b int>>, then 1: id int, with
field IDs at every level.
let a = Arc::new(with_id(Field::new("a", DataType::Int32, true), 4));
let b = Arc::new(with_id(Field::new("b", DataType::Int32, true), 5));
let structs = StructArray::new(
vec![a, b].into(),
vec![
Arc::new(Int32Array::from(vec![10, 20, 30])) as ArrayRef,
Arc::new(Int32Array::from(vec![-10, -20, -30])) as ArrayRef,
],
None,
);
let element = Arc::new(with_id(Field::new("element",
structs.data_type().clone(), true), 3));
let lv = ListViewArray::new(
element.clone(),
ScalarBuffer::from(vec![0, 1, 2]),
ScalarBuffer::from(vec![1, 1, 1]),
Arc::new(structs),
None,
);
let file_schema = Arc::new(ArrowSchema::new(vec![
with_id(Field::new("lv", DataType::ListView(element), true), 2),
with_id(Field::new("id", DataType::Int32, true), 1),
]));
let batch = RecordBatch::try_new(file_schema.clone(), vec![
Arc::new(lv) as ArrayRef,
Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef,
])
.unwrap();
let tmp_dir = TempDir::new().unwrap();
let path = format!("{}/1.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 element_type = Type::Struct(StructType::new(vec![int(4, "a"), int(5,
"b")]));
let element_field = NestedField::list_element(3, element_type,
false).into();
let schema = Arc::new(
Schema::builder()
.with_fields(vec![
NestedField::optional(2, "lv",
Type::List(ListType::new(element_field))).into(),
int(1, "id"),
])
.build()
.unwrap(),
);
let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current()).build();
for projection in [vec![1], vec![2, 1]] {
let task = FileScanTask::builder()
.with_file_size_in_bytes(std::fs::metadata(&path).unwrap().len())
.with_start(0)
.with_length(0)
.with_data_file_path(path.clone())
.with_data_file_format(DataFileFormat::Parquet)
.with_schema(schema.clone())
.with_project_field_ids(projection.clone())
.with_case_sensitive(false)
.build()
.unwrap();
let tasks = Box::pin(futures::stream::iter([Ok(task)])) as
FileScanTaskStream;
match
reader.clone().read(tasks).unwrap().stream().try_collect::<Vec<RecordBatch>>().await
{
Ok(batches) => {
let id =
batches[0].column_by_name("id").unwrap().as_primitive::<Int32Type>();
println!("{projection:?}: id = {:?}",
id.iter().collect::<Vec<_>>());
}
Err(e) => println!("{projection:?}: {e}"),
}
}
}
```
```text
[1]: id = [None, None, None]
[2, 1]: Unexpected => Arrow Schema Error, source: Invalid argument error:
Incorrect number of arrays for StructArray fields, expected 2 got 1
```
With `id int, lv list_view<int>` instead, projecting both columns returns
`lv` as NULL in every row. Projecting only `lv` returns its values, because no
leaf matches and the mask falls back to reading every column.
### Expected behavior
The same results as for a `List` column: `id` is `[1, 2, 3]` for both
projections, and `lv` returns its elements.
One possible fix, not tried yet, is to rewrite `ListView` and
`LargeListView` to `List` and `LargeList` in the schema hint the reader passes
to parquet-rs, since Iceberg reads lists as `List` anyway.
--
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]
