From a80adde276af19f930361b1470239efff665fcca Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 5 Oct 2026 11:40:16 -0700 Subject: [PATCH 1/5] perf(manifest): upgrade to apache-avro 0.22 and read manifest entries without Value --- Cargo.lock | 115 ++++- Cargo.toml | 2 +- crates/iceberg/src/avro/deserializer.rs | 278 +++++++++++ crates/iceberg/src/avro/mod.rs | 4 + crates/iceberg/src/avro/named_types.rs | 255 ++++++++++ crates/iceberg/src/avro/schema.rs | 87 ++-- crates/iceberg/src/encryption/key_metadata.rs | 73 ++- crates/iceberg/src/spec/manifest/_serde.rs | 12 + crates/iceberg/src/spec/manifest/data_file.rs | 8 +- crates/iceberg/src/spec/manifest/mod.rs | 471 ++++++++++++++++-- crates/iceberg/src/spec/manifest/reader.rs | 3 +- crates/iceberg/src/spec/manifest/writer.rs | 80 ++- crates/iceberg/src/spec/manifest_list/mod.rs | 6 +- .../iceberg/src/spec/manifest_list/writer.rs | 3 +- crates/iceberg/src/spec/values/serde.rs | 47 +- crates/iceberg/src/spec/values/tests.rs | 70 ++- crates/iceberg/testdata/manifests/README.md | 37 ++ .../repeated-decimal-type-definitions.avro | Bin 0 -> 3865 bytes 18 files changed, 1437 insertions(+), 114 deletions(-) create mode 100644 crates/iceberg/src/avro/deserializer.rs create mode 100644 crates/iceberg/src/avro/named_types.rs create mode 100644 crates/iceberg/testdata/manifests/README.md create mode 100644 crates/iceberg/testdata/manifests/repeated-decimal-type-definitions.avro diff --git a/Cargo.lock b/Cargo.lock index 7b2115a713..85c1e9b250 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,6 +67,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "aliasable" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd" + [[package]] name = "alloc-no-stdlib" version = "2.0.4" @@ -155,26 +161,26 @@ checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "apache-avro" -version = "0.21.0" +version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf" +checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c" dependencies = [ "bigdecimal", "bon", "crc32fast", - "digest 0.10.7", + "digest 0.11.3", "log", - "miniz_oxide", + "miniz_oxide 0.9.1", "num-bigint 0.4.8", + "ouroboros", "quad-rand", - "rand 0.9.5", + "rand 0.10.2", "regex-lite", "serde", "serde_bytes", "serde_json", "snap", - "strum", - "strum_macros", + "strum 0.28.0", "thiserror 2.0.18", "uuid", "zstd", @@ -1368,7 +1374,7 @@ version = "4.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn", @@ -2105,7 +2111,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", - "miniz_oxide", + "miniz_oxide 0.8.9", "zlib-rs", ] @@ -2443,6 +2449,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0049b265b7f201ca9ab25475b22b47fe444060126a51abe00f77d986fc5cc52e" +[[package]] +name = "heck" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" + [[package]] name = "heck" version = "0.5.0" @@ -2662,7 +2674,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.4", "system-configuration", "tokio", "tower-service", @@ -2747,7 +2759,7 @@ dependencies = [ "serde_json", "serde_repr", "serde_with", - "strum", + "strum 0.27.2", "tempfile", "tokio", "tracing", @@ -2878,7 +2890,7 @@ dependencies = [ "itertools 0.13.0", "regex", "sqlx", - "strum", + "strum 0.27.2", "tempfile", "tokio", "tracing", @@ -3574,6 +3586,15 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", +] + [[package]] name = "mio" version = "1.2.1" @@ -4178,6 +4199,30 @@ dependencies = [ "memchr", ] +[[package]] +name = "ouroboros" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59" +dependencies = [ + "aliasable", + "ouroboros_macro", + "static_assertions", +] + +[[package]] +name = "ouroboros_macro" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn", +] + [[package]] name = "outref" version = "0.5.2" @@ -4525,6 +4570,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc-macro2-diagnostics" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "version_check", + "yansi", +] + [[package]] name = "ptr_meta" version = "0.3.1" @@ -4605,7 +4663,7 @@ version = "0.29.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e343bcec300ff262f5806a33a4e51b6d097a8a46435f512fcb83e95592581625" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn", @@ -4640,7 +4698,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.5.10", + "socket2 0.6.4", "thiserror 2.0.18", "tokio", "tracing", @@ -4679,7 +4737,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.4", "tracing", "windows-sys 0.61.2", ] @@ -5913,7 +5971,7 @@ checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" dependencies = [ "dotenvy", "either", - "heck", + "heck 0.5.0", "hex", "once_cell", "proc-macro2", @@ -6075,7 +6133,16 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" dependencies = [ - "strum_macros", + "strum_macros 0.27.2", +] + +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", ] [[package]] @@ -6084,7 +6151,19 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" dependencies = [ - "heck", + "heck 0.5.0", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck 0.5.0", "proc-macro2", "quote", "syn", diff --git a/Cargo.toml b/Cargo.toml index 661c56b094..24f8819949 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,7 +46,7 @@ unused_qualifications = "deny" [workspace.dependencies] aes-gcm = "0.10" anyhow = "1.0.72" -apache-avro = { version = "0.21", features = ["snappy", "zstandard"] } +apache-avro = { version = "0.22", features = ["snappy", "zstandard"] } array-init = "2" arrow = "59.2" arrow-arith = "59.2" diff --git a/crates/iceberg/src/avro/deserializer.rs b/crates/iceberg/src/avro/deserializer.rs new file mode 100644 index 0000000000..1401247eb8 --- /dev/null +++ b/crates/iceberg/src/avro/deserializer.rs @@ -0,0 +1,278 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Deserialization straight from an Avro writer schema, with the parts of Avro +//! schema resolution that apache-avro's schema-aware deserializer leaves out. +//! +//! `Reader::into_deser_iter` decodes into serde types without building +//! `apache_avro::types::Value`s, but it requires the reader schema to equal the +//! writer schema and applies no resolution rules. [`ResolvingDeserializer`] +//! wraps it and routes every value through `deserialize_any`, which follows the +//! writer schema. As a result: +//! +//! - Avro record names don't have to match serde type names. +//! - A writer value that isn't a union reads into an `Option`. +//! - serde's numeric visitors convert between numeric types, so an `int` reads +//! into an `i64` and a `float` into an `f64`. An integer that doesn't fit the +//! target, such as a `long` above `i32::MAX` read into an `i32`, returns an +//! error. Conversions into `f32` or `f64` use `as` and can lose precision. +//! +//! apache-avro plans a `SchemaAwareResolvingDeserializer` that resolves against +//! a reader schema (). Once a +//! release includes it, readers can pass their reader schema to +//! `Reader::builder` and drop this module, provided it doesn't reject writer +//! record names that differ from the reader's. The Avro spec requires record +//! names to match, and the writers this crate reads from don't all agree on +//! them. + +use std::fmt; + +use serde::de::value::{ + BorrowedBytesDeserializer, BorrowedStrDeserializer, BytesDeserializer, EnumAccessDeserializer, + MapAccessDeserializer, SeqAccessDeserializer, +}; +use serde::de::{ + DeserializeSeed, Deserializer, EnumAccess, Error, IntoDeserializer, MapAccess, SeqAccess, + Visitor, +}; +use serde::{Deserialize, forward_to_deserialize_any}; + +/// Deserializes the wrapped type through [`ResolvingDeserializer`], for use +/// with `Reader::into_deser_iter`. +pub(crate) struct Resolved(pub T); + +impl<'de, T: Deserialize<'de>> Deserialize<'de> for Resolved { + fn deserialize>(deserializer: D) -> Result { + T::deserialize(ResolvingDeserializer(deserializer)).map(Resolved) + } +} + +struct ResolvingDeserializer(D); + +impl<'de, D: Deserializer<'de>> Deserializer<'de> for ResolvingDeserializer { + type Error = D::Error; + + fn deserialize_any>(self, visitor: V) -> Result { + self.0.deserialize_any(ResolvingVisitor(visitor)) + } + + fn deserialize_option>(self, visitor: V) -> Result { + self.0.deserialize_any(OptionVisitor(visitor)) + } + + fn deserialize_ignored_any>(self, visitor: V) -> Result { + self.0.deserialize_ignored_any(visitor) + } + + fn is_human_readable(&self) -> bool { + self.0.is_human_readable() + } + + forward_to_deserialize_any! { + bool i8 i16 i32 i64 i128 u8 u16 u32 u64 u128 f32 f64 char str string bytes byte_buf + unit unit_struct newtype_struct seq tuple tuple_struct map struct enum identifier + } +} + +/// Wraps the sequences and maps a visitor receives, so that their elements are +/// deserialized through [`ResolvingDeserializer`] too. +struct ResolvingVisitor(V); + +/// Forwards `visit_*` methods for values that contain no nested values. +macro_rules! forward_visit { + ($($method:ident($ty:ty)),* $(,)?) => { + $( + fn $method(self, v: $ty) -> Result { + self.0.$method(v) + } + )* + }; +} + +impl<'de, V: Visitor<'de>> Visitor<'de> for ResolvingVisitor { + type Value = V::Value; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + self.0.expecting(formatter) + } + + forward_visit! { + visit_bool(bool), visit_i8(i8), visit_i16(i16), visit_i32(i32), visit_i64(i64), + visit_i128(i128), visit_u8(u8), visit_u16(u16), visit_u32(u32), visit_u64(u64), + visit_u128(u128), visit_f32(f32), visit_f64(f64), visit_char(char), visit_str(&str), + visit_borrowed_str(&'de str), visit_string(String), visit_bytes(&[u8]), + visit_borrowed_bytes(&'de [u8]), visit_byte_buf(Vec), + } + + fn visit_none(self) -> Result { + self.0.visit_none() + } + + fn visit_unit(self) -> Result { + self.0.visit_unit() + } + + fn visit_some>(self, deserializer: D) -> Result { + self.0.visit_some(ResolvingDeserializer(deserializer)) + } + + fn visit_newtype_struct>( + self, + deserializer: D, + ) -> Result { + self.0 + .visit_newtype_struct(ResolvingDeserializer(deserializer)) + } + + fn visit_seq>(self, seq: A) -> Result { + self.0.visit_seq(ResolvingAccess(seq)) + } + + fn visit_map>(self, map: A) -> Result { + self.0.visit_map(ResolvingAccess(map)) + } + + fn visit_enum>(self, data: A) -> Result { + self.0.visit_enum(data) + } +} + +/// Reads an `Option` from any writer value. A null becomes `None`, and anything +/// else becomes `Some`, whether or not the writer schema is a union. +struct OptionVisitor(V); + +/// Implements `visit_*` methods that hand a value to `visit_some` through the +/// deserializer that `deserializer` builds from it. +macro_rules! visit_some { + ($($method:ident($ty:ty) => $deserializer:expr),* $(,)?) => { + $( + fn $method(self, v: $ty) -> Result { + self.0.visit_some($deserializer(v)) + } + )* + }; +} + +impl<'de, V: Visitor<'de>> Visitor<'de> for OptionVisitor { + type Value = V::Value; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + self.0.expecting(formatter) + } + + visit_some! { + visit_bool(bool) => IntoDeserializer::into_deserializer, + visit_i8(i8) => IntoDeserializer::into_deserializer, + visit_i16(i16) => IntoDeserializer::into_deserializer, + visit_i32(i32) => IntoDeserializer::into_deserializer, + visit_i64(i64) => IntoDeserializer::into_deserializer, + visit_i128(i128) => IntoDeserializer::into_deserializer, + visit_u8(u8) => IntoDeserializer::into_deserializer, + visit_u16(u16) => IntoDeserializer::into_deserializer, + visit_u32(u32) => IntoDeserializer::into_deserializer, + visit_u64(u64) => IntoDeserializer::into_deserializer, + visit_u128(u128) => IntoDeserializer::into_deserializer, + visit_f32(f32) => IntoDeserializer::into_deserializer, + visit_f64(f64) => IntoDeserializer::into_deserializer, + visit_char(char) => IntoDeserializer::into_deserializer, + visit_str(&str) => IntoDeserializer::into_deserializer, + visit_borrowed_str(&'de str) => BorrowedStrDeserializer::new, + visit_string(String) => IntoDeserializer::into_deserializer, + visit_bytes(&[u8]) => BytesDeserializer::new, + visit_borrowed_bytes(&'de [u8]) => BorrowedBytesDeserializer::new, + } + + fn visit_byte_buf(self, v: Vec) -> Result { + self.0.visit_some(BytesDeserializer::new(&v)) + } + + fn visit_none(self) -> Result { + self.0.visit_none() + } + + fn visit_unit(self) -> Result { + self.0.visit_none() + } + + fn visit_some>(self, deserializer: D) -> Result { + self.0.visit_some(ResolvingDeserializer(deserializer)) + } + + fn visit_seq>(self, seq: A) -> Result { + self.0 + .visit_some(SeqAccessDeserializer::new(ResolvingAccess(seq))) + } + + fn visit_map>(self, map: A) -> Result { + self.0 + .visit_some(MapAccessDeserializer::new(ResolvingAccess(map))) + } + + fn visit_enum>(self, data: A) -> Result { + self.0.visit_some(EnumAccessDeserializer::new(data)) + } +} + +/// Deserializes the elements of a sequence, or the keys and values of a map, +/// through [`ResolvingDeserializer`]. Keys need it because apache-avro's record +/// field names only support `deserialize_identifier` and `deserialize_any`, +/// and a key read as a `String` calls `deserialize_string`. +struct ResolvingAccess(A); + +impl<'de, A: SeqAccess<'de>> SeqAccess<'de> for ResolvingAccess { + type Error = A::Error; + + fn next_element_seed>( + &mut self, + seed: T, + ) -> Result, A::Error> { + self.0.next_element_seed(ResolvingSeed(seed)) + } + + fn size_hint(&self) -> Option { + self.0.size_hint() + } +} + +impl<'de, A: MapAccess<'de>> MapAccess<'de> for ResolvingAccess { + type Error = A::Error; + + fn next_key_seed>( + &mut self, + seed: K, + ) -> Result, A::Error> { + self.0.next_key_seed(ResolvingSeed(seed)) + } + + fn next_value_seed>(&mut self, seed: T) -> Result { + self.0.next_value_seed(ResolvingSeed(seed)) + } + + fn size_hint(&self) -> Option { + self.0.size_hint() + } +} + +struct ResolvingSeed(S); + +impl<'de, S: DeserializeSeed<'de>> DeserializeSeed<'de> for ResolvingSeed { + type Value = S::Value; + + fn deserialize>(self, deserializer: D) -> Result { + self.0.deserialize(ResolvingDeserializer(deserializer)) + } +} diff --git a/crates/iceberg/src/avro/mod.rs b/crates/iceberg/src/avro/mod.rs index f2a9310e7c..5a065dc4a9 100644 --- a/crates/iceberg/src/avro/mod.rs +++ b/crates/iceberg/src/avro/mod.rs @@ -16,5 +16,9 @@ // under the License. //! Avro related codes. +mod deserializer; +mod named_types; mod schema; +pub(crate) use deserializer::Resolved; +pub(crate) use named_types::define_named_types_once; pub(crate) use schema::*; diff --git a/crates/iceberg/src/avro/named_types.rs b/crates/iceberg/src/avro/named_types.rs new file mode 100644 index 0000000000..9a4eb033fa --- /dev/null +++ b/crates/iceberg/src/avro/named_types.rs @@ -0,0 +1,255 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Reading Avro files whose schema defines a named type more than once. +//! +//! The Avro specification allows one definition per full name. iceberg-rust +//! wrote manifests that define a `fixed` or `decimal` type once for every field +//! that uses it, while `apache-avro` 0.21 accepted that. `apache-avro` 0.22 +//! rejects such a file's header. Rewriting each repeated definition as a +//! reference to the first one keeps those files readable, because a reference +//! decodes the same bytes as the definition it refers to. + +use std::collections::HashMap; +use std::io::Cursor; + +use apache_avro::Schema as AvroSchema; +use apache_avro::reader::datum::GenericDatumReader; +use apache_avro::schema::MapSchema; +use apache_avro::types::Value as AvroValue; +use apache_avro::writer::datum::GenericDatumWriter; +use serde_json::{Map, Value as JsonValue}; + +use crate::Result; +use crate::error::invalid_data; + +const MAGIC: &[u8] = b"Obj\x01"; +const SCHEMA_KEY: &str = "avro.schema"; + +/// Rewrites the header of an Avro object container file so that each named type +/// is defined once and referenced by name afterwards. Returns the rewritten +/// file and the full names of the repeated types, or `None` if a repeated +/// definition differs from the first one, which leaves the file ambiguous. +pub(crate) fn define_named_types_once(bs: &[u8]) -> Result, Vec)>> { + let body = bs + .strip_prefix(MAGIC) + .ok_or_else(|| invalid_data!("Not an Avro object container file"))?; + // The header's metadata is an Avro `map`, followed by the sync marker. + let metadata_schema = AvroSchema::Map(MapSchema { + types: Box::new(AvroSchema::Bytes), + attributes: Default::default(), + }); + let mut cursor = Cursor::new(body); + let AvroValue::Map(mut metadata) = GenericDatumReader::builder(&metadata_schema) + .build()? + .read_value(&mut cursor)? + else { + return Err(invalid_data!("Avro file metadata is not a map")); + }; + let rest = &body[cursor.position() as usize..]; + + let Some(AvroValue::Bytes(schema)) = metadata.get(SCHEMA_KEY) else { + return Err(invalid_data!("Avro file metadata has no {SCHEMA_KEY}")); + }; + let mut schema: JsonValue = serde_json::from_slice(schema)?; + let Some(repeated) = reference_repeated_definitions(&mut schema) else { + return Ok(None); + }; + metadata.insert( + SCHEMA_KEY.to_string(), + AvroValue::Bytes(serde_json::to_vec(&schema)?), + ); + let metadata = GenericDatumWriter::builder(&metadata_schema) + .build()? + .write_value_to_vec(AvroValue::Map(metadata))?; + + Ok(Some(([MAGIC, &metadata, rest].concat(), repeated))) +} + +/// Replaces each repeated definition of a named type in `schema` with its full +/// name. Returns the replaced names, or `None` if a repeated definition differs +/// from the first one. +fn reference_repeated_definitions(schema: &mut JsonValue) -> Option> { + let mut defined = HashMap::new(); + let mut repeated = Vec::new(); + reference_repeated(schema, None, &mut defined, &mut repeated).then_some(repeated) +} + +/// Walks `schema` in the order Avro resolves names, recording each named type's +/// definition without its name. Returns `false` on a repeated name whose +/// definition differs from the first one. +fn reference_repeated( + schema: &mut JsonValue, + namespace: Option<&str>, + defined: &mut HashMap>, + repeated: &mut Vec, +) -> bool { + let object = match schema { + JsonValue::Array(variants) => { + return variants + .iter_mut() + .all(|variant| reference_repeated(variant, namespace, defined, repeated)); + } + JsonValue::Object(object) => object, + _ => return true, + }; + let type_name = match object.get_mut("type") { + Some(JsonValue::String(type_name)) => type_name.as_str(), + Some(nested) => return reference_repeated(nested, namespace, defined, repeated), + None => return true, + }; + match type_name { + "record" | "error" | "enum" | "fixed" => {} + "array" => { + return object + .get_mut("items") + .is_none_or(|items| reference_repeated(items, namespace, defined, repeated)); + } + "map" => { + return object + .get_mut("values") + .is_none_or(|values| reference_repeated(values, namespace, defined, repeated)); + } + _ => return true, + } + + let Some(full_name) = full_name(object, namespace) else { + // Leave a malformed schema for apache-avro to report. + return true; + }; + let mut definition = object.clone(); + definition.remove("name"); + definition.remove("namespace"); + if let Some(first) = defined.get(&full_name) { + if *first != definition { + return false; + } + *schema = JsonValue::String(full_name.clone()); + repeated.push(full_name); + return true; + } + defined.insert(full_name.clone(), definition); + + let record_namespace = full_name.rsplit_once('.').map(|(namespace, _)| namespace); + match object.get_mut("fields") { + Some(JsonValue::Array(fields)) => fields.iter_mut().all(|field| { + field.get_mut("type").is_none_or(|field_type| { + reference_repeated(field_type, record_namespace, defined, repeated) + }) + }), + _ => true, + } +} + +/// The full name of a named type, following the Avro specification: a name +/// containing a dot is already full, and otherwise the type's own `namespace` +/// or the enclosing namespace qualifies it. +fn full_name(object: &Map, enclosing_namespace: Option<&str>) -> Option { + let name = object.get("name")?.as_str()?; + if name.contains('.') { + return Some(name.to_string()); + } + let namespace = match object.get("namespace") { + Some(namespace) => namespace.as_str(), + None => enclosing_namespace, + }; + Some(match namespace { + Some(namespace) if !namespace.is_empty() => format!("{namespace}.{name}"), + _ => name.to_string(), + }) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + fn decimal_10_2() -> JsonValue { + json!({"type": "fixed", "name": "decimal_10_2", "size": 5, + "logicalType": "decimal", "precision": 10, "scale": 2}) + } + + #[test] + fn test_reference_repeated_definitions() { + let mut schema = json!({"type": "record", "name": "r102", "fields": [ + {"name": "a", "type": ["null", decimal_10_2()]}, + {"name": "b", "type": ["null", decimal_10_2()]}, + {"name": "c", "type": {"type": "array", "items": decimal_10_2()}}, + ]}); + + let repeated = reference_repeated_definitions(&mut schema); + + assert_eq!( + repeated, + Some(vec!["decimal_10_2".to_string(), "decimal_10_2".to_string()]) + ); + assert_eq!( + schema, + json!({"type": "record", "name": "r102", "fields": [ + {"name": "a", "type": ["null", decimal_10_2()]}, + {"name": "b", "type": ["null", "decimal_10_2"]}, + {"name": "c", "type": {"type": "array", "items": "decimal_10_2"}}, + ]}) + ); + } + + #[test] + fn test_reference_repeated_definitions_leaves_unique_names() { + let original = json!({"type": "record", "name": "r", "fields": [ + {"name": "a", "type": decimal_10_2()}, + {"name": "b", "type": "decimal_10_2"}, + {"name": "c", "type": {"type": "fixed", "name": "fixed_4", "size": 4}}, + ]}); + let mut schema = original.clone(); + + assert_eq!(reference_repeated_definitions(&mut schema), Some(vec![])); + assert_eq!(schema, original); + } + + #[test] + fn test_reference_repeated_definitions_uses_full_names() { + // The same short name in two namespaces names two different types. + let mut schema = json!({"type": "record", "name": "r", "fields": [ + {"name": "a", "type": {"type": "fixed", "name": "f", "namespace": "x", "size": 4}}, + {"name": "b", "type": {"type": "fixed", "name": "f", "namespace": "y", "size": 8}}, + {"name": "c", "type": {"type": "fixed", "name": "x.f", "size": 4}}, + {"name": "d", "type": {"type": "record", "name": "n", "namespace": "x", "fields": [ + {"name": "e", "type": {"type": "fixed", "name": "f", "size": 4}}, + ]}}, + ]}); + + let repeated = reference_repeated_definitions(&mut schema); + + assert_eq!(repeated, Some(vec!["x.f".to_string(), "x.f".to_string()])); + assert_eq!(schema["fields"][2]["type"], json!("x.f")); + assert_eq!( + schema["fields"][3]["type"]["fields"][0]["type"], + json!("x.f") + ); + } + + #[test] + fn test_reference_repeated_definitions_rejects_different_definitions() { + let mut schema = json!({"type": "record", "name": "r", "fields": [ + {"name": "a", "type": {"type": "fixed", "name": "f", "size": 4}}, + {"name": "b", "type": {"type": "fixed", "name": "f", "size": 8}}, + ]}); + + assert_eq!(reference_repeated_definitions(&mut schema), None); + } +} diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index 908dff5c9f..96999ebedc 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/crates/iceberg/src/avro/schema.rs @@ -16,12 +16,12 @@ // under the License. //! Conversion between iceberg and avro schema. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; use apache_avro::Schema as AvroSchema; use apache_avro::schema::{ - ArraySchema, DecimalSchema, FixedSchema, MapSchema, Name, RecordField as AvroRecordField, - RecordFieldOrder, RecordSchema, UnionSchema, + ArraySchema, DecimalSchema, FixedSchema, InnerDecimalSchema, MapSchema, Name, + RecordField as AvroRecordField, RecordSchema, UnionSchema, UuidSchema, }; use itertools::{Either, Itertools}; use serde_json::{Number, Value}; @@ -43,6 +43,29 @@ const LOGICAL_TYPE: &str = "logicalType"; struct SchemaToAvroSchema { schema: String, + /// Names of the `fixed` types defined so far. Avro allows one definition per + /// name, and the visitor reaches fields in the order they're serialized. + defined_names: HashSet, +} + +impl SchemaToAvroSchema { + /// Returns a reference to `schema` by name if a type with its name was + /// already defined, and `schema` itself otherwise. + fn define_once(&mut self, schema: AvroSchema) -> AvroSchema { + let name = match &schema { + AvroSchema::Fixed(FixedSchema { name, .. }) + | AvroSchema::Decimal(DecimalSchema { + inner: InnerDecimalSchema::Fixed(FixedSchema { name, .. }), + .. + }) => name.clone(), + _ => return schema, + }; + if self.defined_names.insert(name.clone()) { + schema + } else { + AvroSchema::Ref { name } + } + } } type AvroSchemaOrField = Either; @@ -54,7 +77,7 @@ impl SchemaVisitor for SchemaToAvroSchema { let mut avro_schema = value.unwrap_left(); if let AvroSchema::Record(record) = &mut avro_schema { - record.name = Name::from(self.schema.as_str()); + record.name = Name::new(self.schema.as_str())?; } else { return Err(Error::new( ErrorKind::Unexpected, @@ -72,7 +95,7 @@ impl SchemaVisitor for SchemaToAvroSchema { ) -> Result { let mut field_schema = avro_schema.unwrap_left(); if let AvroSchema::Record(record) = &mut field_schema { - record.name = Name::from(format!("r{}", field.id).as_str()); + record.name = Name::new(format!("r{}", field.id))?; } if !field.required { @@ -96,10 +119,8 @@ impl SchemaVisitor for SchemaToAvroSchema { let mut avro_record_field = AvroRecordField { name: field.name.clone(), schema: field_schema, - order: RecordFieldOrder::Ignore, - position: 0, doc: field.doc.clone(), - aliases: None, + aliases: Vec::new(), default, custom_attributes: Default::default(), }; @@ -130,7 +151,7 @@ impl SchemaVisitor for SchemaToAvroSchema { let mut field_schema = value.unwrap_left(); if let AvroSchema::Record(record) = &mut field_schema { - record.name = Name::from(format!("r{}", list.element_field.id).as_str()); + record.name = Name::new(format!("r{}", list.element_field.id))?; } if !list.element_field.required { @@ -191,11 +212,9 @@ impl SchemaVisitor for SchemaToAvroSchema { let mut field = AvroRecordField { name: map.key_field.name.clone(), doc: None, - aliases: None, + aliases: Vec::new(), default: None, schema: key_field_schema, - order: RecordFieldOrder::Ascending, - position: 0, custom_attributes: Default::default(), }; field.custom_attributes.insert( @@ -209,11 +228,9 @@ impl SchemaVisitor for SchemaToAvroSchema { let mut field = AvroRecordField { name: map.value_field.name.clone(), doc: None, - aliases: None, + aliases: Vec::new(), default: None, schema: value_field_schema, - order: RecordFieldOrder::Ignore, - position: 0, custom_attributes: Default::default(), }; field.custom_attributes.insert( @@ -254,11 +271,11 @@ impl SchemaVisitor for SchemaToAvroSchema { PrimitiveType::TimestampNs => AvroSchema::TimestampNanos, PrimitiveType::TimestamptzNs => AvroSchema::TimestampNanos, PrimitiveType::String => AvroSchema::String, - PrimitiveType::Uuid => AvroSchema::Uuid, - PrimitiveType::Fixed(len) => avro_fixed_schema((*len) as usize)?, + PrimitiveType::Uuid => AvroSchema::Uuid(UuidSchema::String), + PrimitiveType::Fixed(len) => self.define_once(avro_fixed_schema((*len) as usize)?), PrimitiveType::Binary => AvroSchema::Bytes, PrimitiveType::Decimal { precision, scale } => { - avro_decimal_schema(*precision as usize, *scale as usize)? + self.define_once(avro_decimal_schema(*precision as usize, *scale as usize)?) } }; Ok(Either::Left(avro_schema)) @@ -276,6 +293,7 @@ impl SchemaVisitor for SchemaToAvroSchema { pub(crate) fn schema_to_avro_schema(name: impl ToString, schema: &Schema) -> Result { let mut converter = SchemaToAvroSchema { schema: name.to_string(), + defined_names: HashSet::new(), }; visit_schema(schema, &mut converter).map(Either::unwrap_left) @@ -305,7 +323,6 @@ pub(crate) fn avro_fixed_schema(len: usize) -> Result { doc: None, size: len, attributes: Default::default(), - default: None, })) } @@ -317,16 +334,15 @@ pub(crate) fn avro_decimal_schema(precision: usize, scale: usize) -> Result Type::Primitive(PrimitiveType::Long), AvroSchema::Float => Type::Primitive(PrimitiveType::Float), AvroSchema::Double => Type::Primitive(PrimitiveType::Double), - AvroSchema::Uuid => Type::Primitive(PrimitiveType::Uuid), + AvroSchema::Uuid(_) => Type::Primitive(PrimitiveType::Uuid), AvroSchema::String | AvroSchema::Enum(_) => Type::Primitive(PrimitiveType::String), AvroSchema::Fixed(fixed) => Type::Primitive(PrimitiveType::Fixed(fixed.size as u64)), AvroSchema::Bytes => Type::Primitive(PrimitiveType::Binary), @@ -627,7 +643,7 @@ mod tests { use std::sync::Arc; use apache_avro::Schema as AvroSchema; - use apache_avro::schema::{Namespace, UnionSchema}; + use apache_avro::schema::UnionSchema; use super::*; use crate::avro::schema::AvroSchemaToSchema; @@ -656,11 +672,9 @@ mod tests { assert_eq!(iceberg_schema, converted_iceberg_schema); // 2. iceberg to avro - let converted_avro_schema = schema_to_avro_schema( - avro_schema.name().unwrap().fullname(Namespace::None), - &iceberg_schema, - ) - .unwrap(); + let converted_avro_schema = + schema_to_avro_schema(avro_schema.name().unwrap().fullname(None), &iceberg_schema) + .unwrap(); assert_eq!(avro_schema, converted_avro_schema); // 3.iceberg to avro to iceberg back @@ -975,6 +989,10 @@ mod tests { } #[test] + #[ignore = "apache-avro 0.22 drops `logicalType: map` when parsing schema JSON \ + (https://github.com/apache/avro-rs/issues/654), so avro_schema_to_schema \ + reads a parsed map array as a list. Fixed by \ + https://github.com/apache/avro-rs/pull/656, which is not yet released."] fn test_schema_with_array_map() { let avro_schema = { AvroSchema::parse_str( @@ -1202,7 +1220,14 @@ mod tests { fn test_unknown_primitive() { let mut converter = AvroSchemaToSchema; - assert!(converter.primitive(&AvroSchema::Duration).is_err()); + let duration = AvroSchema::Duration(FixedSchema { + name: Name::new("duration").unwrap(), + aliases: None, + doc: None, + size: 12, + attributes: Default::default(), + }); + assert!(converter.primitive(&duration).is_err()); } #[test] diff --git a/crates/iceberg/src/encryption/key_metadata.rs b/crates/iceberg/src/encryption/key_metadata.rs index d56c8dbd3b..b3222eec53 100644 --- a/crates/iceberg/src/encryption/key_metadata.rs +++ b/crates/iceberg/src/encryption/key_metadata.rs @@ -98,7 +98,7 @@ impl StandardKeyMetadata { /// Encodes to Java-compatible format: `[0x01] [Avro binary datum]` pub fn encode(&self) -> Result> { - _serde::StandardKeyMetadataV1::from(self).encode() + _serde::StandardKeyMetadataV1::try_from(self)?.encode() } /// Decodes from Java-compatible format. @@ -131,7 +131,9 @@ mod _serde { use std::io::Cursor; use std::sync::{Arc, LazyLock}; - use apache_avro::{Schema as AvroSchema, from_avro_datum, from_value, to_avro_datum, to_value}; + use apache_avro::reader::datum::GenericDatumReader; + use apache_avro::writer::datum::GenericDatumWriter; + use apache_avro::{Schema as AvroSchema, from_value, to_value}; use serde::{Deserialize, Serialize}; use super::*; @@ -173,7 +175,7 @@ mod _serde { pub(super) struct StandardKeyMetadataV1 { pub encryption_key: serde_bytes::ByteBuf, pub aad_prefix: Option, - pub file_length: Option, + pub file_length: Option, } impl StandardKeyMetadataV1 { @@ -185,9 +187,13 @@ mod _serde { .with_source(e) })?; - let datum = to_avro_datum(&AVRO_SCHEMA_V1, value).map_err(|e| { - Error::new(ErrorKind::Unexpected, "Failed to encode key metadata").with_source(e) - })?; + let datum = GenericDatumWriter::builder(&AVRO_SCHEMA_V1) + .build() + .and_then(|writer| writer.write_value_to_vec(value)) + .map_err(|e| { + Error::new(ErrorKind::Unexpected, "Failed to encode key metadata") + .with_source(e) + })?; let mut result = Vec::with_capacity(1 + datum.len()); result.push(V1); @@ -209,7 +215,9 @@ mod _serde { } let mut reader = Cursor::new(&bytes[1..]); - let value = from_avro_datum(&AVRO_SCHEMA_V1, &mut reader, None) + let value = GenericDatumReader::builder(&AVRO_SCHEMA_V1) + .build() + .and_then(|datum_reader| datum_reader.read_value(&mut reader)) .map_err(|e| invalid_data!("Failed to decode key metadata").with_source(e))?; from_value(&value) @@ -217,16 +225,29 @@ mod _serde { } } - impl From<&StandardKeyMetadata> for StandardKeyMetadataV1 { - fn from(metadata: &StandardKeyMetadata) -> Self { - Self { + impl TryFrom<&StandardKeyMetadata> for StandardKeyMetadataV1 { + type Error = Error; + + fn try_from(metadata: &StandardKeyMetadata) -> Result { + let file_length = metadata + .file_length + .map(i64::try_from) + .transpose() + .map_err(|e| { + Error::new( + ErrorKind::DataInvalid, + "Key metadata file length exceeds the Avro long range", + ) + .with_source(e) + })?; + Ok(Self { encryption_key: serde_bytes::ByteBuf::from(metadata.encryption_key.as_bytes()), aad_prefix: metadata .aad_prefix .as_ref() .map(|b| serde_bytes::ByteBuf::from(b.as_ref())), - file_length: metadata.file_length, - } + file_length, + }) } } @@ -240,7 +261,9 @@ mod _serde { Ok(Self { encryption_key, aad_prefix: v1.aad_prefix.map(|b| b.into_vec().into_boxed_slice()), - file_length: v1.file_length, + file_length: v1.file_length.map(u64::try_from).transpose().map_err(|e| { + invalid_data!("Negative file length in key metadata").with_source(e) + })?, }) } } @@ -328,6 +351,30 @@ mod tests { } } + #[test] + fn test_encode_rejects_file_length_above_long_max() { + let metadata = StandardKeyMetadata::try_new(&[0u8; 16]) + .unwrap() + .with_file_length(i64::MAX as u64 + 1); + + let err = metadata.encode().unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + #[test] + fn test_decode_rejects_negative_file_length() { + let serialized = _serde::StandardKeyMetadataV1 { + encryption_key: serde_bytes::ByteBuf::from(vec![0u8; 16]), + aad_prefix: None, + file_length: Some(-1), + } + .encode() + .unwrap(); + + let err = StandardKeyMetadata::decode(&serialized).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + #[test] fn test_decode_rejects_invalid_key_length() { // Craft wire bytes carrying an invalid-length DEK directly via the diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs index 336c883f1a..fb7139e498 100644 --- a/crates/iceberg/src/spec/manifest/_serde.rs +++ b/crates/iceberg/src/spec/manifest/_serde.rs @@ -58,6 +58,7 @@ impl ManifestEntryV2 { file_sequence_number: self.file_sequence_number, data_file: self .data_file + .project_partition_by_name(partition_type)? .try_into(partition_spec_id, partition_type, schema)?, }) } @@ -92,6 +93,7 @@ impl ManifestEntryV1 { file_sequence_number: Some(0), data_file: self .data_file + .project_partition_by_name(partition_type)? .try_into(partition_spec_id, partition_type, schema)?, }) } @@ -161,6 +163,16 @@ impl DataFileSerde { }) } + /// Matches the partition value's fields to `partition_type` by name. Manifest + /// entries are deserialized with the writer's partition fields and order, + /// which can differ from the partition spec's. + fn project_partition_by_name(mut self, partition_type: &Type) -> Result { + if let Type::Struct(struct_type) = partition_type { + self.partition = self.partition.project_by_name(struct_type)?; + } + Ok(self) + } + pub fn try_into( self, partition_spec_id: i32, diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index 1fcdaa3a80..94568ab44e 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -301,7 +301,7 @@ pub fn write_data_files_to_avro( FormatVersion::V2 => data_file_schema_v2(partition_type).unwrap(), FormatVersion::V3 => data_file_schema_v3(partition_type).unwrap(), }; - let mut writer = AvroWriter::new(&avro_schema, writer); + let mut writer = AvroWriter::new(&avro_schema, writer)?; // Wrap once and reuse across files so the field-name lookup is not rebuilt // for each data file. @@ -313,7 +313,7 @@ pub fn write_data_files_to_avro( FormatVersion::V1, )?)? .resolve(&avro_schema)?; - writer.append(value)?; + writer.append_value(value)?; } Ok(writer.flush()?) @@ -333,7 +333,9 @@ pub fn read_data_files_from_avro( FormatVersion::V3 => data_file_schema_v3(partition_type).unwrap(), }; - let reader = AvroReader::with_schema(&avro_schema, reader)?; + let reader = AvroReader::builder(reader) + .reader_schema(&avro_schema) + .build()?; // Wrap once and reuse across files so the field-name lookup is not rebuilt // for each data file. let partition_struct_type = Type::Struct(partition_type.clone()); diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index 3bd4dc8ad8..a07851df67 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -28,13 +28,15 @@ pub use reader::*; mod writer; use std::sync::Arc; -use apache_avro::{Reader as AvroReader, from_value}; +use apache_avro::Reader as AvroReader; +use apache_avro::error::Details; 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::error::{Result, invalid_data}; /// A manifest contains metadata and a list of entries. @@ -45,9 +47,30 @@ pub struct Manifest { } impl Manifest { - /// Parse manifest metadata and entries from bytes of avro file. - pub(crate) fn try_from_avro_bytes(bs: &[u8]) -> Result<(ManifestMetadata, Vec)> { - let reader = AvroReader::new(bs)?; + /// Parse manifest metadata and entries from bytes of avro file. `location` + /// names the manifest in warnings. + pub(crate) fn try_from_avro_bytes( + bs: &[u8], + location: Option<&str>, + ) -> Result<(ManifestMetadata, Vec)> { + let rewritten; + let reader = match AvroReader::new(bs) { + Ok(reader) => reader, + Err(e) if matches!(e.details(), Details::AmbiguousSchemaDefinition(_)) => { + let Some((bs, repeated)) = define_named_types_once(bs)? else { + return Err(e.into()); + }; + tracing::warn!( + "Manifest {} defines Avro named types {repeated:?} more than once, which the \ + Avro specification doesn't allow. Reading it with each repeated definition \ + replaced by a reference to the first.", + location.unwrap_or("") + ); + rewritten = bs; + AvroReader::new(rewritten.as_slice())? + } + Err(e) => return Err(e.into()), + }; // Parse manifest metadata let meta = reader.user_metadata(); @@ -61,35 +84,27 @@ impl Manifest { let partition_struct_type = Type::Struct(partition_type.clone()); let entries = match metadata.format_version { - FormatVersion::V1 => { - let schema = manifest_schema_v1(&partition_type)?; - let reader = AvroReader::with_schema(&schema, bs)?; - reader - .into_iter() - .map(|value| { - from_value::<_serde::ManifestEntryV1>(&value?)?.try_into( - metadata.partition_spec.spec_id(), - &partition_struct_type, - &metadata.schema, - ) - }) - .collect::>>()? - } + FormatVersion::V1 => reader + .into_deser_iter::>() + .map(|entry| { + entry?.0.try_into( + metadata.partition_spec.spec_id(), + &partition_struct_type, + &metadata.schema, + ) + }) + .collect::>>()?, // Manifest Schema & Manifest Entry did not change between V2 and V3 - FormatVersion::V2 | FormatVersion::V3 => { - let schema = manifest_schema_v2(&partition_type)?; - let reader = AvroReader::with_schema(&schema, bs)?; - reader - .into_iter() - .map(|value| { - from_value::<_serde::ManifestEntryV2>(&value?)?.try_into( - metadata.partition_spec.spec_id(), - &partition_struct_type, - &metadata.schema, - ) - }) - .collect::>>()? - } + FormatVersion::V2 | FormatVersion::V3 => reader + .into_deser_iter::>() + .map(|entry| { + entry?.0.try_into( + metadata.partition_spec.spec_id(), + &partition_struct_type, + &metadata.schema, + ) + }) + .collect::>>()?, }; Ok((metadata, entries)) @@ -97,7 +112,7 @@ impl Manifest { /// Parse manifest from bytes of avro file. pub fn parse_avro(bs: &[u8]) -> Result { - let (metadata, entries) = Self::try_from_avro_bytes(bs)?; + let (metadata, entries) = Self::try_from_avro_bytes(bs, None)?; Ok(Self::new(metadata, entries)) } @@ -163,6 +178,7 @@ mod tests { use tempfile::TempDir; use super::*; + use crate::ErrorKind; use crate::io::FileIO; use crate::spec::{Literal, NestedField, PrimitiveType, Struct, Transform, Type}; @@ -356,7 +372,7 @@ mod tests { .partition_type(&metadata.schema) .unwrap(); let avro_schema = manifest_schema_v2(&partition_type).unwrap(); - let mut writer = Writer::with_codec(&avro_schema, Vec::new(), Codec::Snappy); + let mut writer = Writer::with_codec(&avro_schema, Vec::new(), Codec::Snappy).unwrap(); writer .add_user_metadata("schema".to_string(), to_vec(&metadata.schema).unwrap()) .unwrap(); @@ -397,7 +413,7 @@ mod tests { .unwrap() .resolve(&avro_schema) .unwrap(); - writer.append(value).unwrap(); + writer.append_value(value).unwrap(); let bs = writer.into_inner().unwrap(); let parsed_manifest = Manifest::parse_avro(&bs).unwrap(); @@ -1361,4 +1377,389 @@ mod tests { assert_eq!(deserialized_data_file1, original_data_file1); assert_eq!(deserialized_data_file2, original_data_file2); } + + /// Metadata for the writer schema tests: a V2 data manifest whose spec has + /// identity partitions on a long, a string, and a double column. + fn writer_schema_test_metadata() -> ManifestMetadata { + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "name", Type::Primitive(PrimitiveType::String)).into(), + NestedField::optional(3, "score", Type::Primitive(PrimitiveType::Double)) + .into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .add_partition_field("id", "id", Transform::Identity) + .unwrap() + .add_partition_field("name", "name", Transform::Identity) + .unwrap() + .add_partition_field("score", "score", Transform::Identity) + .unwrap() + .build() + .unwrap(); + ManifestMetadata { + schema_id: 0, + schema, + partition_spec, + content: ManifestContentType::Data, + format_version: FormatVersion::V2, + } + } + + /// A V2 manifest entry writer schema with Java's record names and only the + /// required `data_file` fields. + fn v2_writer_schema(partition_fields: Value) -> Value { + serde_json::json!({ + "type": "record", + "name": "manifest_entry", + "fields": [ + {"name": "status", "type": "int", "field-id": 0}, + {"name": "snapshot_id", "type": ["null", "long"], "default": null, "field-id": 1}, + {"name": "sequence_number", "type": ["null", "long"], "default": null, "field-id": 3}, + {"name": "file_sequence_number", "type": ["null", "long"], "default": null, "field-id": 4}, + {"name": "data_file", "field-id": 2, "type": { + "type": "record", + "name": "r2", + "fields": [ + {"name": "content", "type": "int", "field-id": 134}, + {"name": "file_path", "type": "string", "field-id": 100}, + {"name": "file_format", "type": "string", "field-id": 101}, + {"name": "partition", "field-id": 102, "type": { + "type": "record", + "name": "r102", + "fields": partition_fields, + }}, + {"name": "record_count", "type": "long", "field-id": 103}, + {"name": "file_size_in_bytes", "type": "long", "field-id": 104}, + ], + }}, + ], + }) + } + + /// The field named by `path` in a record schema, descending through nested + /// record types. + fn writer_schema_field<'a>(record: &'a mut Value, path: &[&str]) -> &'a mut Value { + let (name, rest) = path.split_first().unwrap(); + let field = record["fields"] + .as_array_mut() + .unwrap() + .iter_mut() + .find(|field| field["name"] == *name) + .unwrap(); + if rest.is_empty() { + field + } else { + writer_schema_field(&mut field["type"], rest) + } + } + + fn v2_entry(partition: Value) -> Value { + serde_json::json!({ + "status": 1, + "snapshot_id": 7, + "sequence_number": null, + "file_sequence_number": null, + "data_file": { + "content": 0, + "file_path": "s3://bucket/table/data/a.parquet", + "file_format": "PARQUET", + "partition": partition, + "record_count": 10, + "file_size_in_bytes": 100, + }, + }) + } + + fn expected_entry(partition: Struct) -> ManifestEntry { + ManifestEntry { + status: ManifestStatus::Added, + snapshot_id: Some(7), + sequence_number: None, + file_sequence_number: None, + data_file: DataFile { + content: DataContentType::Data, + file_path: "s3://bucket/table/data/a.parquet".to_string(), + file_format: DataFileFormat::Parquet, + partition, + record_count: 10, + file_size_in_bytes: 100, + column_sizes: HashMap::new(), + value_counts: HashMap::new(), + null_value_counts: HashMap::new(), + nan_value_counts: HashMap::new(), + lower_bounds: HashMap::new(), + upper_bounds: HashMap::new(), + key_metadata: None, + split_offsets: None, + equality_ids: None, + sort_order_id: None, + partition_spec_id: 0, + first_row_id: None, + referenced_data_file: None, + content_offset: None, + content_size_in_bytes: None, + }, + } + } + + /// Writes `entries` as a manifest with the given writer schema, encoding each + /// JSON value with the writer schema's types. + fn write_with_writer_schema( + metadata: &ManifestMetadata, + writer_schema: &Value, + entries: Vec, + ) -> Vec { + let avro_schema = apache_avro::Schema::parse(writer_schema).unwrap(); + let mut writer = Writer::new(&avro_schema, Vec::new()).unwrap(); + for (key, value) in [ + ("schema", to_vec(&metadata.schema).unwrap()), + ("schema-id", metadata.schema_id.to_string().into_bytes()), + ( + "partition-spec", + to_vec(&metadata.partition_spec.fields()).unwrap(), + ), + ( + "partition-spec-id", + metadata.partition_spec.spec_id().to_string().into_bytes(), + ), + ( + "format-version", + (metadata.format_version as u8).to_string().into_bytes(), + ), + ("content", metadata.content.to_string().into_bytes()), + ] { + writer.add_user_metadata(key.to_string(), value).unwrap(); + } + for entry in entries { + let value = apache_avro::types::Value::try_from(entry) + .unwrap() + .resolve(&avro_schema) + .unwrap(); + writer.append_value(value).unwrap(); + } + writer.into_inner().unwrap() + } + + #[test] + fn test_parse_manifest_matches_partition_fields_by_name() { + // The writer orders the partition fields differently from the spec, omits + // `name`, and adds a field the spec doesn't have. + let metadata = writer_schema_test_metadata(); + let writer_schema = v2_writer_schema(serde_json::json!([ + {"name": "score", "type": ["null", "double"], "default": null, "field-id": 1002}, + {"name": "extra", "type": ["null", "int"], "default": null, "field-id": 1003}, + {"name": "id", "type": ["null", "long"], "default": null, "field-id": 1000}, + ])); + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![v2_entry( + serde_json::json!({"score": 2.5, "extra": 9, "id": 5}), + )]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let partition = + Struct::from_iter([Some(Literal::long(5)), None, Some(Literal::double(2.5))]); + assert_eq!( + manifest, + Manifest::new(metadata, vec![expected_entry(partition)]) + ); + } + + #[test] + fn test_parse_manifest_promotes_writer_types() { + // Avro promotes int to long and float to double. + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(serde_json::json!([ + {"name": "id", "type": ["null", "int"], "default": null, "field-id": 1000}, + {"name": "name", "type": ["null", "string"], "default": null, "field-id": 1001}, + {"name": "score", "type": ["null", "float"], "default": null, "field-id": 1002}, + ])); + for field in ["record_count", "file_size_in_bytes"] { + writer_schema_field(&mut writer_schema, &["data_file", field])["type"] = + serde_json::json!("int"); + } + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![v2_entry( + serde_json::json!({"id": 5, "name": "a", "score": 2.5}), + )]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let partition = Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ]); + assert_eq!( + manifest, + Manifest::new(metadata, vec![expected_entry(partition)]) + ); + } + + #[test] + fn test_parse_manifest_reads_required_writer_fields_as_optional() { + // Fields that are optional in the reader schema are written without a + // union. + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(serde_json::json!([ + {"name": "id", "type": "long", "field-id": 1000}, + {"name": "name", "type": "string", "field-id": 1001}, + {"name": "score", "type": "double", "field-id": 1002}, + ])); + *writer_schema_field(&mut writer_schema, &["snapshot_id"]) = + serde_json::json!({"name": "snapshot_id", "type": "long", "field-id": 1}); + writer_schema_field(&mut writer_schema, &["data_file"])["type"]["fields"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({"name": "sort_order_id", "type": "int", "field-id": 140})); + let mut entry = v2_entry(serde_json::json!({"id": 5, "name": "a", "score": 2.5})); + entry["data_file"]["sort_order_id"] = serde_json::json!(3); + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let mut expected = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + expected.data_file.sort_order_id = Some(3); + assert_eq!(manifest, Manifest::new(metadata, vec![expected])); + } + + #[test] + fn test_parse_manifest_ignores_record_names_and_unknown_fields() { + // Record names differ from the ones Java writes, and the writer has fields + // the reader schema doesn't. + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(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": 1002}, + ])); + writer_schema["name"] = serde_json::json!("entry"); + writer_schema["fields"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({"name": "unknown", "type": "string"})); + let data_file = &mut writer_schema_field(&mut writer_schema, &["data_file"])["type"]; + data_file["name"] = serde_json::json!("file"); + data_file["fields"][3]["type"]["name"] = serde_json::json!("part"); + let data_file_fields = data_file["fields"].as_array_mut().unwrap(); + data_file_fields.push(serde_json::json!({ + "name": "column_sizes", + "field-id": 108, + "default": null, + "type": ["null", { + "type": "array", + "logicalType": "map", + "items": { + "type": "record", + "name": "sizes", + "fields": [ + {"name": "key", "type": "int", "field-id": 117}, + {"name": "value", "type": "long", "field-id": 118}, + ], + }, + }], + })); + data_file_fields.push( + serde_json::json!({"name": "block_size_in_bytes", "type": "long", "field-id": 105}), + ); + let mut entry = v2_entry(serde_json::json!({"id": 5, "name": "a", "score": 2.5})); + entry["unknown"] = serde_json::json!("ignored"); + entry["data_file"]["column_sizes"] = serde_json::json!([{"key": 1, "value": 40}]); + entry["data_file"]["block_size_in_bytes"] = serde_json::json!(64); + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let mut expected = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + expected.data_file.column_sizes = HashMap::from([(1, 40)]); + assert_eq!(manifest, Manifest::new(metadata, vec![expected])); + } + + #[test] + fn test_parse_manifest_equality_ids_written_as_long() { + // PyIceberg wrote `equality_ids` as `array` before + // apache/iceberg-python#3842. + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(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": 1002}, + ])); + writer_schema_field(&mut writer_schema, &["data_file"])["type"]["fields"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "name": "equality_ids", + "field-id": 135, + "default": null, + "type": ["null", {"type": "array", "items": "long", "element-id": 136}], + })); + let mut entry = v2_entry(serde_json::json!({"id": 5, "name": "a", "score": 2.5})); + entry["data_file"]["equality_ids"] = serde_json::json!([1, 2]); + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let mut expected = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + expected.data_file.equality_ids = Some(vec![1, 2]); + assert_eq!(manifest, Manifest::new(metadata, vec![expected])); + } + + #[test] + fn test_parse_manifest_rejects_equality_id_above_int_range() { + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(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": 1002}, + ])); + writer_schema_field(&mut writer_schema, &["data_file"])["type"]["fields"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "name": "equality_ids", + "field-id": 135, + "default": null, + "type": ["null", {"type": "array", "items": "long", "element-id": 136}], + })); + let mut entry = v2_entry(serde_json::json!({"id": 5, "name": "a", "score": 2.5})); + entry["data_file"]["equality_ids"] = serde_json::json!([i64::from(i32::MAX) + 1]); + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let err = Manifest::parse_avro(&bs).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + #[test] + fn test_parse_manifest_with_repeated_named_type_definitions() { + let bs = fs::read(format!( + "{}/testdata/manifests/repeated-decimal-type-definitions.avro", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + assert_eq!( + *manifest.entries()[0].data_file().partition(), + Struct::from_iter([Some(Literal::decimal(12345)), Some(Literal::decimal(-678))]) + ); + } } diff --git a/crates/iceberg/src/spec/manifest/reader.rs b/crates/iceberg/src/spec/manifest/reader.rs index e7044a56bc..a056d207c4 100644 --- a/crates/iceberg/src/spec/manifest/reader.rs +++ b/crates/iceberg/src/spec/manifest/reader.rs @@ -51,7 +51,8 @@ impl ManifestReader { None => input_file.read().await?, }; - let (metadata, mut entries) = Manifest::try_from_avro_bytes(&bytes)?; + let (metadata, mut entries) = + Manifest::try_from_avro_bytes(&bytes, Some(&manifest_file.manifest_path))?; for entry in &mut entries { entry.inherit_data(manifest_file); diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index 15608a63f8..52a11a6e5b 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -432,7 +432,7 @@ impl ManifestWriter { // Manifest schema did not change between V2 and V3 FormatVersion::V2 | FormatVersion::V3 => manifest_schema_v2(&partition_type)?, }; - let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new()); + let mut avro_writer = AvroWriter::new(&avro_schema, Vec::new())?; avro_writer.add_user_metadata( "schema".to_string(), to_vec(table_schema) @@ -479,7 +479,7 @@ impl ManifestWriter { } }; - avro_writer.append(value)?; + avro_writer.append_value(value)?; } let content = avro_writer.into_inner()?; @@ -581,7 +581,10 @@ mod tests { use super::*; use crate::io::FileIO; - use crate::spec::{DataFileFormat, Manifest, NestedField, PrimitiveType, Schema, Struct, Type}; + use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Literal, Manifest, NestedField, + PrimitiveType, Schema, Struct, Transform, Type, + }; #[tokio::test] async fn test_add_delete_existing() { @@ -819,4 +822,75 @@ mod tests { ManifestContentType::Deletes, ); } + + /// Writes a manifest partitioned by identity on two columns of `field_type` + /// and reads it back. + async fn roundtrip_two_partition_fields_of_type(field_type: Type, values: [Literal; 2]) { + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "a", field_type.clone()).into(), + NestedField::optional(2, "b", field_type).into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .add_partition_field("a", "a", Transform::Identity) + .unwrap() + .add_partition_field("b", "b", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let partition = Struct::from_iter(values.map(Some)); + let tmp_dir = TempDir::new().unwrap(); + let path = tmp_dir.path().join("manifest.avro"); + let output_file = FileIO::new_with_fs() + .new_output(path.to_str().unwrap()) + .unwrap(); + let mut writer = ManifestWriterBuilder::new(output_file, Some(1), schema, partition_spec) + .build_v2_data(); + writer + .add_file( + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("s3://bucket/table/data/a.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(partition.clone()) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap(), + 1, + ) + .unwrap(); + writer.write_manifest_file().await.unwrap(); + + let manifest = Manifest::parse_avro(&fs::read(&path).unwrap()).unwrap(); + + assert_eq!(*manifest.entries()[0].data_file().partition(), partition); + } + + #[tokio::test] + async fn test_write_manifest_with_repeated_decimal_partition_type() { + roundtrip_two_partition_fields_of_type( + Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }), + [Literal::decimal(12345), Literal::decimal(-678)], + ) + .await; + } + + #[tokio::test] + async fn test_write_manifest_with_repeated_fixed_partition_type() { + roundtrip_two_partition_fields_of_type(Type::Primitive(PrimitiveType::Fixed(4)), [ + Literal::fixed(vec![1, 2, 3, 4]), + Literal::fixed(vec![5, 6, 7, 8]), + ]) + .await; + } } diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index 25dd5bb93e..24a4debf9e 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/crates/iceberg/src/spec/manifest_list/mod.rs @@ -61,7 +61,9 @@ impl ManifestList { pub fn parse_with_version(bs: &[u8], version: FormatVersion) -> Result { match version { FormatVersion::V1 => { - let reader = Reader::with_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1, bs)?; + let reader = Reader::builder(bs) + .reader_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1) + .build()?; let values = Value::Array(reader.collect::, _>>()?); from_value::<_serde::ManifestListV1>(&values)?.try_into() } @@ -265,7 +267,7 @@ mod test { let manifest_entry: ManifestFileV2 = manifest_list.entries[0].clone().try_into().unwrap(); let mut writer = - Writer::with_codec(&MANIFEST_LIST_AVRO_SCHEMA_V2, Vec::new(), Codec::Snappy); + Writer::with_codec(&MANIFEST_LIST_AVRO_SCHEMA_V2, Vec::new(), Codec::Snappy).unwrap(); writer.append_ser(manifest_entry).unwrap(); let bs = writer.into_inner().unwrap(); diff --git a/crates/iceberg/src/spec/manifest_list/writer.rs b/crates/iceberg/src/spec/manifest_list/writer.rs index 5ce2cfb259..eb8a50b6ae 100644 --- a/crates/iceberg/src/spec/manifest_list/writer.rs +++ b/crates/iceberg/src/spec/manifest_list/writer.rs @@ -149,7 +149,8 @@ impl ManifestListWriter { FormatVersion::V2 => &MANIFEST_LIST_AVRO_SCHEMA_V2, FormatVersion::V3 => &MANIFEST_LIST_AVRO_SCHEMA_V3, }; - let mut avro_writer = Writer::new(avro_schema, Vec::new()); + let mut avro_writer = Writer::new(avro_schema, Vec::new()) + .expect("Manifest list Avro schemas have no named references to resolve."); for (key, value) in metadata { avro_writer .add_user_metadata(key, value) diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs index 896235b1e3..27726a050c 100644 --- a/crates/iceberg/src/spec/values/serde.rs +++ b/crates/iceberg/src/spec/values/serde.rs @@ -18,6 +18,8 @@ //\! Serialization and deserialization support for Iceberg values pub(crate) mod _serde { + use std::collections::HashMap; + use serde::de::Visitor; use serde::ser::{SerializeMap, SerializeSeq, SerializeStruct}; use serde::{Deserialize, Serialize}; @@ -27,7 +29,7 @@ pub(crate) mod _serde { use crate::Error; use crate::error::invalid_data; use crate::spec::values::{Literal, Map, PrimitiveLiteral, Struct}; - use crate::spec::{MAP_KEY_FIELD_NAME, MAP_VALUE_FIELD_NAME, PrimitiveType, Type}; + use crate::spec::{MAP_KEY_FIELD_NAME, MAP_VALUE_FIELD_NAME, PrimitiveType, StructType, Type}; #[derive(SerializeDerive, DeserializeDerive, Debug)] #[serde(transparent)] @@ -44,6 +46,49 @@ pub(crate) mod _serde { pub fn try_into(self, ty: &Type) -> Result, Error> { self.0.try_into(ty) } + + /// Matches the fields of a record to `struct_type` by name, the way Avro + /// schema resolution matches record fields. The result has + /// `struct_type`'s fields in its order, with null for an optional field the + /// record lacks. Record fields that `struct_type` lacks are dropped. Values + /// other than records are returned unchanged. + pub fn project_by_name(self, struct_type: &StructType) -> Result { + let (mut required, optional) = match self.0 { + RawLiteralEnum::Record(Record { required, optional }) => (required, optional), + other => return Ok(Self(other)), + }; + required.extend( + optional + .into_iter() + .map(|(name, value)| (name, value.unwrap_or(RawLiteralEnum::Null))), + ); + let fields = struct_type.fields(); + let required = if required.len() == fields.len() + && required + .iter() + .zip(fields) + .all(|((name, _), field)| *name == field.name) + { + required + } else { + let mut values: HashMap = required.into_iter().collect(); + fields + .iter() + .map(|field| match values.remove(&field.name) { + Some(value) => Ok((field.name.clone(), value)), + None if field.required => Err(invalid_data!( + "Record has no value for required field {}", + field.name + )), + None => Ok((field.name.clone(), RawLiteralEnum::Null)), + }) + .collect::>()? + }; + Ok(Self(RawLiteralEnum::Record(Record { + required, + optional: Vec::new(), + }))) + } } #[derive(SerializeDerive, Clone, Debug)] diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index ff51f318e3..89e84741e5 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -64,10 +64,13 @@ fn check_avro_bytes_serde(input: Vec, expected_datum: Datum, expected_type: let datum = Datum::try_from_bytes(&bytes, expected_type.clone()).unwrap(); assert_eq!(datum, expected_datum); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).unwrap(); writer.append_ser(datum.to_bytes().unwrap()).unwrap(); let encoded = writer.into_inner().unwrap(); - let reader = apache_avro::Reader::with_schema(&schema, &*encoded).unwrap(); + let reader = apache_avro::Reader::builder(&*encoded) + .reader_schema(&schema) + .build() + .unwrap(); for record in reader { let result = apache_avro::from_value::(&record.unwrap()).unwrap(); @@ -86,7 +89,7 @@ fn check_convert_with_avro(expected_literal: Literal, expected_type: &Type) { let struct_type = Type::Struct(StructType::new(fields)); let struct_literal = Literal::Struct(Struct::from_iter(vec![Some(expected_literal.clone())])); - let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()).unwrap(); let raw_literal = RawLiteral::try_from(struct_literal.clone(), &struct_type).unwrap(); writer.append_ser(raw_literal).unwrap(); let encoded = writer.into_inner().unwrap(); @@ -110,7 +113,7 @@ fn check_serialize_avro(literal: Literal, ty: &Type, expect_value: Value) { let avro_schema = schema_to_avro_schema("test", &schema).unwrap(); let struct_type = Type::Struct(StructType::new(fields)); let struct_literal = Literal::Struct(Struct::from_iter(vec![Some(literal.clone())])); - let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()).unwrap(); let raw_literal = RawLiteral::try_from(struct_literal.clone(), &struct_type).unwrap(); let value = to_value(raw_literal) .unwrap() @@ -924,7 +927,7 @@ fn check_convert_with_avro_map(expected_literal: Literal, expected_type: &Type) let struct_type = Type::Struct(StructType::new(fields)); let struct_literal = Literal::Struct(Struct::from_iter(vec![Some(expected_literal.clone())])); - let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&avro_schema, Vec::new()).unwrap(); let raw_literal = RawLiteral::try_from(struct_literal.clone(), &struct_type).unwrap(); writer.append_ser(raw_literal).unwrap(); let encoded = writer.into_inner().unwrap(); @@ -1719,3 +1722,60 @@ fn test_datum_to_decimal_rejects_scale_change() { .contains("Decimal scale conversion is not supported") ); } + +#[test] +fn raw_literal_project_by_name_reorders_and_fills_fields() { + let written = StructType::new(vec![ + NestedField::required(1, "a", Primitive(PrimitiveType::Int)).into(), + NestedField::optional(2, "b", Primitive(PrimitiveType::Int)).into(), + NestedField::optional(3, "extra", Primitive(PrimitiveType::Int)).into(), + ]); + let raw = RawLiteral::try_from( + Literal::Struct(Struct::from_iter([ + Some(Literal::int(1)), + Some(Literal::int(2)), + Some(Literal::int(3)), + ])), + &Type::Struct(written), + ) + .unwrap(); + let expected = StructType::new(vec![ + NestedField::optional(4, "c", Primitive(PrimitiveType::Int)).into(), + NestedField::optional(2, "b", Primitive(PrimitiveType::Int)).into(), + NestedField::required(1, "a", Primitive(PrimitiveType::Int)).into(), + ]); + + let projected = raw + .project_by_name(&expected) + .unwrap() + .try_into(&Type::Struct(expected)) + .unwrap(); + + assert_eq!( + projected, + Some(Literal::Struct(Struct::from_iter([ + None, + Some(Literal::int(2)), + Some(Literal::int(1)), + ]))) + ); +} + +#[test] +fn raw_literal_project_by_name_rejects_missing_required_field() { + let written = StructType::new(vec![ + NestedField::optional(1, "a", Primitive(PrimitiveType::Int)).into(), + ]); + let raw = RawLiteral::try_from( + Literal::Struct(Struct::from_iter([Some(Literal::int(1))])), + &Type::Struct(written), + ) + .unwrap(); + let expected = StructType::new(vec![ + NestedField::optional(1, "a", Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "b", Primitive(PrimitiveType::Int)).into(), + ]); + + let err = raw.project_by_name(&expected).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); +} diff --git a/crates/iceberg/testdata/manifests/README.md b/crates/iceberg/testdata/manifests/README.md new file mode 100644 index 0000000000..657243417c --- /dev/null +++ b/crates/iceberg/testdata/manifests/README.md @@ -0,0 +1,37 @@ + + + +# Manifest test data + +`repeated-decimal-type-definitions.avro` is a V2 data manifest written by +iceberg-rust at commit `ecba0b5`, which used `apache-avro` 0.21. Its table schema +has two optional `decimal(10, 2)` columns, `d1` (id 1) and `d2` (id 2), and its +partition spec has an identity partition on each. It holds one entry for +`s3://b/t/a.parquet` with partition values `123.45` and `-6.78`. + +The Avro schema in its header defines the named fixed type `decimal_10_2` twice. +The Avro specification allows only one definition of each name, and +`apache-avro` 0.22 rejects the header, but iceberg-rust wrote manifests like +this one before it moved to 0.22. Tests use it to check that such manifests +still read. + +To regenerate it, check out a commit that uses `apache-avro` 0.21, write the +manifest described above with `ManifestWriterBuilder::build_v2_data`, and copy +the file here. diff --git a/crates/iceberg/testdata/manifests/repeated-decimal-type-definitions.avro b/crates/iceberg/testdata/manifests/repeated-decimal-type-definitions.avro new file mode 100644 index 0000000000000000000000000000000000000000..598268c03003ddd9ba71cdf7ae46a69146efe0d8 GIT binary patch literal 3865 zcmd5yC1ri!jfaD~*vw7Ja1vL*r0Vo=ccW23veK9i|FvvX>1wTLm zUx36XP{A*uaA)@6S+8S<#RX0iujjtbJ?GBd2cuW(SFX`hk#75G!2C-|9z`2|nkbg2 zO955%Mn;9=il@nz%$WbN_u@U0X)b)W#RD{S`zYir3W_g|-2BoyC!9A4Y^j*(F->^Lq#^*77&o&ok4QyTF5#!LP@dFP77%E7tFsBsIciixRAo^|d`JkgeeXZdp|K?BJ%JjH=_QG4PKAz6~44u@W{mS{hXMZ_3)Xm1|km zYOHO}FVl7A7g{LO$fP1e?Ih%HK@2q?OBoQ?A=;PZ??797nmaGKPopP=*2+`*06c^f zAd?KJ%_WQtm=VroAUR$fj7{dmG($_saX1^KfO<8x51@AJbg6wXo7NPlb^|de z04Vg=wcA1TbS4Jn6#{@=4Z!>{vAFrzo=Qubk8vjeU~)_^vxFuquira_`n^F10(dQ6 zE+T+8=v2RFD=bRiFhNEzXM&ZGAMfklA9Sc4?^|t|>3ucKtg8|qbSfV2SyA#VTW#@L z`Cb?0V_THVi}$)oAKUZ-C6bsas&|~4W!1<>1>y^DSkD%#%p$Hx8itaoixa7u=qzF} zET_f3?|F-dw`X6Ju&j`nI%uCZl<+jO}=md$m;ybGFrqrI%u zX{aIVPExqB+&S*#N&~0!F?<;EK;l0O?p5A`~zpgoh-r3pN@8|mL^bdY8 IpINtm03fg3 Date: Mon, 5 Oct 2026 15:49:55 -0700 Subject: [PATCH 2/5] add test cases, update docs --- crates/iceberg/src/avro/deserializer.rs | 2 + crates/iceberg/src/spec/manifest/mod.rs | 108 +++++++++++++++++++++++- 2 files changed, 106 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/avro/deserializer.rs b/crates/iceberg/src/avro/deserializer.rs index 1401247eb8..79f4fb623d 100644 --- a/crates/iceberg/src/avro/deserializer.rs +++ b/crates/iceberg/src/avro/deserializer.rs @@ -26,6 +26,8 @@ //! //! - Avro record names don't have to match serde type names. //! - A writer value that isn't a union reads into an `Option`. +//! - A writer union reads into a type that isn't an `Option`. A null value +//! returns an error. //! - serde's numeric visitors convert between numeric types, so an `int` reads //! into an `i64` and a `float` into an `f64`. An integer that doesn't fit the //! target, such as a `long` above `i32::MAX` read into an `i32`, returns an diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index a07851df67..fd4521156f 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -173,6 +173,7 @@ mod tests { use std::fs; use std::sync::Arc; + use apache_avro::types::Value as AvroValue; use apache_avro::{Codec, Writer, to_value}; use serde_json::{Value, to_vec}; use tempfile::TempDir; @@ -1514,6 +1515,22 @@ mod tests { metadata: &ManifestMetadata, writer_schema: &Value, entries: Vec, + ) -> Vec { + write_avro_values_with_writer_schema( + metadata, + writer_schema, + entries + .into_iter() + .map(|entry| AvroValue::try_from(entry).unwrap()) + .collect(), + ) + } + + /// Like [`write_with_writer_schema`], for entries that JSON can't express. + fn write_avro_values_with_writer_schema( + metadata: &ManifestMetadata, + writer_schema: &Value, + entries: Vec, ) -> Vec { let avro_schema = apache_avro::Schema::parse(writer_schema).unwrap(); let mut writer = Writer::new(&avro_schema, Vec::new()).unwrap(); @@ -1537,11 +1554,9 @@ mod tests { writer.add_user_metadata(key.to_string(), value).unwrap(); } for entry in entries { - let value = apache_avro::types::Value::try_from(entry) - .unwrap() - .resolve(&avro_schema) + writer + .append_value(entry.resolve(&avro_schema).unwrap()) .unwrap(); - writer.append_value(value).unwrap(); } writer.into_inner().unwrap() } @@ -1631,6 +1646,91 @@ mod tests { assert_eq!(manifest, Manifest::new(metadata, vec![expected])); } + #[test] + fn test_parse_manifest_reads_union_writer_fields_as_required() { + // A field that is required in the reader schema is written as a union with + // null. + let metadata = writer_schema_test_metadata(); + let mut writer_schema = v2_writer_schema(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": 1002}, + ])); + writer_schema_field(&mut writer_schema, &["data_file", "record_count"])["type"] = + serde_json::json!(["null", "long"]); + let partition = serde_json::json!({"id": 5, "name": "a", "score": 2.5}); + let bs = + write_with_writer_schema(&metadata, &writer_schema, vec![v2_entry(partition.clone())]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let expected = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + assert_eq!(manifest, Manifest::new(metadata.clone(), vec![expected])); + + let mut entry = v2_entry(partition); + entry["data_file"]["record_count"] = Value::Null; + let bs = write_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let err = Manifest::parse_avro(&bs).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + #[test] + fn test_parse_manifest_with_fixed_uuid_partition() { + // The spec stores a uuid in Avro as a 16-byte fixed with logical type uuid. + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "u", Type::Primitive(PrimitiveType::Uuid)).into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .add_partition_field("u", "u", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let metadata = ManifestMetadata { + schema_id: 0, + schema, + partition_spec, + content: ManifestContentType::Data, + format_version: FormatVersion::V2, + }; + let writer_schema = v2_writer_schema(serde_json::json!([ + {"name": "u", "default": null, "field-id": 1000, "type": ["null", { + "type": "fixed", "name": "uuid_fixed", "size": 16, "logicalType": "uuid", + }]}, + ])); + let uuid = uuid::Uuid::from_u128(0xf79c3e09_677c_4bbd_a479_3f349cb785e7); + // `Value::resolve` doesn't convert a JSON string to a fixed uuid, so the + // partition value is set as an Avro value. + let mut entry = AvroValue::try_from(v2_entry(serde_json::json!({}))).unwrap(); + let AvroValue::Map(fields) = &mut entry else { + unreachable!("a JSON object converts to an Avro map"); + }; + let Some(AvroValue::Map(data_file)) = fields.get_mut("data_file") else { + unreachable!("a JSON object converts to an Avro map"); + }; + data_file.insert( + "partition".to_string(), + AvroValue::Map(HashMap::from([("u".to_string(), AvroValue::Uuid(uuid))])), + ); + let bs = write_avro_values_with_writer_schema(&metadata, &writer_schema, vec![entry]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let expected = expected_entry(Struct::from_iter([Some(Literal::uuid(uuid))])); + assert_eq!(manifest, Manifest::new(metadata, vec![expected])); + } + #[test] fn test_parse_manifest_ignores_record_names_and_unknown_fields() { // Record names differ from the ones Java writes, and the writer has fields From 91486882b29deb0d97560543e0764aa00a2988fd Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 6 Oct 2026 09:42:58 -0700 Subject: [PATCH 3/5] address PR feedback --- crates/iceberg/src/avro/deserializer.rs | 177 ++++++++++++ crates/iceberg/src/avro/mod.rs | 2 + crates/iceberg/src/avro/named_types.rs | 60 +++- crates/iceberg/src/avro/schema.rs | 69 +++++ crates/iceberg/src/spec/manifest/_serde.rs | 57 +++- crates/iceberg/src/spec/manifest/mod.rs | 268 +++++++++++++++++- crates/iceberg/src/spec/manifest/writer.rs | 38 ++- crates/iceberg/src/spec/values/serde.rs | 4 + crates/iceberg/src/spec/values/tests.rs | 4 +- crates/iceberg/testdata/manifests/README.md | 62 ++++ .../testdata/manifests/pyiceberg-v2-data.avro | Bin 0 -> 4787 bytes 11 files changed, 723 insertions(+), 18 deletions(-) create mode 100644 crates/iceberg/testdata/manifests/pyiceberg-v2-data.avro diff --git a/crates/iceberg/src/avro/deserializer.rs b/crates/iceberg/src/avro/deserializer.rs index 79f4fb623d..e130925443 100644 --- a/crates/iceberg/src/avro/deserializer.rs +++ b/crates/iceberg/src/avro/deserializer.rs @@ -278,3 +278,180 @@ impl<'de, S: DeserializeSeed<'de>> DeserializeSeed<'de> for ResolvingSeed { self.0.deserialize(ResolvingDeserializer(deserializer)) } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::fmt::Debug; + + use apache_avro::types::Value as AvroValue; + use apache_avro::{Reader, Schema as AvroSchema, Writer}; + use serde::de::DeserializeOwned; + use serde_bytes::ByteBuf; + use serde_derive::Deserialize; + + use super::*; + + #[derive(Deserialize, Debug, PartialEq)] + struct Pair { + a: i64, + b: Option, + } + + #[derive(Deserialize, Debug, PartialEq)] + enum Suit { + Hearts, + Spades, + } + + /// Writes `value` with `writer_schema` and reads it back as a `T`. + fn read( + writer_schema: &str, + value: AvroValue, + ) -> apache_avro::AvroResult { + let schema = AvroSchema::parse_str(writer_schema).unwrap(); + let mut writer = Writer::new(&schema, Vec::new()).unwrap(); + writer.append_value(value).unwrap(); + let bs = writer.into_inner().unwrap(); + let mut values = Reader::new(bs.as_slice()) + .unwrap() + .into_deser_iter::>(); + values.next().unwrap().map(|value| value.0) + } + + fn check( + writer_schema: &str, + value: AvroValue, + expected: T, + ) { + assert_eq!( + read::(writer_schema, value).unwrap(), + expected, + "{writer_schema}" + ); + } + + const PAIR: &str = r#"{"type": "record", "name": "writer_name", "fields": [ + {"name": "b", "type": "string"}, + {"name": "extra", "type": "int"}, + {"name": "a", "type": "int"} + ]}"#; + + fn pair_value() -> AvroValue { + AvroValue::Record(vec![ + ("b".to_string(), AvroValue::String("x".to_string())), + ("extra".to_string(), AvroValue::Int(9)), + ("a".to_string(), AvroValue::Int(1)), + ]) + } + + fn pair() -> Pair { + Pair { + a: 1, + b: Some("x".to_string()), + } + } + + const SUIT: &str = r#"{"type": "enum", "name": "suit", "symbols": ["Hearts", "Spades"]}"#; + + #[test] + fn test_resolved_reads_writer_values() { + check(r#""boolean""#, AvroValue::Boolean(true), true); + check(r#""int""#, AvroValue::Int(-3), -3i64); + check(r#""long""#, AvroValue::Long(7), 7i32); + check(r#""float""#, AvroValue::Float(1.5), 1.5f64); + check(r#""double""#, AvroValue::Double(2.5), 2.5f64); + check( + r#""string""#, + AvroValue::String("s".to_string()), + "s".to_string(), + ); + check( + r#""bytes""#, + AvroValue::Bytes(vec![1, 2]), + ByteBuf::from(vec![1, 2]), + ); + check( + r#"{"type": "fixed", "name": "f", "size": 2}"#, + AvroValue::Fixed(2, vec![3, 4]), + ByteBuf::from(vec![3, 4]), + ); + check( + r#"{"type": "array", "items": "int"}"#, + AvroValue::Array(vec![AvroValue::Int(1), AvroValue::Int(2)]), + vec![1i64, 2], + ); + check( + r#"{"type": "map", "values": "long"}"#, + AvroValue::Map(HashMap::from([("k".to_string(), AvroValue::Long(5))])), + HashMap::from([("k".to_string(), 5i32)]), + ); + check(PAIR, pair_value(), pair()); + check(SUIT, AvroValue::Enum(1, "Spades".to_string()), Suit::Spades); + check( + r#"["null", "long"]"#, + AvroValue::Union(1, Box::new(AvroValue::Long(5))), + 5i64, + ); + } + + #[test] + fn test_resolved_reads_any_writer_value_as_option() { + check(r#""null""#, AvroValue::Null, None::); + check( + r#"["null", "long"]"#, + AvroValue::Union(0, Box::new(AvroValue::Null)), + None::, + ); + check( + r#"["null", "int"]"#, + AvroValue::Union(1, Box::new(AvroValue::Int(5))), + Some(5i64), + ); + check(r#""boolean""#, AvroValue::Boolean(true), Some(true)); + check(r#""int""#, AvroValue::Int(-3), Some(-3i64)); + check(r#""long""#, AvroValue::Long(7), Some(7i32)); + check(r#""float""#, AvroValue::Float(1.5), Some(1.5f64)); + check(r#""double""#, AvroValue::Double(2.5), Some(2.5f64)); + check( + r#""string""#, + AvroValue::String("s".to_string()), + Some("s".to_string()), + ); + check( + r#""bytes""#, + AvroValue::Bytes(vec![1, 2]), + Some(ByteBuf::from(vec![1, 2])), + ); + check( + r#"{"type": "array", "items": "int"}"#, + AvroValue::Array(vec![AvroValue::Int(1)]), + Some(vec![1i64]), + ); + check( + r#"{"type": "map", "values": "long"}"#, + AvroValue::Map(HashMap::from([("k".to_string(), AvroValue::Long(5))])), + Some(HashMap::from([("k".to_string(), 5i32)])), + ); + check(PAIR, pair_value(), Some(pair())); + check( + SUIT, + AvroValue::Enum(0, "Hearts".to_string()), + Some(Suit::Hearts), + ); + } + + #[test] + fn test_resolved_rejects_values_that_do_not_fit() { + assert!(read::(r#""long""#, AvroValue::Long(i64::from(i32::MAX) + 1)).is_err()); + assert!(read::(r#""int""#, AvroValue::Int(-1)).is_err()); + assert!( + read::( + r#"["null", "long"]"#, + AvroValue::Union(0, Box::new(AvroValue::Null)) + ) + .is_err() + ); + assert!(read::(r#""long""#, AvroValue::Long(1)).is_err()); + } +} diff --git a/crates/iceberg/src/avro/mod.rs b/crates/iceberg/src/avro/mod.rs index 5a065dc4a9..0bc5a70041 100644 --- a/crates/iceberg/src/avro/mod.rs +++ b/crates/iceberg/src/avro/mod.rs @@ -21,4 +21,6 @@ mod named_types; mod schema; pub(crate) use deserializer::Resolved; pub(crate) use named_types::define_named_types_once; +#[cfg(test)] +pub(crate) use named_types::define_named_types_repeatedly; pub(crate) use schema::*; diff --git a/crates/iceberg/src/avro/named_types.rs b/crates/iceberg/src/avro/named_types.rs index 9a4eb033fa..a006a84156 100644 --- a/crates/iceberg/src/avro/named_types.rs +++ b/crates/iceberg/src/avro/named_types.rs @@ -45,6 +45,16 @@ const SCHEMA_KEY: &str = "avro.schema"; /// file and the full names of the repeated types, or `None` if a repeated /// definition differs from the first one, which leaves the file ambiguous. pub(crate) fn define_named_types_once(bs: &[u8]) -> Result, Vec)>> { + rewrite_schema(bs, reference_repeated_definitions) +} + +/// Rewrites the schema in the header of an Avro object container file with +/// `rewrite`, leaving the data blocks as they are. Returns `None` if `rewrite` +/// does. +fn rewrite_schema( + bs: &[u8], + rewrite: impl FnOnce(&mut JsonValue) -> Option, +) -> Result, T)>> { let body = bs .strip_prefix(MAGIC) .ok_or_else(|| invalid_data!("Not an Avro object container file"))?; @@ -60,13 +70,16 @@ pub(crate) fn define_named_types_once(bs: &[u8]) -> Result, Vec< else { return Err(invalid_data!("Avro file metadata is not a map")); }; - let rest = &body[cursor.position() as usize..]; + let rest = usize::try_from(cursor.position()) + .ok() + .and_then(|position| body.get(position..)) + .ok_or_else(|| invalid_data!("Avro file metadata extends past the end of the file"))?; let Some(AvroValue::Bytes(schema)) = metadata.get(SCHEMA_KEY) else { return Err(invalid_data!("Avro file metadata has no {SCHEMA_KEY}")); }; let mut schema: JsonValue = serde_json::from_slice(schema)?; - let Some(repeated) = reference_repeated_definitions(&mut schema) else { + let Some(rewritten) = rewrite(&mut schema) else { return Ok(None); }; metadata.insert( @@ -77,7 +90,48 @@ pub(crate) fn define_named_types_once(bs: &[u8]) -> Result, Vec< .build()? .write_value_to_vec(AvroValue::Map(metadata))?; - Ok(Some(([MAGIC, &metadata, rest].concat(), repeated))) + Ok(Some(([MAGIC, &metadata, rest].concat(), rewritten))) +} + +/// Rewrites the header of an Avro object container file so that each reference +/// to a named type repeats its definition, as iceberg-rust wrote manifests with +/// `apache-avro` 0.21. +#[cfg(test)] +pub(crate) fn define_named_types_repeatedly(bs: &[u8]) -> Vec { + fn inline_references(schema: &mut JsonValue, defined: &mut HashMap) { + match schema { + JsonValue::String(name) => { + if let Some(definition) = defined.get(name.as_str()) { + *schema = definition.clone(); + } + } + JsonValue::Array(schemas) => schemas + .iter_mut() + .for_each(|schema| inline_references(schema, defined)), + JsonValue::Object(object) => { + if let (Some(JsonValue::String(name)), Some("record" | "enum" | "fixed")) = ( + object.get("name"), + object.get("type").and_then(JsonValue::as_str), + ) { + defined.insert(name.clone(), JsonValue::Object(object.clone())); + } + for key in ["type", "items", "values", "fields"] { + if let Some(child) = object.get_mut(key) { + inline_references(child, defined); + } + } + } + _ => {} + } + } + + rewrite_schema(bs, |schema| { + inline_references(schema, &mut HashMap::new()); + Some(()) + }) + .unwrap() + .unwrap() + .0 } /// Replaces each repeated definition of a named type in `schema` with its full diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index 96999ebedc..f7ba256dd5 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/crates/iceberg/src/avro/schema.rs @@ -988,6 +988,75 @@ mod tests { check_schema_conversion(avro_schema, iceberg_schema); } + #[test] + fn test_schema_to_avro_schema_defines_named_types_before_references() { + let decimal = Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }); + let iceberg_schema = Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "a", decimal.clone()).into(), + NestedField::required( + 2, + "list", + Type::List(ListType::new( + NestedField::list_element(3, decimal.clone(), false).into(), + )), + ) + .into(), + NestedField::required( + 4, + "map", + Type::Map(MapType { + key_field: NestedField::map_key_element(5, PrimitiveType::Int.into()) + .into(), + value_field: NestedField::map_value_element(6, decimal, false).into(), + }), + ) + .into(), + ]) + .build() + .unwrap(); + + let json = schema_to_avro_schema("r", &iceberg_schema) + .unwrap() + .canonical_form(); + + // Parsing fails on a reference that precedes its definition. + AvroSchema::parse_str(&json).unwrap(); + assert_eq!(json.matches(r#""name":"decimal_10_2""#).count(), 1); + } + + #[test] + fn test_schema_with_array_map_without_parsing() { + // Unlike `test_schema_with_array_map`, this converts an Avro schema built + // in code, which keeps `logicalType: map` in apache-avro 0.22. + let map = |key_id, value_id, value_required| { + Type::Map(MapType { + key_field: NestedField::map_key_element(key_id, PrimitiveType::Boolean.into()) + .into(), + value_field: NestedField::map_value_element( + value_id, + PrimitiveType::Boolean.into(), + value_required, + ) + .into(), + }) + }; + let iceberg_schema = Schema::builder() + .with_fields(vec![ + NestedField::required(100, "optional_values", map(102, 103, false)).into(), + NestedField::optional(104, "required_values", map(105, 106, true)).into(), + ]) + .build() + .unwrap(); + + let avro_schema = schema_to_avro_schema("avro_schema", &iceberg_schema).unwrap(); + + assert_eq!(avro_schema_to_schema(&avro_schema).unwrap(), iceberg_schema); + } + #[test] #[ignore = "apache-avro 0.22 drops `logicalType: map` when parsing schema JSON \ (https://github.com/apache/avro-rs/issues/654), so avro_schema_to_schema \ diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs index fb7139e498..860a524261 100644 --- a/crates/iceberg/src/spec/manifest/_serde.rs +++ b/crates/iceberg/src/spec/manifest/_serde.rs @@ -15,6 +15,14 @@ // specific language governing permissions and limitations // under the License. +//! Serde forms of manifest entries. +//! +//! `Manifest::parse_avro` deserializes these structs from the writer schema +//! without a reader schema, so they decide which manifests read. A field that +//! a writer may omit, per the spec's read rules for every format version the +//! struct reads, must be an `Option` or have `#[serde(default)]`. +//! `test_parse_manifest_without_each_field` checks each field. + use std::collections::HashMap; use serde_derive::{Deserialize, Serialize}; @@ -327,9 +335,10 @@ mod tests { use std::io::Cursor; use std::sync::Arc; + use crate::avro::define_named_types_repeatedly; use crate::spec::manifest::_serde::{I64Entry, parse_i64_entry}; use crate::spec::{ - DataContentType, DataFile, DataFileFormat, Datum, FormatVersion, NestedField, + DataContentType, DataFile, DataFileFormat, Datum, FormatVersion, Literal, NestedField, PrimitiveType, Schema, Struct, StructType, Type, read_data_files_from_avro, write_data_files_to_avro, }; @@ -424,6 +433,52 @@ mod tests { assert_eq!(data_files, actual_data_file); } + /// apache-avro 0.22 checks the writer schema for repeated names only when + /// the reader has no reader schema, so this path reads 0.21 output without + /// the header rewrite that `Manifest::parse_avro` needs. + #[test] + fn test_read_data_files_with_repeated_named_type_definitions() { + let decimal = Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }); + let schema = Schema::builder() + .with_fields(schema().as_struct().fields().to_vec()) + .with_fields(vec![ + NestedField::optional(4, "a", decimal.clone()).into(), + NestedField::optional(5, "b", decimal.clone()).into(), + ]) + .build() + .unwrap(); + let partition_type = StructType::new(vec![ + NestedField::optional(1000, "a", decimal.clone()).into(), + NestedField::optional(1001, "b", decimal).into(), + ]); + let mut data_files = data_files(); + data_files[0].partition = + Struct::from_iter([Some(Literal::decimal(12345)), Some(Literal::decimal(-678))]); + let mut buffer = Vec::new(); + write_data_files_to_avro( + &mut buffer, + data_files.clone(), + &partition_type, + FormatVersion::V2, + ) + .unwrap(); + let legacy = define_named_types_repeatedly(&buffer); + + let actual = read_data_files_from_avro( + &mut Cursor::new(legacy), + &schema, + 0, + &partition_type, + FormatVersion::V2, + ) + .unwrap(); + + assert_eq!(actual, data_files); + } + #[tokio::test] async fn test_data_file_serialize_deserialize_v1_data_on_v2_reader() { let schema = schema(); diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index fd4521156f..e49dc26200 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -27,6 +27,7 @@ mod reader; pub use reader::*; mod writer; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use apache_avro::Reader as AvroReader; use apache_avro::error::Details; @@ -39,6 +40,10 @@ use super::{ use crate::avro::{Resolved, define_named_types_once}; use crate::error::{Result, invalid_data}; +/// Whether a manifest with repeated Avro named type definitions was logged at +/// warn level. A table written by an affected release can have many of them. +static WARNED_REPEATED_DEFINITIONS: AtomicBool = AtomicBool::new(false); + /// A manifest contains metadata and a list of entries. #[derive(Debug, PartialEq, Eq, Clone)] pub struct Manifest { @@ -56,16 +61,26 @@ impl Manifest { let rewritten; let reader = match AvroReader::new(bs) { Ok(reader) => reader, + // iceberg-rust repeated `decimal` definitions before + // `schema_to_avro_schema` defined each named type once, so this + // fallback stays while tables can contain manifests it wrote. Err(e) if matches!(e.details(), Details::AmbiguousSchemaDefinition(_)) => { - let Some((bs, repeated)) = define_named_types_once(bs)? else { + let Ok(Some((bs, repeated))) = define_named_types_once(bs) else { return Err(e.into()); }; - tracing::warn!( - "Manifest {} defines Avro named types {repeated:?} more than once, which the \ - Avro specification doesn't allow. Reading it with each repeated definition \ - replaced by a reference to the first.", - location.unwrap_or("") - ); + let location = location.unwrap_or(""); + if WARNED_REPEATED_DEFINITIONS.swap(true, Ordering::Relaxed) { + tracing::debug!( + "Manifest {location} defines Avro named types {repeated:?} more than once." + ); + } else { + tracing::warn!( + "Manifest {location} defines Avro named types {repeated:?} more than once, \ + which the Avro specification doesn't allow. Reading it with each repeated \ + definition replaced by a reference to the first. Later manifests like \ + this are logged at debug level." + ); + } rewritten = bs; AvroReader::new(rewritten.as_slice())? } @@ -1561,6 +1576,157 @@ mod tests { writer.into_inner().unwrap() } + /// Removes the field named by `path` from a record schema and from a record + /// value of that schema. + fn remove_writer_field(schema: &mut Value, entry: &mut AvroValue, path: &[&str]) { + let (name, rest) = path.split_first().unwrap(); + let AvroValue::Record(values) = entry else { + unreachable!("the entry is a record"); + }; + let fields = schema["fields"].as_array_mut().unwrap(); + if rest.is_empty() { + fields.retain(|field| field["name"] != *name); + values.retain(|(value_name, _)| value_name != name); + } else { + let field = fields.iter_mut().find(|field| field["name"] == *name); + let value = values.iter_mut().find(|(value_name, _)| value_name == name); + remove_writer_field(&mut field.unwrap()["type"], &mut value.unwrap().1, rest); + } + } + + #[test] + fn test_parse_manifest_without_each_field() { + // Fields that the spec doesn't require in every version must read as + // their default when the writer omits them. + let metadata = writer_schema_test_metadata(); + let partition_type = metadata + .partition_spec + .partition_type(&metadata.schema) + .unwrap(); + let full_schema = + serde_json::to_value(manifest_schema_v2(&partition_type).unwrap()).unwrap(); + let mut full_entry = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + full_entry.sequence_number = Some(3); + full_entry.file_sequence_number = Some(4); + let data_file = &mut full_entry.data_file; + data_file.content = DataContentType::PositionDeletes; + data_file.column_sizes = HashMap::from([(1, 40)]); + data_file.value_counts = HashMap::from([(1, 10)]); + data_file.null_value_counts = HashMap::from([(1, 1)]); + data_file.nan_value_counts = HashMap::from([(3, 2)]); + data_file.lower_bounds = HashMap::from([(1, Datum::long(1))]); + data_file.upper_bounds = HashMap::from([(1, Datum::long(9))]); + data_file.key_metadata = Some(vec![1, 2]); + data_file.split_offsets = Some(vec![4]); + data_file.equality_ids = Some(vec![1]); + data_file.sort_order_id = Some(0); + data_file.first_row_id = Some(100); + data_file.referenced_data_file = Some("s3://bucket/table/data/b.parquet".to_string()); + data_file.content_offset = Some(4); + data_file.content_size_in_bytes = Some(8); + let full_value = to_value( + _serde::ManifestEntryV2::try_from( + full_entry.clone(), + &Type::Struct(partition_type.clone()), + ) + .unwrap(), + ) + .unwrap(); + + type ResetField = fn(&mut ManifestEntry); + let optional: [(&[&str], ResetField); 18] = [ + (&["snapshot_id"], |e| e.snapshot_id = None), + (&["sequence_number"], |e| e.sequence_number = None), + (&["file_sequence_number"], |e| e.file_sequence_number = None), + (&["data_file", "content"], |e| { + e.data_file.content = DataContentType::Data + }), + (&["data_file", "column_sizes"], |e| { + e.data_file.column_sizes.clear() + }), + (&["data_file", "value_counts"], |e| { + e.data_file.value_counts.clear() + }), + (&["data_file", "null_value_counts"], |e| { + e.data_file.null_value_counts.clear() + }), + (&["data_file", "nan_value_counts"], |e| { + e.data_file.nan_value_counts.clear() + }), + (&["data_file", "lower_bounds"], |e| { + e.data_file.lower_bounds.clear() + }), + (&["data_file", "upper_bounds"], |e| { + e.data_file.upper_bounds.clear() + }), + (&["data_file", "key_metadata"], |e| { + e.data_file.key_metadata = None + }), + (&["data_file", "split_offsets"], |e| { + e.data_file.split_offsets = None + }), + (&["data_file", "equality_ids"], |e| { + e.data_file.equality_ids = None + }), + (&["data_file", "sort_order_id"], |e| { + e.data_file.sort_order_id = None + }), + (&["data_file", "first_row_id"], |e| { + e.data_file.first_row_id = None + }), + (&["data_file", "referenced_data_file"], |e| { + e.data_file.referenced_data_file = None + }), + (&["data_file", "content_offset"], |e| { + e.data_file.content_offset = None + }), + (&["data_file", "content_size_in_bytes"], |e| { + e.data_file.content_size_in_bytes = None + }), + ]; + for (path, reset) in optional { + let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); + remove_writer_field(&mut schema, &mut value, path); + let bs = write_avro_values_with_writer_schema(&metadata, &schema, vec![value]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let mut expected = full_entry.clone(); + reset(&mut expected); + assert_eq!( + manifest, + Manifest::new(metadata.clone(), vec![expected]), + "{path:?}" + ); + } + + let required: [&[&str]; 7] = [ + &["status"], + &["data_file"], + &["data_file", "file_path"], + &["data_file", "file_format"], + &["data_file", "partition"], + &["data_file", "record_count"], + &["data_file", "file_size_in_bytes"], + ]; + for path in required { + let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); + remove_writer_field(&mut schema, &mut value, path); + let bs = write_avro_values_with_writer_schema(&metadata, &schema, vec![value]); + + let err = Manifest::parse_avro(&bs).unwrap_err(); + + assert!( + err.to_string().contains(path.last().unwrap()), + "{path:?}: {err}" + ); + } + } + #[test] fn test_parse_manifest_matches_partition_fields_by_name() { // The writer orders the partition fields differently from the spec, omits @@ -1847,6 +2013,94 @@ mod tests { assert_eq!(err.kind(), ErrorKind::DataInvalid); } + #[test] + fn test_parse_manifest_written_by_pyiceberg() { + let bs = fs::read(format!( + "{}/testdata/manifests/pyiceberg-v2-data.avro", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "category", Type::Primitive(PrimitiveType::String)) + .into(), + NestedField::optional(3, "score", Type::Primitive(PrimitiveType::Double)) + .into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .add_partition_field("category", "category", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let metadata = ManifestMetadata { + schema_id: 0, + schema, + partition_spec, + content: ManifestContentType::Data, + format_version: FormatVersion::V2, + }; + let entries = (0..2) + .map(|i: i64| { + let category = format!("c{i}"); + ManifestEntry { + status: ManifestStatus::Added, + snapshot_id: Some(7), + sequence_number: Some(1), + file_sequence_number: Some(1), + data_file: DataFile { + content: DataContentType::Data, + file_path: format!( + "s3://bucket/table/data/category={category}/0000{i}.parquet" + ), + file_format: DataFileFormat::Parquet, + partition: Struct::from_iter([Some(Literal::string(&category))]), + record_count: 100 + i as u64, + file_size_in_bytes: 1000 + i as u64, + column_sizes: HashMap::from([ + (1, 10 + i as u64), + (2, 20 + i as u64), + (3, 30 + i as u64), + ]), + value_counts: HashMap::from([ + (1, 100 + i as u64), + (2, 100 + i as u64), + (3, 100 + i as u64), + ]), + null_value_counts: HashMap::from([(1, 0), (2, i as u64), (3, 1)]), + nan_value_counts: HashMap::from([(3, i as u64)]), + lower_bounds: HashMap::from([ + (1, Datum::long(i)), + (2, Datum::string(&category)), + ]), + upper_bounds: HashMap::from([ + (1, Datum::long(i + 50)), + (2, Datum::string(&category)), + ]), + key_metadata: None, + split_offsets: Some(vec![4]), + equality_ids: None, + sort_order_id: Some(0), + partition_spec_id: 0, + first_row_id: None, + referenced_data_file: None, + content_offset: None, + content_size_in_bytes: None, + }, + } + }) + .collect(); + assert_eq!(manifest, Manifest::new(metadata, entries)); + } + #[test] fn test_parse_manifest_with_repeated_named_type_definitions() { let bs = fs::read(format!( diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index 52a11a6e5b..a79312dfab 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -580,6 +580,7 @@ mod tests { use tempfile::TempDir; use super::*; + use crate::avro::define_named_types_repeatedly; use crate::io::FileIO; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, Literal, Manifest, NestedField, @@ -823,9 +824,12 @@ mod tests { ); } - /// Writes a manifest partitioned by identity on two columns of `field_type` - /// and reads it back. - async fn roundtrip_two_partition_fields_of_type(field_type: Type, values: [Literal; 2]) { + /// Writes a manifest partitioned by identity on two columns of `field_type`, + /// reads it back, and returns the manifest file. + async fn roundtrip_two_partition_fields_of_type( + field_type: Type, + values: [Literal; 2], + ) -> Vec { let schema = Arc::new( Schema::builder() .with_fields(vec![ @@ -867,10 +871,34 @@ mod tests { ) .unwrap(); writer.write_manifest_file().await.unwrap(); + let bs = fs::read(&path).unwrap(); - let manifest = Manifest::parse_avro(&fs::read(&path).unwrap()).unwrap(); + // Also read the manifest as iceberg-rust wrote it before defining each + // named type once. + for bs in [bs.clone(), define_named_types_repeatedly(&bs)] { + let manifest = Manifest::parse_avro(&bs).unwrap(); - assert_eq!(*manifest.entries()[0].data_file().partition(), partition); + assert_eq!(*manifest.entries()[0].data_file().partition(), partition); + } + bs + } + + #[tokio::test] + async fn test_write_manifest_header_marks_int_keyed_maps() { + // Java and PyIceberg read an array of key-value records as a map only if + // it has `logicalType: map`. apache-avro 0.22 drops the attribute when it + // parses a schema (https://github.com/apache/avro-rs/issues/654), so this + // checks the schema JSON in the written header. + let bs = roundtrip_two_partition_fields_of_type(Type::Primitive(PrimitiveType::Int), [ + Literal::int(1), + Literal::int(2), + ]) + .await; + + let file = String::from_utf8_lossy(&bs); + // `column_sizes`, `value_counts`, `null_value_counts`, + // `nan_value_counts`, `lower_bounds`, and `upper_bounds`. + assert_eq!(file.matches(r#""logicalType":"map""#).count(), 6); } #[tokio::test] diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs index 27726a050c..a9c6115171 100644 --- a/crates/iceberg/src/spec/values/serde.rs +++ b/crates/iceberg/src/spec/values/serde.rs @@ -52,6 +52,10 @@ pub(crate) mod _serde { /// `struct_type`'s fields in its order, with null for an optional field the /// 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. 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/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index 89e84741e5..c756bc3f8e 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -1724,7 +1724,7 @@ fn test_datum_to_decimal_rejects_scale_change() { } #[test] -fn raw_literal_project_by_name_reorders_and_fills_fields() { +fn test_raw_literal_project_by_name_reorders_and_fills_fields() { let written = StructType::new(vec![ NestedField::required(1, "a", Primitive(PrimitiveType::Int)).into(), NestedField::optional(2, "b", Primitive(PrimitiveType::Int)).into(), @@ -1762,7 +1762,7 @@ fn raw_literal_project_by_name_reorders_and_fills_fields() { } #[test] -fn raw_literal_project_by_name_rejects_missing_required_field() { +fn test_raw_literal_project_by_name_rejects_missing_required_field() { let written = StructType::new(vec![ NestedField::optional(1, "a", Primitive(PrimitiveType::Int)).into(), ]); diff --git a/crates/iceberg/testdata/manifests/README.md b/crates/iceberg/testdata/manifests/README.md index 657243417c..6b68b71ed0 100644 --- a/crates/iceberg/testdata/manifests/README.md +++ b/crates/iceberg/testdata/manifests/README.md @@ -35,3 +35,65 @@ still read. To regenerate it, check out a commit that uses `apache-avro` 0.21, write the manifest described above with `ManifestWriterBuilder::build_v2_data`, and copy the file here. + +`pyiceberg-v2-data.avro` is a V2 data manifest written by PyIceberg 0.12.0 +(apache/iceberg-python commit `0d584073`) with deflate compression. Its table +schema has optional columns `id` (long), `category` (string), and `score` +(double), and its partition spec has an identity partition on `category`. It +holds two entries with column metrics. Its Avro schema is PyIceberg's, so tests +use it to check reading a manifest whose writer schema iceberg-rust didn't +generate. + +To regenerate it, run this script with PyIceberg installed, passing the output +path: + +```python +import sys + +from pyiceberg.io.pyarrow import PyArrowFileIO +from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus, write_manifest +from pyiceberg.partitioning import PartitionField, PartitionSpec +from pyiceberg.schema import Schema +from pyiceberg.transforms import IdentityTransform +from pyiceberg.typedef import Record +from pyiceberg.types import DoubleType, LongType, NestedField, StringType + +SCHEMA = Schema( + NestedField(1, "id", LongType(), required=False), + NestedField(2, "category", StringType(), required=False), + NestedField(3, "score", DoubleType(), required=False), +) +SPEC = PartitionSpec(PartitionField(source_id=2, field_id=1000, transform=IdentityTransform(), name="category"), spec_id=0) + + +def entry(i: int) -> ManifestEntry: + data_file = DataFile.from_args( + content=DataFileContent.DATA, + file_path=f"s3://bucket/table/data/category=c{i}/{i:05d}.parquet", + file_format=FileFormat.PARQUET, + partition=Record(f"c{i}"), + record_count=100 + i, + file_size_in_bytes=1000 + i, + column_sizes={1: 10 + i, 2: 20 + i, 3: 30 + i}, + value_counts={1: 100 + i, 2: 100 + i, 3: 100 + i}, + null_value_counts={1: 0, 2: i, 3: 1}, + nan_value_counts={3: i}, + lower_bounds={1: (i).to_bytes(8, "little"), 2: f"c{i}".encode()}, + upper_bounds={1: (i + 50).to_bytes(8, "little"), 2: f"c{i}".encode()}, + split_offsets=[4], + sort_order_id=0, + ) + return ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, snapshot_id=7, sequence_number=1, file_sequence_number=1, data_file=data_file + ) + + +def main(path: str) -> None: + with write_manifest(2, SPEC, SCHEMA, PyArrowFileIO().new_output(path), snapshot_id=7, avro_compression="deflate") as writer: + for i in range(2): + writer.add_entry(entry(i)) + + +if __name__ == "__main__": + main(sys.argv[1]) +``` diff --git a/crates/iceberg/testdata/manifests/pyiceberg-v2-data.avro b/crates/iceberg/testdata/manifests/pyiceberg-v2-data.avro new file mode 100644 index 0000000000000000000000000000000000000000..c8c6866449aade34c2f7e53e0f76e521d658f991 GIT binary patch literal 4787 zcmb_gU1%It6y9W05c`8jksw8`(>|!3)ScOyY=Yo}O*GP0u@Qu_jCW@4He+XJrg!e7 zjb)+LQmj^?wGVj^6zPLc;)_1F(qdi$B7G?8i&_xGzKNh1^qhNlW_NbCYqBncke!`- z&Ue0Z&Ueqri}vdyJGRBnBCFHyil;TP95HQ1iv>>{q3K%9W4;?}Ge=Kro~z9mx)#tn zz5$=~^ld*3j=?9+P9z>@@LG-fF>C1RjG3Krs9?uJzT7$qe!PBX%h}o32{~&Wbi>5< z@pGrss;;SMCJ=8)2cRqv9>l>XWH9g?_A;$*#g5fSD$~&}ImJl{X2MzygO^K~lFF z?=V{))NIZ3nHBf7y?;>K$hd_LI>4GOgBwjjGl?CBjB*<`nFZu9bbvo?4*U{T9|RLQ zLooL`3Auy62$T~L+L1TOMnuJ8rz0rUR=U#;%X}RSp-`a!$ty3hjQ58K@PqhcgcYMU!<*m#;xGkO@t`p0SHFYFZ`TjG#wRlD>@E$@eA zXOXZ*l!cgh${D6D??NUr_TFHY7g+YPV4d<{l-f&Esd*OWLMW(D&;~&p=tkMdb-WY$ zNj3sT}ep>L;@4(<5p8?U^ttjVh+z507~c^%vwVNwhkRv_=)4=d}CIDVmWEBG z>>V8n)A17*q)6UJga$4-Ob_R|#hc3NN3hXLAX){v$yzlHB-k9BVAB|YU~_l`BVR=F z1vQKcm4g$|pf8!_sbN-YxdLRCrw1pqT*;x|P~(@Uhe;+5L#PK(mx|R)wUt8T>Gi{t zj9WJF4ihUd*)B-tQ^TazoWwev%A9Ok2+Ya;)b_k}%&FmVE4d<)t2~KG6y2M|p;@(|@za@C7OxKyn=6BGxXtVo(WLb+`g%&sI&VXZ=u-s@_d4II84 z$w?HD_Ek7?1rIw3g*TRAhw4VS!B*`+IN6R+E^YnToWngG28@^_IC#RC;WEK^-3u7u z;YpGL#PKcr+Xa^b77v*1;(Z;Qha4#9F7!_KTDZU0B`H^$eh}vmz^EE=7USX!w{#3H zV7UPoxcb1L?8OpF9tdYO#&MT)t$VF##{RZBIjezmFX;zq4z9OlB6A1g>JmTQjUMY> zjvhzhrfSxg)38{if>5|06^$`O*MzEO!gOTiurz8j_{cvIZ%8knh*6-Z7 zeCzkWKY8JMx7a8?@Xg-m^(XX06F+W^w%xxi*-q{G$;$OpXFs~|-u$`Qx7N;&tX8hC z@;5Kt9ozl(+O@9=dsbF%kB_gsx-vHY@tG~3Zr5KsrxzB!9@#TKUYK2dxC_r88^`l( zqex1(dH{O*=BWz^=;G|VYwrV~o2&etOLuo7piyn*+St|)3cHJApBKJ3yXE4gEoXkM OoEa+?3kNTBq4^i@F9Rk3 literal 0 HcmV?d00001 From fefcbd6d17a262ce240a6b3ae0ee99164d8f6fed Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Wed, 7 Oct 2026 09:35:21 -0700 Subject: [PATCH 4/5] address PR feedback. --- crates/iceberg/src/avro/schema.rs | 2 + crates/iceberg/src/spec/manifest/_serde.rs | 3 +- crates/iceberg/src/spec/manifest/mod.rs | 197 +++++++++++++++--- .../iceberg/src/spec/manifest_list/writer.rs | 5 +- 4 files changed, 175 insertions(+), 32 deletions(-) diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index f7ba256dd5..47254317ed 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/crates/iceberg/src/avro/schema.rs @@ -271,6 +271,8 @@ impl SchemaVisitor for SchemaToAvroSchema { PrimitiveType::TimestampNs => AvroSchema::TimestampNanos, PrimitiveType::TimestamptzNs => AvroSchema::TimestampNanos, PrimitiveType::String => AvroSchema::String, + // TODO(#2913): write the spec's `fixed[16]` uuid. The manifest reader + // already accepts both forms. PrimitiveType::Uuid => AvroSchema::Uuid(UuidSchema::String), PrimitiveType::Fixed(len) => self.define_once(avro_fixed_schema((*len) as usize)?), PrimitiveType::Binary => AvroSchema::Bytes, diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs index 860a524261..79244a6e99 100644 --- a/crates/iceberg/src/spec/manifest/_serde.rs +++ b/crates/iceberg/src/spec/manifest/_serde.rs @@ -21,7 +21,8 @@ //! without a reader schema, so they decide which manifests read. A field that //! a writer may omit, per the spec's read rules for every format version the //! struct reads, must be an `Option` or have `#[serde(default)]`. -//! `test_parse_manifest_without_each_field` checks each field. +//! `test_parse_manifest_without_each_field` and +//! `test_parse_v1_manifest_without_each_field` check each field. use std::collections::HashMap; diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index e49dc26200..b22ff62864 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -38,7 +38,7 @@ use super::{ UNASSIGNED_SEQUENCE_NUMBER, }; use crate::avro::{Resolved, define_named_types_once}; -use crate::error::{Result, invalid_data}; +use crate::error::{Error, Result, invalid_data}; /// Whether a manifest with repeated Avro named type definitions was logged at /// warn level. A table written by an affected release can have many of them. @@ -82,7 +82,12 @@ impl Manifest { ); } rewritten = bs; - AvroReader::new(rewritten.as_slice())? + AvroReader::new(rewritten.as_slice()).map_err(|retry_error| { + Error::from(e).with_context( + "error after defining each named type once", + retry_error.to_string(), + ) + })? } Err(e) => return Err(e.into()), }; @@ -1594,6 +1599,49 @@ mod tests { } } + type ResetField = fn(&mut ManifestEntry); + + /// Removes each field in `optional` and `required` in turn from a manifest + /// that `full_schema` and `full_value` write as `full_entry`. Checks that the + /// manifest reads with an optional field reset and fails naming a required one. + fn assert_reads_without_each_field( + metadata: &ManifestMetadata, + full_schema: &Value, + full_value: &AvroValue, + full_entry: &ManifestEntry, + optional: &[(&[&str], ResetField)], + required: &[&[&str]], + ) { + for (path, reset) in optional { + let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); + remove_writer_field(&mut schema, &mut value, path); + let bs = write_avro_values_with_writer_schema(metadata, &schema, vec![value]); + + let manifest = Manifest::parse_avro(&bs).unwrap(); + + let mut expected = full_entry.clone(); + reset(&mut expected); + assert_eq!( + manifest, + Manifest::new(metadata.clone(), vec![expected]), + "{path:?}" + ); + } + + for path in required { + let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); + remove_writer_field(&mut schema, &mut value, path); + let bs = write_avro_values_with_writer_schema(metadata, &schema, vec![value]); + + let err = Manifest::parse_avro(&bs).unwrap_err(); + + assert!( + err.to_string().contains(path.last().unwrap()), + "{path:?}: {err}" + ); + } + } + #[test] fn test_parse_manifest_without_each_field() { // Fields that the spec doesn't require in every version must read as @@ -1637,7 +1685,6 @@ mod tests { ) .unwrap(); - type ResetField = fn(&mut ManifestEntry); let optional: [(&[&str], ResetField); 18] = [ (&["snapshot_id"], |e| e.snapshot_id = None), (&["sequence_number"], |e| e.sequence_number = None), @@ -1688,22 +1735,6 @@ mod tests { e.data_file.content_size_in_bytes = None }), ]; - for (path, reset) in optional { - let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); - remove_writer_field(&mut schema, &mut value, path); - let bs = write_avro_values_with_writer_schema(&metadata, &schema, vec![value]); - - let manifest = Manifest::parse_avro(&bs).unwrap(); - - let mut expected = full_entry.clone(); - reset(&mut expected); - assert_eq!( - manifest, - Manifest::new(metadata.clone(), vec![expected]), - "{path:?}" - ); - } - let required: [&[&str]; 7] = [ &["status"], &["data_file"], @@ -1713,18 +1744,102 @@ mod tests { &["data_file", "record_count"], &["data_file", "file_size_in_bytes"], ]; - for path in required { - let (mut schema, mut value) = (full_schema.clone(), full_value.clone()); - remove_writer_field(&mut schema, &mut value, path); - let bs = write_avro_values_with_writer_schema(&metadata, &schema, vec![value]); + assert_reads_without_each_field( + &metadata, + &full_schema, + &full_value, + &full_entry, + &optional, + &required, + ); + } - let err = Manifest::parse_avro(&bs).unwrap_err(); + #[test] + fn test_parse_v1_manifest_without_each_field() { + let mut metadata = writer_schema_test_metadata(); + metadata.format_version = FormatVersion::V1; + let partition_type = metadata + .partition_spec + .partition_type(&metadata.schema) + .unwrap(); + let full_schema = + serde_json::to_value(manifest_schema_v1(&partition_type).unwrap()).unwrap(); + let mut full_entry = expected_entry(Struct::from_iter([ + Some(Literal::long(5)), + Some(Literal::string("a")), + Some(Literal::double(2.5)), + ])); + // V1 has no sequence numbers, and every V1 entry reads with 0. + full_entry.sequence_number = Some(0); + full_entry.file_sequence_number = Some(0); + let data_file = &mut full_entry.data_file; + data_file.column_sizes = HashMap::from([(1, 40)]); + data_file.value_counts = HashMap::from([(1, 10)]); + data_file.null_value_counts = HashMap::from([(1, 1)]); + data_file.nan_value_counts = HashMap::from([(3, 2)]); + data_file.lower_bounds = HashMap::from([(1, Datum::long(1))]); + data_file.upper_bounds = HashMap::from([(1, Datum::long(9))]); + data_file.key_metadata = Some(vec![1, 2]); + data_file.split_offsets = Some(vec![4]); + data_file.sort_order_id = Some(0); + let full_value = to_value( + _serde::ManifestEntryV1::try_from( + full_entry.clone(), + &Type::Struct(partition_type.clone()), + ) + .unwrap(), + ) + .unwrap(); - assert!( - err.to_string().contains(path.last().unwrap()), - "{path:?}: {err}" - ); - } + let optional: [(&[&str], ResetField); 10] = [ + // Required in V1, but deprecated and not read. + (&["data_file", "block_size_in_bytes"], |_| {}), + (&["data_file", "column_sizes"], |e| { + e.data_file.column_sizes.clear() + }), + (&["data_file", "value_counts"], |e| { + e.data_file.value_counts.clear() + }), + (&["data_file", "null_value_counts"], |e| { + e.data_file.null_value_counts.clear() + }), + (&["data_file", "nan_value_counts"], |e| { + e.data_file.nan_value_counts.clear() + }), + (&["data_file", "lower_bounds"], |e| { + e.data_file.lower_bounds.clear() + }), + (&["data_file", "upper_bounds"], |e| { + e.data_file.upper_bounds.clear() + }), + (&["data_file", "key_metadata"], |e| { + e.data_file.key_metadata = None + }), + (&["data_file", "split_offsets"], |e| { + e.data_file.split_offsets = None + }), + (&["data_file", "sort_order_id"], |e| { + e.data_file.sort_order_id = None + }), + ]; + let required: [&[&str]; 8] = [ + &["status"], + &["snapshot_id"], + &["data_file"], + &["data_file", "file_path"], + &["data_file", "file_format"], + &["data_file", "partition"], + &["data_file", "record_count"], + &["data_file", "file_size_in_bytes"], + ]; + assert_reads_without_each_field( + &metadata, + &full_schema, + &full_value, + &full_entry, + &optional, + &required, + ); } #[test] @@ -2116,4 +2231,28 @@ mod tests { Struct::from_iter([Some(Literal::decimal(12345)), Some(Literal::decimal(-678))]) ); } + + #[test] + fn test_parse_manifest_with_repeated_named_type_definitions_reports_original_error() { + let bs = fs::read(format!( + "{}/testdata/manifests/repeated-decimal-type-definitions.avro", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + // Cut the header's sync marker in half, so the rewritten header parses + // its schema but fails to read the marker. The file ends with the marker. + let marker = &bs[bs.len() - 16..]; + let header_marker = bs.windows(16).position(|w| w == marker).unwrap(); + let truncated = &bs[..header_marker + 8]; + + let err = Manifest::parse_avro(truncated).unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + let message = err.to_string(); + assert!( + message.contains("Two named schema defined for same fullname"), + "{message}" + ); + assert!(message.contains("Failed to read marker bytes"), "{message}"); + } } diff --git a/crates/iceberg/src/spec/manifest_list/writer.rs b/crates/iceberg/src/spec/manifest_list/writer.rs index eb8a50b6ae..108daa951e 100644 --- a/crates/iceberg/src/spec/manifest_list/writer.rs +++ b/crates/iceberg/src/spec/manifest_list/writer.rs @@ -149,8 +149,9 @@ impl ManifestListWriter { FormatVersion::V2 => &MANIFEST_LIST_AVRO_SCHEMA_V2, FormatVersion::V3 => &MANIFEST_LIST_AVRO_SCHEMA_V3, }; - let mut avro_writer = Writer::new(avro_schema, Vec::new()) - .expect("Manifest list Avro schemas have no named references to resolve."); + let mut avro_writer = Writer::new(avro_schema, Vec::new()).expect( + "Manifest list Avro schemas should resolve because they refer to no named types.", + ); for (key, value) in metadata { avro_writer .add_user_metadata(key, value) From 95667f5be29b8c24a8a5155b80c1fb5986a1c0d0 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 8 Oct 2026 15:48:08 -0700 Subject: [PATCH 5/5] address PR feedback --- Cargo.lock | 1 + crates/iceberg/Cargo.toml | 1 + crates/iceberg/src/spec/manifest/_serde.rs | 2 ++ crates/iceberg/src/spec/manifest/mod.rs | 39 ++++++++++++++++------ 4 files changed, 33 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 85c1e9b250..a86ca2362c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2763,6 +2763,7 @@ dependencies = [ "tempfile", "tokio", "tracing", + "tracing-subscriber", "typed-builder", "typetag", "url", diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 8783251fd5..f22ed59b9b 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -95,6 +95,7 @@ rand = { workspace = true } regex = { workspace = true } serde_arrow = { version = "0.14", features = ["arrow-59"] } tempfile = { workspace = true } +tracing-subscriber = { workspace = true } [package.metadata.cargo-machete] # These dependencies are added to ensure minimal dependency version diff --git a/crates/iceberg/src/spec/manifest/_serde.rs b/crates/iceberg/src/spec/manifest/_serde.rs index 79244a6e99..7d89af64a9 100644 --- a/crates/iceberg/src/spec/manifest/_serde.rs +++ b/crates/iceberg/src/spec/manifest/_serde.rs @@ -76,6 +76,8 @@ impl ManifestEntryV2 { #[derive(Serialize, Deserialize)] pub(super) struct ManifestEntryV1 { status: i32, + // Required in v1 by the spec. Java declares it optional and writes null when + // snapshot ID inheritance is enabled, which #3371 tracks. pub snapshot_id: i64, data_file: DataFileSerde, } diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs index b22ff62864..f4269638a2 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -65,9 +65,23 @@ impl Manifest { // `schema_to_avro_schema` defined each named type once, so this // fallback stays while tables can contain manifests it wrote. Err(e) if matches!(e.details(), Details::AmbiguousSchemaDefinition(_)) => { - let Ok(Some((bs, repeated))) = define_named_types_once(bs) else { - return Err(e.into()); + let (bs, repeated) = match define_named_types_once(bs) { + Ok(Some(rewrite)) => rewrite, + Ok(None) => return Err(e.into()), + Err(rewrite_error) => { + return Err(Error::from(e).with_context( + "error defining each named type once", + rewrite_error.to_string(), + )); + } }; + rewritten = bs; + let reader = AvroReader::new(rewritten.as_slice()).map_err(|retry_error| { + Error::from(e).with_context( + "error after defining each named type once", + retry_error.to_string(), + ) + })?; let location = location.unwrap_or(""); if WARNED_REPEATED_DEFINITIONS.swap(true, Ordering::Relaxed) { tracing::debug!( @@ -81,13 +95,7 @@ impl Manifest { this are logged at debug level." ); } - rewritten = bs; - AvroReader::new(rewritten.as_slice()).map_err(|retry_error| { - Error::from(e).with_context( - "error after defining each named type once", - retry_error.to_string(), - ) - })? + reader } Err(e) => return Err(e.into()), }; @@ -197,6 +205,8 @@ mod tests { use apache_avro::{Codec, Writer, to_value}; use serde_json::{Value, to_vec}; use tempfile::TempDir; + use tracing::Level; + use tracing::subscriber::with_default; use super::*; use crate::ErrorKind; @@ -2244,8 +2254,14 @@ mod tests { let marker = &bs[bs.len() - 16..]; let header_marker = bs.windows(16).position(|w| w == marker).unwrap(); let truncated = &bs[..header_marker + 8]; + let log_dir = TempDir::new().unwrap(); + let log_path = log_dir.path().join("log"); + let subscriber = tracing_subscriber::fmt() + .with_max_level(Level::DEBUG) + .with_writer(Arc::new(fs::File::create(&log_path).unwrap())) + .finish(); - let err = Manifest::parse_avro(truncated).unwrap_err(); + let err = with_default(subscriber, || Manifest::parse_avro(truncated)).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); let message = err.to_string(); @@ -2254,5 +2270,8 @@ mod tests { "{message}" ); assert!(message.contains("Failed to read marker bytes"), "{message}"); + // The fallback logs only manifests that it reads. + let log = fs::read_to_string(&log_path).unwrap(); + assert!(!log.contains("more than once"), "{log}"); } }