diff --git a/bindings/python/tests/test_manifest.py b/bindings/python/tests/test_manifest.py index 147b81fccc..b095befd45 100644 --- a/bindings/python/tests/test_manifest.py +++ b/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 == { diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index 9af68022a2..9c87ad94e8 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/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"; diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index f4269638a2..d4fb2477f3 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/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 = 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. diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs index 028c2d796e..b8aad07b26 100644 --- a/crates/iceberg/src/spec/values/serde.rs +++ b/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 { let (mut required, optional) = match self.0 { RawLiteralEnum::Record(Record { required, optional }) => (required, optional), diff --git a/crates/iceberg/testdata/manifests/README.md b/crates/iceberg/testdata/manifests/README.md index 6b68b71ed0..cabaa5512a 100644 --- a/crates/iceberg/testdata/manifests/README.md +++ b/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 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()); + } + } + } +} +``` diff --git a/crates/iceberg/testdata/manifests/java-v1-nested-bucket.avro b/crates/iceberg/testdata/manifests/java-v1-nested-bucket.avro new file mode 100644 index 0000000000..bcdef51464 Binary files /dev/null and b/crates/iceberg/testdata/manifests/java-v1-nested-bucket.avro differ diff --git a/crates/iceberg/testdata/manifests/java-v2-nested-bucket.avro b/crates/iceberg/testdata/manifests/java-v2-nested-bucket.avro new file mode 100644 index 0000000000..32a5d0f644 Binary files /dev/null and b/crates/iceberg/testdata/manifests/java-v2-nested-bucket.avro differ