[I] FixedSizeList columns fail to read with "cannot promote FixedSizeList to List" [iceberg-rust]

["comphead (via GitHub)" <gi...@apache.org>]

comphead opened a new issue, #3471:
URL: https://github.com/apache/iceberg-rust/issues/3471

   ### Apache Iceberg Rust version
   
   main @ `f83bec422`
   
   ### Describe the bug
   
   parquet-rs reads a Parquet LIST as `FixedSizeList` when the file's embedded Arrow schema says so, for example in a file that arrow-rs wrote from a `FixedSizeListArray`. Projecting such a column fails. `PromotePlan::build` in `arrow/record_batch_transformer.rs` converts a `List` or `LargeList` from the file to the table's `List`, but it has no `FixedSizeList` arm and returns "cannot promote". Iceberg has no fixed-size list type, so the table column is a plain `list`. Projecting only the other columns works.
   
   This is the same kind of problem as #3393, an Arrow list type from the embedded schema that the reader does not normalize. #3467 fixes the list view case by reading list views as lists.
   
   ### To Reproduce
   
   Added to the tests in `crates/iceberg/src/arrow/reader/projection.rs`:
   
   ```rust
   #[tokio::test]
   async fn fixed_size_list_column() {
       use arrow_array::types::Int32Type;
       use arrow_array::{FixedSizeListArray, Int32Array};
   
       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())]))
       };
   
       // 2: fsl fixed_size_list<3: int, 2>, then 1: id int, with field IDs at every level.
       let element = Arc::new(with_id(Field::new("element", DataType::Int32, true), 3));
       let fsl = FixedSizeListArray::new(
           element.clone(),
           2,
           Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])),
           None,
       );
       let file_schema = Arc::new(ArrowSchema::new(vec![
           with_id(Field::new("fsl", DataType::FixedSizeList(element, 2), true), 2),
           with_id(Field::new("id", DataType::Int32, true), 1),
       ]));
       let batch = RecordBatch::try_new(file_schema.clone(), vec![
           Arc::new(fsl) 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_field = NestedField::list_element(3, Type::Primitive(PrimitiveType::Int), false);
       let schema = Arc::new(
           Schema::builder()
               .with_fields(vec![
                   NestedField::optional(2, "fsl", Type::List(ListType::new(element_field.into()))).into(),
                   NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
               ])
               .build()
               .unwrap(),
       );
       let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
       for projection in [vec![2, 1], vec![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
   [2, 1]: DataInvalid => cannot promote FixedSizeList(Field { name: "element", data_type: Int32, nullable: true, metadata: {"PARQUET:field_id": "3"} }, 2) to List(Field { name: "element", data_type: Int32, nullable: true, metadata: {"PARQUET:field_id": "3"} })
   [1]: id = [Some(1), Some(2), Some(3)]
   ```
   
   ### Expected behavior
   
   `fsl` reads as a `List` with the values `[[1, 2], [3, 4], [5, 6]]`, and `id` reads as `[1, 2, 3]`.
   
   There are two possible fixes, and I have not tried either. One rewrites `FixedSizeList` to `List` in the schema hint, next to `coerce_list_views` from #3467. The other converts it to `List` in `PromotePlan`, as `apply` already does for `LargeList`.
   


-- 
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: issues-unsubscribe@iceberg.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org