Reading a name-mapped map fails with "Map field must have exactly 2 fields" when none of the map value's leaves are in the file #3557

#[tokio::test]
async fn map_value_with_all_leaves_missing() {
    use arrow_array::{Int32Array, MapArray, StringArray, StructArray};
    use arrow_buffer::{OffsetBuffer, ScalarBuffer};
    use arrow_schema::Fields;

    // 1: id int, 2: m map<3: string, 4: struct<5: x int>>, mapped by name
    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,
                    "m",
                    Type::Map(crate::spec::MapType::new(
                        NestedField::map_key_element(3, Type::Primitive(PrimitiveType::String))
                            .into(),
                        NestedField::map_value_element(
                            4,
                            Type::Struct(StructType::new(vec![
                                NestedField::optional(5, "x", Type::Primitive(PrimitiveType::Int))
                                    .into(),
                            ])),
                            false,
                        )
                        .into(),
                    )),
                )
                .into(),
            ])
            .build()
            .unwrap(),
    );
    let name_mapping = Arc::new(NameMapping::new(vec![
        MappedField::new(Some(1), vec!["id".to_string()], vec![]),
        MappedField::new(Some(2), vec!["m".to_string()], vec![
            MappedField::new(Some(3), vec!["key".to_string()], vec![]),
            MappedField::new(Some(4), vec!["value".to_string()], vec![MappedField::new(
                Some(5),
                vec!["x".to_string()],
                vec![],
            )]),
        ]),
    ]));

    // The value struct's child is named `x`, as in the mapping, and then `old_x`. No field ids.
    for physical in ["x", "old_x"] {
        let value = StructArray::from(vec![(
            Arc::new(Field::new(physical, DataType::Int32, true)),
            Arc::new(Int32Array::from(vec![10, 20, 30])) as ArrayRef,
        )]);
        let entry_fields = Fields::from(vec![
            Field::new("key", DataType::Utf8, false),
            Field::new("value", value.data_type().clone(), true),
        ]);
        let entries = StructArray::new(
            entry_fields.clone(),
            vec![
                Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef,
                Arc::new(value) as ArrayRef,
            ],
            None,
        );
        let map = MapArray::new(
            Arc::new(Field::new("key_value", DataType::Struct(entry_fields), false)),
            OffsetBuffer::new(ScalarBuffer::from(vec![0i32, 1, 2, 3])),
            entries,
            None,
            false,
        );
        let to_write = RecordBatch::try_from_iter_with_nullable(vec![
            ("id", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef, true),
            ("m", Arc::new(map) as ArrayRef, true),
        ])
        .unwrap();
        let tmp_dir = TempDir::new().unwrap();
        let file_path = format!("{}/1.parquet", tmp_dir.path().to_str().unwrap());
        let mut writer =
            ArrowWriter::try_new(File::create(&file_path).unwrap(), to_write.schema(), None)
                .unwrap();
        writer.write(&to_write).unwrap();
        writer.close().unwrap();

        let tasks = Box::pin(futures::stream::iter([Ok(FileScanTask::builder()
            .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
            .with_start(0)
            .with_length(0)
            .with_data_file_path(file_path.clone())
            .with_data_file_format(DataFileFormat::Parquet)
            .with_schema(schema.clone())
            .with_project_field_ids(vec![2])
            .with_case_sensitive(false)
            .with_name_mapping(Some(name_mapping.clone()))
            .build()
            .unwrap())])) as FileScanTaskStream;
        let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
        let result = reader
            .read(tasks)
            .unwrap()
            .stream()
            .try_collect::<Vec<RecordBatch>>()
            .await;
        assert!(result.is_ok(), "physical={physical}: {:?}", result.err());
    }
}