Disambiguate partition values.

Author: mbutrovich

crates/iceberg/src/arrow/record_batch_transformer.rs

@@ -274,8 +274,8 @@ impl RecordBatchTransformer {
         let field_id_to_source_schema_map =
             Self::build_field_id_to_arrow_schema_map(source_schema)?;
 
-        // Check if partition columns (initial_default) have field IDs that also exist in Parquet.
-        // This means the same field ID maps to different columns, requiring name-based mapping.
+        // Check if partition columns (initial_default) conflict with Parquet field IDs.
+        // If so, use name-based mapping for data columns (common in add_files imports).
         let has_field_id_conflict = projected_iceberg_field_ids.iter().any(|field_id| {
             if let Some(iceberg_field) = snapshot_schema.field_by_id(*field_id) {
                 // If this field has initial_default (partition column) and its field ID exists in Parquet
@@ -300,13 +300,25 @@ impl RecordBatchTransformer {
             )?;
             let target_type = target_field.data_type();
 
-            // Partition columns (initial_default) use constant values, not Parquet data.
-            // See Iceberg Java: BaseParquetReaders.java PartitionUtil.constantsMap().
             let iceberg_field = snapshot_schema.field_by_id(*field_id).ok_or(
                 Error::new(ErrorKind::Unexpected, "Field not found in snapshot schema")
             )?;
 
-            let column_source = if iceberg_field.initial_default.is_some() {
+            // Check if this field is present in the Parquet file
+            let is_in_parquet = if has_field_id_conflict {
+                name_to_source_schema_map.contains_key(iceberg_field.name.as_str())
+            } else {
+                field_id_to_source_schema_map.contains_key(field_id)
+            };
+
+            // Fields with initial_default that are NOT in Parquet are partition columns
+            // (common in add_files scenario where partition columns are in directory paths).
+            // Per spec (Scan Planning: "Return value from partition metadata for Identity transforms").
+            // See Iceberg Java: BaseParquetReaders.java uses PartitionUtil.constantsMap() for this.
+            //
+            // Fields with initial_default that ARE in Parquet should be read normally
+            // (e.g., source columns for bucket partitioning like 'id' in PARTITIONED BY bucket(4, id)).
+            let column_source = if iceberg_field.initial_default.is_some() && !is_in_parquet {
                 // This is a partition column - use the constant value, don't read from file
                 let default_value = if let Some(iceberg_default_value) = &iceberg_field.initial_default {
                     let Literal::Primitive(primitive_literal) = iceberg_default_value else {