ListView columns read as NULL and shift later columns to the wrong Parquet leaf #3393

#[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}"),
        }
    }
}