Name mapping is applied to top-level columns only, so nested fields of migrated files read as NULL #3374

#[tokio::test]
async fn name_mapping_nested_struct() {
    // 1: id int, 2: s struct<3: a int, 4: b int>
    let schema = Arc::new(
        Schema::builder()
            .with_schema_id(1)
            .with_fields(vec![
                NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
                NestedField::optional(
                    2,
                    "s",
                    Type::Struct(StructType::new(vec![
                        NestedField::optional(3, "a", Type::Primitive(PrimitiveType::Int)).into(),
                        NestedField::optional(4, "b", Type::Primitive(PrimitiveType::Int)).into(),
                    ])),
                )
                .into(),
            ])
            .build()
            .unwrap(),
    );

    // A Parquet file with no field ids at any level.
    let struct_fields = arrow_schema::Fields::from(vec![
        Field::new("a", DataType::Int32, true),
        Field::new("b", DataType::Int32, true),
    ]);
    let arrow_schema = Arc::new(ArrowSchema::new(vec![
        Field::new("id", DataType::Int32, true),
        Field::new("s", DataType::Struct(struct_fields.clone()), true),
    ]));

    // The mapping covers the struct's fields, as Iceberg Java's MappingUtil.create writes it.
    let name_mapping = Arc::new(NameMapping::new(vec![
        MappedField::new(Some(1), vec!["id".to_string()], vec![]),
        MappedField::new(Some(2), vec!["s".to_string()], vec![
            MappedField::new(Some(3), vec!["a".to_string()], vec![]),
            MappedField::new(Some(4), vec!["b".to_string()], vec![]),
        ]),
    ]));

    let tmp_dir = TempDir::new().unwrap();
    let path = format!("{}/1.parquet", tmp_dir.path().to_str().unwrap());
    let s_col = arrow_array::StructArray::new(
        struct_fields,
        vec![
            Arc::new(arrow_array::Int32Array::from(vec![0, 1, 2])) as ArrayRef,
            Arc::new(arrow_array::Int32Array::from(vec![0, -1, -2])) as ArrayRef,
        ],
        None,
    );
    let batch = RecordBatch::try_new(arrow_schema.clone(), vec![
        Arc::new(arrow_array::Int32Array::from(vec![0, 1, 2])) as ArrayRef,
        Arc::new(s_col) as ArrayRef,
    ])
    .unwrap();
    let mut writer = ArrowWriter::try_new(File::create(&path).unwrap(), batch.schema(), None).unwrap();
    writer.write(&batch).unwrap();
    writer.close().unwrap();

    for project_field_ids in [vec![1, 2], vec![2]] {
        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.clone())
                .with_data_file_format(DataFileFormat::Parquet)
                .with_schema(schema.clone())
                .with_project_field_ids(project_field_ids.clone())
                .with_case_sensitive(false)
                .with_name_mapping(Some(name_mapping.clone()))
                .build()
                .unwrap())]
            .into_iter(),
        )) as FileScanTaskStream;
        let batches: Vec<RecordBatch> =
            reader.read(tasks).unwrap().stream().try_collect().await.unwrap();
        let s = batches[0].column(batches[0].num_columns() - 1).as_struct();
        println!("{project_field_ids:?}: {} null structs, a={:?}, b={:?}", s.null_count(), s.column(0), s.column(1));
    }
}