fix(spec): match manifest partition fields by field ID

bindings/python/tests/test_manifest.py

@@ -146,7 +146,7 @@ def test_read_manifest_entry(generated_manifest_entry_file: str) -> None:
         == "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=null/00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet"
     )
     assert data_file.file_format == FileFormat.PARQUET
-    assert repr(data_file.partition) == "[1, None]"
+    assert repr(data_file.partition) == "[1, 1925]"
     assert data_file.record_count == 19513
     assert data_file.file_size_in_bytes == 388872
     assert data_file.column_sizes == {

crates/iceberg/src/avro/schema.rs

@@ -34,7 +34,7 @@ use crate::spec::{
 use crate::{Error, ErrorKind, Result, ensure_data_valid};
 
 const ELEMENT_ID: &str = "element-id";
-const FIELD_ID_PROP: &str = "field-id";
+pub(crate) const FIELD_ID_PROP: &str = "field-id";
 const KEY_ID: &str = "key-id";
 const VALUE_ID: &str = "value-id";
 const MAP_LOGICAL_TYPE: &str = "map";

crates/iceberg/src/spec/manifest/mod.rs

@@ -26,18 +26,19 @@ pub use metadata::*;
 mod reader;
 pub use reader::*;
 mod writer;
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
-use apache_avro::Reader as AvroReader;
 use apache_avro::error::Details;
+use apache_avro::{Reader as AvroReader, Schema as AvroSchema};
 pub use writer::*;
 
 use super::{
     Datum, FormatVersion, ManifestContentType, PartitionSpec, PrimitiveType, Schema, Struct, Type,
     UNASSIGNED_SEQUENCE_NUMBER,
 };
-use crate::avro::{Resolved, define_named_types_once};
+use crate::avro::{FIELD_ID_PROP, Resolved, define_named_types_once};
 use crate::error::{Error, Result, invalid_data};
 
 /// Whether a manifest with repeated Avro named type definitions was logged at
@@ -109,7 +110,8 @@ impl Manifest {
         // Wrap the partition type once and share it across all entries: the
         // per-entry conversion needs a `&Type`, and building it here keeps the
         // lazily-populated field-name lookup from being rebuilt for every entry.
-        let partition_struct_type = Type::Struct(partition_type.clone());
+        let partition_struct_type =
+            Type::Struct(with_writer_names(reader.writer_schema(), &partition_type));
 
         let entries = match metadata.format_version {
             FormatVersion::V1 => reader
@@ -169,6 +171,61 @@ impl Manifest {
     }
 }
 
+/// `partition_type` with each field renamed to the writer's Avro name for it,
+/// matched by `field-id`. If the writer's partition fields have no field IDs,
+/// they are matched by name. Java writes a name that isn't a valid Avro name,
+/// such as `a.b`, as `a_x2Eb`.
+fn with_writer_names(
+    writer_schema: &AvroSchema,
+    partition_type: &super::StructType,
+) -> super::StructType {
+    fn field<'a>(schema: &'a AvroSchema, name: &str) -> Option<&'a AvroSchema> {
+        match schema {
+            AvroSchema::Record(record) => {
+                let i = *record.lookup.get(name)?;
+                record.fields.get(i).map(|f| &f.schema)
+            }
+            _ => None,
+        }
+    }
+    let partition_fields = match field(writer_schema, "data_file")
+        .and_then(|data_file| field(data_file, "partition"))
+    {
+        Some(AvroSchema::Record(partition)) => partition.fields.as_slice(),
+        _ => &[],
+    };
+    let names: HashMap<i64, &str> = partition_fields
+        .iter()
+        .filter_map(|f| {
+            Some((
+                f.custom_attributes.get(FIELD_ID_PROP)?.as_i64()?,
+                f.name.as_str(),
+            ))
+        })
+        .collect();
+    if names.is_empty() {
+        return partition_type.clone();
+    }
+    super::StructType::new(
+        partition_type
+            .fields()
+            .iter()
+            .map(|f| {
+                let name = match names.get(&i64::from(f.id)) {
+                    Some(&name) if name == f.name => return f.clone(),
+                    Some(&name) => name.to_string(),
+                    // apache-avro rejects `#` in names, so this matches nothing and reads as null.
+                    None => format!("{}#{}", f.name, f.id),
+                };
+                Arc::new(super::NestedField {
+                    name,
+                    ..f.as_ref().clone()
+                })
+            })
+            .collect(),
+    )
+}
+
 /// Serialize a DataFile to a JSON string.
 pub fn serialize_data_file_to_json(
     data_file: DataFile,
@@ -1876,6 +1933,83 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_parse_manifest_written_by_java_with_sanitized_partition_name() {
+        // Java writes the partition field `user.id_bucket` as `user_x2Eid_bucket`.
+        for (file, format_version) in [
+            ("java-v1-nested-bucket.avro", FormatVersion::V1),
+            ("java-v2-nested-bucket.avro", FormatVersion::V2),
+        ] {
+            let bs = fs::read(format!(
+                "{}/testdata/manifests/{file}",
+                env!("CARGO_MANIFEST_DIR")
+            ))
+            .unwrap();
+
+            let manifest = Manifest::parse_avro(&bs).unwrap();
+
+            assert_eq!(manifest.metadata().format_version, format_version);
+            assert_eq!(
+                manifest.metadata().partition_spec.fields()[0].name,
+                "user.id_bucket"
+            );
+            assert_eq!(
+                *manifest.entries()[0].data_file().partition(),
+                Struct::from_iter([Some(Literal::int(3))])
+            );
+        }
+    }
+
+    #[test]
+    fn test_parse_manifest_matches_partition_fields_by_field_id() {
+        let metadata = writer_schema_test_metadata();
+        let cases = [
+            // In spec order, but `name` and `score` have each other's field IDs.
+            (
+                serde_json::json!([
+                    {"name": "id", "type": ["null", "long"], "default": null, "field-id": 1000},
+                    {"name": "name", "type": ["null", "double"], "default": null, "field-id": 1002},
+                    {"name": "score", "type": ["null", "string"], "default": null, "field-id": 1001},
+                ]),
+                serde_json::json!({"id": 5, "name": 2.5, "score": "a"}),
+                Some(Literal::double(2.5)),
+            ),
+            // With field IDs, a field whose name matches but whose ID doesn't reads as null.
+            (
+                serde_json::json!([
+                    {"name": "id", "type": ["null", "long"], "default": null, "field-id": 1000},
+                    {"name": "name", "type": ["null", "string"], "default": null, "field-id": 1001},
+                    {"name": "score", "type": ["null", "double"], "default": null, "field-id": 1005},
+                ]),
+                serde_json::json!({"id": 5, "name": "a", "score": 2.5}),
+                None,
+            ),
+            // Without field IDs, fields match by name.
+            (
+                serde_json::json!([
+                    {"name": "score", "type": ["null", "double"], "default": null},
+                    {"name": "id", "type": ["null", "long"], "default": null},
+                    {"name": "name", "type": ["null", "string"], "default": null},
+                ]),
+                serde_json::json!({"score": 2.5, "id": 5, "name": "a"}),
+                Some(Literal::double(2.5)),
+            ),
+        ];
+        for (partition_fields, partition, score) in cases {
+            let bs =
+                write_with_writer_schema(&metadata, &v2_writer_schema(partition_fields), vec![
+                    v2_entry(partition),
+                ]);
+
+            let partition =
+                Struct::from_iter([Some(Literal::long(5)), Some(Literal::string("a")), score]);
+            assert_eq!(
+                Manifest::parse_avro(&bs).unwrap(),
+                Manifest::new(metadata.clone(), vec![expected_entry(partition)])
+            );
+        }
+    }
+
     #[test]
     fn test_parse_manifest_promotes_writer_types() {
         // Avro promotes int to long and float to double.

crates/iceberg/src/spec/values/serde.rs

@@ -53,9 +53,8 @@ pub(crate) mod _serde {
         /// record lacks. Record fields that `struct_type` lacks are dropped. Values
         /// other than records are returned unchanged.
         ///
-        /// Fields aren't matched by field ID. A writer that stores a field under
-        /// another name, such as Java replacing characters that Avro names don't
-        /// allow, reads as a missing field.
+        /// Fields aren't matched by field ID; callers that need that rename
+        /// `struct_type`'s fields to the writer's names first.
         pub fn project_by_name(self, struct_type: &StructType) -> Result<Self, Error> {
             let (mut required, optional) = match self.0 {
                 RawLiteralEnum::Record(Record { required, optional }) => (required, optional),

crates/iceberg/testdata/manifests/README.md

@@ -97,3 +97,46 @@ def main(path: str) -> None:
 if __name__ == "__main__":
     main(sys.argv[1])
 ```
+
+`java-v1-nested-bucket.avro` and `java-v2-nested-bucket.avro` are V1 and V2 data
+manifests written by iceberg-java 1.12.0. Their table schema has optional columns
+`event` (string, id 1) and `user` (struct, id 2) with an optional `id` (long,
+id 3), and their partition spec is `bucket(16, user.id)`. Java names the
+partition field `user.id_bucket`, which isn't a valid Avro name, so the Avro
+schema stores it as `user_x2Eid_bucket` with `field-id` 1000 and the original in
+`iceberg-field-name`. Each holds one entry for `s3://bucket/table/data/a.parquet`
+with partition value 3.
+
+To regenerate them, run this program against iceberg-java 1.12.0
+(`iceberg-core` and its dependencies), passing the output directory:
+
+```java
+import java.io.File;
+import org.apache.iceberg.*;
+import org.apache.iceberg.types.Types;
+
+public class WriteManifests {
+  public static void main(String[] args) throws Exception {
+    Schema schema = new Schema(
+        Types.NestedField.optional(1, "event", Types.StringType.get()),
+        Types.NestedField.optional(2, "user", Types.StructType.of(
+            Types.NestedField.optional(3, "id", Types.LongType.get()))));
+    PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("user.id", 16).build();
+    for (int version : new int[] {1, 2}) {
+      File out = new File(args[0], "java-v" + version + "-nested-bucket.avro");
+      PartitionData partition = new PartitionData(spec.partitionType());
+      partition.set(0, 3);
+      try (ManifestWriter<DataFile> writer =
+          ManifestFiles.write(version, spec, Files.localOutput(out), 7L)) {
+        writer.add(DataFiles.builder(spec)
+            .withPath("s3://bucket/table/data/a.parquet")
+            .withFormat(FileFormat.PARQUET)
+            .withPartition(partition)
+            .withRecordCount(10)
+            .withFileSizeInBytes(100)
+            .build());
+      }
+    }
+  }
+}
+```