feat(arrow): Allow ArrowArrayAccessor to use field.name to match fields (apache#1561)

## Which issue does this PR close?


- Closes apache#1560 

## What changes are included in this PR?


- Added `FieldMatchMode` to `ArrowArrayAccessor`
- Added `match_mode` to `ParquetWriterBuilder`, `NanValueCountVisitor`
so user have control over this behavior on the writer level
- Hardcoded `IcebergWriteExec` to use `FieldMatchMode::Name`

## Are these changes tested?
Yes, added uts


---------

Co-authored-by: Florian Valeye <[email protected]>
Co-authored-by: Renjie Liu <[email protected]>

Author: 3 people

crates/iceberg/src/arrow/nan_val_cnt_visitor.rs

@@ -25,7 +25,7 @@ use arrow_array::{ArrayRef, Float32Array, Float64Array, RecordBatch, StructArray
 use arrow_schema::DataType;
 
 use crate::Result;
-use crate::arrow::ArrowArrayAccessor;
+use crate::arrow::{ArrowArrayAccessor, FieldMatchMode};
 use crate::spec::{
     ListType, MapType, NestedFieldRef, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor,
     StructType, visit_struct_with_partner,
@@ -71,6 +71,7 @@ macro_rules! count_float_nans {
 pub struct NanValueCountVisitor {
     /// Stores field ID to NaN value count mapping
     pub nan_value_counts: HashMap<i32, u64>,
+    match_mode: FieldMatchMode,
 }
 
 impl SchemaWithPartnerVisitor<ArrayRef> for NanValueCountVisitor {
@@ -149,14 +150,20 @@ impl SchemaWithPartnerVisitor<ArrayRef> for NanValueCountVisitor {
 impl NanValueCountVisitor {
     /// Creates new instance of NanValueCountVisitor
     pub fn new() -> Self {
+        Self::new_with_match_mode(FieldMatchMode::Id)
+    }
+
+    /// Creates new instance of NanValueCountVisitor with explicit match mode
+    pub fn new_with_match_mode(match_mode: FieldMatchMode) -> Self {
         Self {
             nan_value_counts: HashMap::new(),
+            match_mode,
         }
     }
 
     /// Compute nan value counts in given schema and record batch
     pub fn compute(&mut self, schema: SchemaRef, batch: RecordBatch) -> Result<()> {
-        let arrow_arr_partner_accessor = ArrowArrayAccessor {};
+        let arrow_arr_partner_accessor = ArrowArrayAccessor::new_with_match_mode(self.match_mode);
 
         let struct_arr = Arc::new(StructArray::from(batch)) as ArrayRef;
         visit_struct_with_partner(

crates/iceberg/src/arrow/value.rs

@@ -21,7 +21,7 @@ use arrow_array::{
     LargeListArray, LargeStringArray, ListArray, MapArray, StringArray, StructArray,
     Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray,
 };
-use arrow_schema::DataType;
+use arrow_schema::{DataType, FieldRef};
 use uuid::Uuid;
 
 use super::get_field_id;
@@ -425,11 +425,63 @@ impl SchemaWithPartnerVisitor<ArrayRef> for ArrowArrayToIcebergStructConverter {
     }
 }
 
+/// Defines how Arrow fields are matched with Iceberg fields when converting data.
+///
+/// This enum provides two strategies for matching fields:
+/// - `Id`: Match fields by their ID, which is stored in Arrow field metadata.
+/// - `Name`: Match fields by their name, ignoring the field ID.
+///
+/// The ID matching mode is the default and preferred approach as it's more robust
+/// against schema evolution where field names might change but IDs remain stable.
+/// The name matching mode can be useful in scenarios where field IDs are not available
+/// or when working with systems that don't preserve field IDs.
+#[derive(Clone, Copy, Debug)]
+pub enum FieldMatchMode {
+    /// Match fields by their ID stored in Arrow field metadata
+    Id,
+    /// Match fields by their name, ignoring field IDs
+    Name,
+}
+
+impl FieldMatchMode {
+    /// Determines if an Arrow field matches an Iceberg field based on the matching mode.
+    pub fn match_field(&self, arrow_field: &FieldRef, iceberg_field: &NestedField) -> bool {
+        match self {
+            FieldMatchMode::Id => get_field_id(arrow_field)
+                .map(|id| id == iceberg_field.id)
+                .unwrap_or(false),
+            FieldMatchMode::Name => arrow_field.name() == &iceberg_field.name,
+        }
+    }
+}
+
 /// Partner type representing accessing and walking arrow arrays alongside iceberg schema
-pub struct ArrowArrayAccessor;
+pub struct ArrowArrayAccessor {
+    match_mode: FieldMatchMode,
+}
+
+impl ArrowArrayAccessor {
+    /// Creates a new instance of ArrowArrayAccessor with the default ID matching mode
+    pub fn new() -> Self {
+        Self {
+            match_mode: FieldMatchMode::Id,
+        }
+    }
+
+    /// Creates a new instance of ArrowArrayAccessor with the specified matching mode
+    pub fn new_with_match_mode(match_mode: FieldMatchMode) -> Self {
+        Self { match_mode }
+    }
+}
+
+impl Default for ArrowArrayAccessor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 
 impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
-    fn struct_parner<'a>(&self, schema_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
+    fn struct_partner<'a>(&self, schema_partner: &'a ArrayRef) -> Result<&'a ArrayRef> {
         if !matches!(schema_partner.data_type(), DataType::Struct(_)) {
             return Err(Error::new(
                 ErrorKind::DataInvalid,
@@ -451,18 +503,17 @@ impl PartnerAccessor<ArrayRef> for ArrowArrayAccessor {
             .ok_or_else(|| {
                 Error::new(
                     ErrorKind::DataInvalid,
-                    "The struct partner is not a struct array",
+                    format!(
+                        "The struct partner is not a struct array, partner: {:?}",
+                        struct_partner
+                    ),
                 )
             })?;
 
         let field_pos = struct_array
             .fields()
             .iter()
-            .position(|arrow_field| {
-                get_field_id(arrow_field)
-                    .map(|id| id == field.id)
-                    .unwrap_or(false)
-            })
+            .position(|arrow_field| self.match_mode.match_field(arrow_field, field))
             .ok_or_else(|| {
                 Error::new(
                     ErrorKind::DataInvalid,
@@ -549,7 +600,7 @@ pub fn arrow_struct_to_literal(
         ty,
         struct_array,
         &mut ArrowArrayToIcebergStructConverter,
-        &ArrowArrayAccessor,
+        &ArrowArrayAccessor::new(),
     )
 }
 
@@ -899,6 +950,183 @@ mod test {
         assert_eq!(result, vec![None; 0]);
     }
 
+    #[test]
+    fn test_find_field_by_id() {
+        // Create Arrow arrays for the nested structure
+        let field_a_array = Int32Array::from(vec![Some(42), Some(43), None]);
+        let field_b_array = StringArray::from(vec![Some("value1"), Some("value2"), None]);
+
+        // Create the nested struct array with field IDs in metadata
+        let nested_struct_array =
+            Arc::new(StructArray::from(vec![
+                (
+                    Arc::new(Field::new("field_a", DataType::Int32, true).with_metadata(
+                        HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
+                    )),
+                    Arc::new(field_a_array) as ArrayRef,
+                ),
+                (
+                    Arc::new(Field::new("field_b", DataType::Utf8, true).with_metadata(
+                        HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "2".to_string())]),
+                    )),
+                    Arc::new(field_b_array) as ArrayRef,
+                ),
+            ])) as ArrayRef;
+
+        let field_c_array = Int32Array::from(vec![Some(100), Some(200), None]);
+
+        // Create the top-level struct array with field IDs in metadata
+        let struct_array = Arc::new(StructArray::from(vec![
+            (
+                Arc::new(
+                    Field::new(
+                        "nested_struct",
+                        DataType::Struct(Fields::from(vec![
+                            Field::new("field_a", DataType::Int32, true).with_metadata(
+                                HashMap::from([(
+                                    PARQUET_FIELD_ID_META_KEY.to_string(),
+                                    "1".to_string(),
+                                )]),
+                            ),
+                            Field::new("field_b", DataType::Utf8, true).with_metadata(
+                                HashMap::from([(
+                                    PARQUET_FIELD_ID_META_KEY.to_string(),
+                                    "2".to_string(),
+                                )]),
+                            ),
+                        ])),
+                        true,
+                    )
+                    .with_metadata(HashMap::from([(
+                        PARQUET_FIELD_ID_META_KEY.to_string(),
+                        "3".to_string(),
+                    )])),
+                ),
+                nested_struct_array,
+            ),
+            (
+                Arc::new(Field::new("field_c", DataType::Int32, true).with_metadata(
+                    HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "4".to_string())]),
+                )),
+                Arc::new(field_c_array) as ArrayRef,
+            ),
+        ])) as ArrayRef;
+
+        // Create an ArrowArrayAccessor with ID matching mode
+        let accessor = ArrowArrayAccessor::new_with_match_mode(FieldMatchMode::Id);
+
+        // Test finding fields by ID
+        let nested_field = NestedField::optional(
+            3,
+            "nested_struct",
+            Type::Struct(StructType::new(vec![
+                Arc::new(NestedField::optional(
+                    1,
+                    "field_a",
+                    Type::Primitive(PrimitiveType::Int),
+                )),
+                Arc::new(NestedField::optional(
+                    2,
+                    "field_b",
+                    Type::Primitive(PrimitiveType::String),
+                )),
+            ])),
+        );
+        let nested_partner = accessor
+            .field_partner(&struct_array, &nested_field)
+            .unwrap();
+
+        // Verify we can access the nested field
+        let field_a = NestedField::optional(1, "field_a", Type::Primitive(PrimitiveType::Int));
+        let field_a_partner = accessor.field_partner(nested_partner, &field_a).unwrap();
+
+        // Verify the field has the expected value
+        let int_array = field_a_partner
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .unwrap();
+        assert_eq!(int_array.value(0), 42);
+        assert_eq!(int_array.value(1), 43);
+        assert!(int_array.is_null(2));
+    }
+
+    #[test]
+    fn test_find_field_by_name() {
+        // Create Arrow arrays for the nested structure
+        let field_a_array = Int32Array::from(vec![Some(42), Some(43), None]);
+        let field_b_array = StringArray::from(vec![Some("value1"), Some("value2"), None]);
+
+        // Create the nested struct array WITHOUT field IDs in metadata
+        let nested_struct_array = Arc::new(StructArray::from(vec![
+            (
+                Arc::new(Field::new("field_a", DataType::Int32, true)),
+                Arc::new(field_a_array) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("field_b", DataType::Utf8, true)),
+                Arc::new(field_b_array) as ArrayRef,
+            ),
+        ])) as ArrayRef;
+
+        let field_c_array = Int32Array::from(vec![Some(100), Some(200), None]);
+
+        // Create the top-level struct array WITHOUT field IDs in metadata
+        let struct_array = Arc::new(StructArray::from(vec![
+            (
+                Arc::new(Field::new(
+                    "nested_struct",
+                    DataType::Struct(Fields::from(vec![
+                        Field::new("field_a", DataType::Int32, true),
+                        Field::new("field_b", DataType::Utf8, true),
+                    ])),
+                    true,
+                )),
+                nested_struct_array,
+            ),
+            (
+                Arc::new(Field::new("field_c", DataType::Int32, true)),
+                Arc::new(field_c_array) as ArrayRef,
+            ),
+        ])) as ArrayRef;
+
+        // Create an ArrowArrayAccessor with Name matching mode
+        let accessor = ArrowArrayAccessor::new_with_match_mode(FieldMatchMode::Name);
+
+        // Test finding fields by name
+        let nested_field = NestedField::optional(
+            3,
+            "nested_struct",
+            Type::Struct(StructType::new(vec![
+                Arc::new(NestedField::optional(
+                    1,
+                    "field_a",
+                    Type::Primitive(PrimitiveType::Int),
+                )),
+                Arc::new(NestedField::optional(
+                    2,
+                    "field_b",
+                    Type::Primitive(PrimitiveType::String),
+                )),
+            ])),
+        );
+        let nested_partner = accessor
+            .field_partner(&struct_array, &nested_field)
+            .unwrap();
+
+        // Verify we can access the nested field by name
+        let field_a = NestedField::optional(1, "field_a", Type::Primitive(PrimitiveType::Int));
+        let field_a_partner = accessor.field_partner(nested_partner, &field_a).unwrap();
+
+        // Verify the field has the expected value
+        let int_array = field_a_partner
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .unwrap();
+        assert_eq!(int_array.value(0), 42);
+        assert_eq!(int_array.value(1), 43);
+        assert!(int_array.is_null(2));
+    }
+
     #[test]
     fn test_complex_nested() {
         // complex nested type for test

crates/iceberg/src/spec/schema/visitor.rs

@@ -190,7 +190,7 @@ pub trait SchemaWithPartnerVisitor<P> {
 /// Accessor used to get child partner from parent partner.
 pub trait PartnerAccessor<P> {
     /// Get the struct partner from schema partner.
-    fn struct_parner<'a>(&self, schema_partner: &'a P) -> Result<&'a P>;
+    fn struct_partner<'a>(&self, schema_partner: &'a P) -> Result<&'a P>;
     /// Get the field partner from struct partner.
     fn field_partner<'a>(&self, struct_partner: &'a P, field: &NestedField) -> Result<&'a P>;
     /// Get the list element partner from list partner.
@@ -274,7 +274,7 @@ pub fn visit_schema_with_partner<P, V: SchemaWithPartnerVisitor<P>, A: PartnerAc
 ) -> Result<V::T> {
     let result = visit_struct_with_partner(
         &schema.r#struct,
-        accessor.struct_parner(partner)?,
+        accessor.struct_partner(partner)?,
         visitor,
         accessor,
     )?;