This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3354-b0d8baaf966813f82363279db46d6688bf455d24 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 5b82ab99aeaf50bf4f5b3b776ac9cd6380a5f3e3 Author: Matt Butrovich <[email protected]> AuthorDate: Fri Oct 9 16:08:16 2026 +0000 perf(manifest): upgrade to apache-avro 0.22 and read manifest entries without Value (#3354) * perf(manifest): upgrade to apache-avro 0.22 and read manifest entries without Value * add test cases, update docs * address PR feedback * address PR feedback. * address PR feedback --- Cargo.lock | 110 ++- Cargo.toml | 2 +- crates/iceberg/Cargo.toml | 1 + crates/iceberg/src/avro/deserializer.rs | 457 ++++++++++ crates/iceberg/src/avro/mod.rs | 6 + crates/iceberg/src/avro/named_types.rs | 309 +++++++ crates/iceberg/src/avro/schema.rs | 158 +++- crates/iceberg/src/encryption/key_metadata.rs | 73 +- crates/iceberg/src/spec/manifest/_serde.rs | 72 +- crates/iceberg/src/spec/manifest/data_file.rs | 8 +- crates/iceberg/src/spec/manifest/mod.rs | 985 ++++++++++++++++++++- crates/iceberg/src/spec/manifest/reader.rs | 3 +- crates/iceberg/src/spec/manifest/writer.rs | 108 ++- crates/iceberg/src/spec/manifest_list/mod.rs | 6 +- crates/iceberg/src/spec/manifest_list/writer.rs | 4 +- crates/iceberg/src/spec/values/serde.rs | 51 +- crates/iceberg/src/spec/values/tests.rs | 70 +- crates/iceberg/testdata/manifests/README.md | 99 +++ .../testdata/manifests/pyiceberg-v2-data.avro | Bin 0 -> 4787 bytes .../repeated-decimal-type-definitions.avro | Bin 0 -> 3865 bytes 20 files changed, 2409 insertions(+), 113 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e2900acb2..75694c387 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" @@ -2747,10 +2759,11 @@ dependencies = [ "serde_json", "serde_repr", "serde_with", - "strum", + "strum 0.27.2", "tempfile", "tokio", "tracing", + "tracing-subscriber", "typed-builder", "typetag", "url", @@ -2878,7 +2891,7 @@ dependencies = [ "itertools 0.13.0", "regex", "sqlx", - "strum", + "strum 0.27.2", "tempfile", "tokio", "tracing", @@ -3574,6 +3587,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 +4200,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 +4571,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 +4664,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", @@ -5913,7 +5972,7 @@ checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" dependencies = [ "dotenvy", "either", - "heck", + "heck 0.5.0", "hex", "once_cell", "proc-macro2", @@ -6075,7 +6134,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 +6152,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 661c56b09..24f881994 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/Cargo.toml b/crates/iceberg/Cargo.toml index 8783251fd..f22ed59b9 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/avro/deserializer.rs b/crates/iceberg/src/avro/deserializer.rs new file mode 100644 index 000000000..e13092544 --- /dev/null +++ b/crates/iceberg/src/avro/deserializer.rs @@ -0,0 +1,457 @@ +// 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`. +//! - 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 +//! error. Conversions into `f32` or `f64` use `as` and can lose precision. +//! +//! apache-avro plans a `SchemaAwareResolvingDeserializer` that resolves against +//! a reader schema (<https://github.com/apache/avro-rs/issues/575>). 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<T>(pub T); + +impl<'de, T: Deserialize<'de>> Deserialize<'de> for Resolved<T> { + fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> { + T::deserialize(ResolvingDeserializer(deserializer)).map(Resolved) + } +} + +struct ResolvingDeserializer<D>(D); + +impl<'de, D: Deserializer<'de>> Deserializer<'de> for ResolvingDeserializer<D> { + type Error = D::Error; + + fn deserialize_any<V: Visitor<'de>>(self, visitor: V) -> Result<V::Value, D::Error> { + self.0.deserialize_any(ResolvingVisitor(visitor)) + } + + fn deserialize_option<V: Visitor<'de>>(self, visitor: V) -> Result<V::Value, D::Error> { + self.0.deserialize_any(OptionVisitor(visitor)) + } + + fn deserialize_ignored_any<V: Visitor<'de>>(self, visitor: V) -> Result<V::Value, D::Error> { + 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>(V); + +/// Forwards `visit_*` methods for values that contain no nested values. +macro_rules! forward_visit { + ($($method:ident($ty:ty)),* $(,)?) => { + $( + fn $method<E: Error>(self, v: $ty) -> Result<Self::Value, E> { + self.0.$method(v) + } + )* + }; +} + +impl<'de, V: Visitor<'de>> Visitor<'de> for ResolvingVisitor<V> { + 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<u8>), + } + + fn visit_none<E: Error>(self) -> Result<Self::Value, E> { + self.0.visit_none() + } + + fn visit_unit<E: Error>(self) -> Result<Self::Value, E> { + self.0.visit_unit() + } + + fn visit_some<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> { + self.0.visit_some(ResolvingDeserializer(deserializer)) + } + + fn visit_newtype_struct<D: Deserializer<'de>>( + self, + deserializer: D, + ) -> Result<Self::Value, D::Error> { + self.0 + .visit_newtype_struct(ResolvingDeserializer(deserializer)) + } + + fn visit_seq<A: SeqAccess<'de>>(self, seq: A) -> Result<Self::Value, A::Error> { + self.0.visit_seq(ResolvingAccess(seq)) + } + + fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> { + self.0.visit_map(ResolvingAccess(map)) + } + + fn visit_enum<A: EnumAccess<'de>>(self, data: A) -> Result<Self::Value, A::Error> { + 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>(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<E: Error>(self, v: $ty) -> Result<Self::Value, E> { + self.0.visit_some($deserializer(v)) + } + )* + }; +} + +impl<'de, V: Visitor<'de>> Visitor<'de> for OptionVisitor<V> { + 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<E: Error>(self, v: Vec<u8>) -> Result<Self::Value, E> { + self.0.visit_some(BytesDeserializer::new(&v)) + } + + fn visit_none<E: Error>(self) -> Result<Self::Value, E> { + self.0.visit_none() + } + + fn visit_unit<E: Error>(self) -> Result<Self::Value, E> { + self.0.visit_none() + } + + fn visit_some<D: Deserializer<'de>>(self, deserializer: D) -> Result<Self::Value, D::Error> { + self.0.visit_some(ResolvingDeserializer(deserializer)) + } + + fn visit_seq<A: SeqAccess<'de>>(self, seq: A) -> Result<Self::Value, A::Error> { + self.0 + .visit_some(SeqAccessDeserializer::new(ResolvingAccess(seq))) + } + + fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> { + self.0 + .visit_some(MapAccessDeserializer::new(ResolvingAccess(map))) + } + + fn visit_enum<A: EnumAccess<'de>>(self, data: A) -> Result<Self::Value, A::Error> { + 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>(A); + +impl<'de, A: SeqAccess<'de>> SeqAccess<'de> for ResolvingAccess<A> { + type Error = A::Error; + + fn next_element_seed<T: DeserializeSeed<'de>>( + &mut self, + seed: T, + ) -> Result<Option<T::Value>, A::Error> { + self.0.next_element_seed(ResolvingSeed(seed)) + } + + fn size_hint(&self) -> Option<usize> { + self.0.size_hint() + } +} + +impl<'de, A: MapAccess<'de>> MapAccess<'de> for ResolvingAccess<A> { + type Error = A::Error; + + fn next_key_seed<K: DeserializeSeed<'de>>( + &mut self, + seed: K, + ) -> Result<Option<K::Value>, A::Error> { + self.0.next_key_seed(ResolvingSeed(seed)) + } + + fn next_value_seed<T: DeserializeSeed<'de>>(&mut self, seed: T) -> Result<T::Value, A::Error> { + self.0.next_value_seed(ResolvingSeed(seed)) + } + + fn size_hint(&self) -> Option<usize> { + self.0.size_hint() + } +} + +struct ResolvingSeed<S>(S); + +impl<'de, S: DeserializeSeed<'de>> DeserializeSeed<'de> for ResolvingSeed<S> { + type Value = S::Value; + + fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<S::Value, D::Error> { + 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<String>, + } + + #[derive(Deserialize, Debug, PartialEq)] + enum Suit { + Hearts, + Spades, + } + + /// Writes `value` with `writer_schema` and reads it back as a `T`. + fn read<T: DeserializeOwned>( + writer_schema: &str, + value: AvroValue, + ) -> apache_avro::AvroResult<T> { + 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::<Resolved<T>>(); + values.next().unwrap().map(|value| value.0) + } + + fn check<T: DeserializeOwned + Debug + PartialEq>( + writer_schema: &str, + value: AvroValue, + expected: T, + ) { + assert_eq!( + read::<T>(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::<i64>); + check( + r#"["null", "long"]"#, + AvroValue::Union(0, Box::new(AvroValue::Null)), + None::<i64>, + ); + 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::<i32>(r#""long""#, AvroValue::Long(i64::from(i32::MAX) + 1)).is_err()); + assert!(read::<u8>(r#""int""#, AvroValue::Int(-1)).is_err()); + assert!( + read::<i64>( + r#"["null", "long"]"#, + AvroValue::Union(0, Box::new(AvroValue::Null)) + ) + .is_err() + ); + assert!(read::<String>(r#""long""#, AvroValue::Long(1)).is_err()); + } +} diff --git a/crates/iceberg/src/avro/mod.rs b/crates/iceberg/src/avro/mod.rs index f2a9310e7..0bc5a7004 100644 --- a/crates/iceberg/src/avro/mod.rs +++ b/crates/iceberg/src/avro/mod.rs @@ -16,5 +16,11 @@ // 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; +#[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 new file mode 100644 index 000000000..a006a8415 --- /dev/null +++ b/crates/iceberg/src/avro/named_types.rs @@ -0,0 +1,309 @@ +// 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<Option<(Vec<u8>, Vec<String>)>> { + 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<T>( + bs: &[u8], + rewrite: impl FnOnce(&mut JsonValue) -> Option<T>, +) -> Result<Option<(Vec<u8>, T)>> { + 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<bytes>`, 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 = 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(rewritten) = rewrite(&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(), 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<u8> { + fn inline_references(schema: &mut JsonValue, defined: &mut HashMap<String, JsonValue>) { + 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 +/// name. Returns the replaced names, or `None` if a repeated definition differs +/// from the first one. +fn reference_repeated_definitions(schema: &mut JsonValue) -> Option<Vec<String>> { + 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<String, Map<String, JsonValue>>, + repeated: &mut Vec<String>, +) -> 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<String, JsonValue>, enclosing_namespace: Option<&str>) -> Option<String> { + 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 908dff5c9..47254317e 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<Name>, +} + +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<AvroSchema, AvroRecordField>; @@ -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<AvroSchemaOrField> { 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,13 @@ 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)?, + // 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, 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 +295,7 @@ impl SchemaVisitor for SchemaToAvroSchema { pub(crate) fn schema_to_avro_schema(name: impl ToString, schema: &Schema) -> Result<AvroSchema> { let mut converter = SchemaToAvroSchema { schema: name.to_string(), + defined_names: HashSet::new(), }; visit_schema(schema, &mut converter).map(Either::unwrap_left) @@ -305,7 +325,6 @@ pub(crate) fn avro_fixed_schema(len: usize) -> Result<AvroSchema> { doc: None, size: len, attributes: Default::default(), - default: None, })) } @@ -317,16 +336,15 @@ pub(crate) fn avro_decimal_schema(precision: usize, scale: usize) -> Result<Avro Ok(AvroSchema::Decimal(DecimalSchema { precision, scale, - inner: Box::new(AvroSchema::Fixed(FixedSchema { + inner: InnerDecimalSchema::Fixed(FixedSchema { // Name is not restricted by the spec. // Refer to iceberg-python https://github.com/apache/iceberg-python/blob/d8bc1ca9af7957ce4d4db99a52c701ac75db7688/pyiceberg/utils/schema_conversion.py#L574-L582 - name: Name::new(&format!("decimal_{precision}_{scale}")).unwrap(), + name: Name::new(format!("decimal_{precision}_{scale}")).unwrap(), aliases: None, doc: None, size: Type::decimal_required_bytes(precision as u32)? as usize, attributes: Default::default(), - default: None, - })), + }), })) } @@ -549,7 +567,7 @@ impl AvroSchemaVisitor for AvroSchemaToSchema { AvroSchema::Long => 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 +645,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 +674,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 +991,79 @@ mod tests { } #[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 \ + 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 +1291,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 d56c8dbd3..b3222eec5 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<Box<[u8]>> { - _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<serde_bytes::ByteBuf>, - pub file_length: Option<u64>, + pub file_length: Option<i64>, } 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<Self> { + 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 336c883f1..7d89af64a 100644 --- a/crates/iceberg/src/spec/manifest/_serde.rs +++ b/crates/iceberg/src/spec/manifest/_serde.rs @@ -15,6 +15,15 @@ // 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` and +//! `test_parse_v1_manifest_without_each_field` check each field. + use std::collections::HashMap; use serde_derive::{Deserialize, Serialize}; @@ -58,6 +67,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)?, }) } @@ -66,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, } @@ -92,6 +104,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 +174,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<Self, Error> { + 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, @@ -315,9 +338,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, }; @@ -412,6 +436,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/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index 1fcdaa3a8..94568ab44 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<W: Write>( 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<W: Write>( 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<R: Read>( 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 3bd4dc8ad..f4269638a 100644 --- a/crates/iceberg/src/spec/manifest/mod.rs +++ b/crates/iceberg/src/spec/manifest/mod.rs @@ -27,15 +27,22 @@ mod reader; pub use reader::*; mod writer; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; -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::error::{Result, invalid_data}; +use crate::avro::{Resolved, define_named_types_once}; +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. +static WARNED_REPEATED_DEFINITIONS: AtomicBool = AtomicBool::new(false); /// A manifest contains metadata and a list of entries. #[derive(Debug, PartialEq, Eq, Clone)] @@ -45,9 +52,53 @@ 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<ManifestEntry>)> { - 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<ManifestEntry>)> { + 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 (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("<unknown location>"); + 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." + ); + } + reader + } + Err(e) => return Err(e.into()), + }; // Parse manifest metadata let meta = reader.user_metadata(); @@ -61,35 +112,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::<Result<Vec<_>>>()? - } + FormatVersion::V1 => reader + .into_deser_iter::<Resolved<_serde::ManifestEntryV1>>() + .map(|entry| { + entry?.0.try_into( + metadata.partition_spec.spec_id(), + &partition_struct_type, + &metadata.schema, + ) + }) + .collect::<Result<Vec<_>>>()?, // 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::<Result<Vec<_>>>()? - } + FormatVersion::V2 | FormatVersion::V3 => reader + .into_deser_iter::<Resolved<_serde::ManifestEntryV2>>() + .map(|entry| { + entry?.0.try_into( + metadata.partition_spec.spec_id(), + &partition_struct_type, + &metadata.schema, + ) + }) + .collect::<Result<Vec<_>>>()?, }; Ok((metadata, entries)) @@ -97,7 +140,7 @@ impl Manifest { /// Parse manifest from bytes of avro file. pub fn parse_avro(bs: &[u8]) -> Result<Self> { - let (metadata, entries) = Self::try_from_avro_bytes(bs)?; + let (metadata, entries) = Self::try_from_avro_bytes(bs, None)?; Ok(Self::new(metadata, entries)) } @@ -158,11 +201,15 @@ 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; + use tracing::Level; + use tracing::subscriber::with_default; use super::*; + use crate::ErrorKind; use crate::io::FileIO; use crate::spec::{Literal, NestedField, PrimitiveType, Struct, Transform, Type}; @@ -356,7 +403,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 +444,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 +1408,870 @@ 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<Value>, + ) -> Vec<u8> { + 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<AvroValue>, + ) -> Vec<u8> { + 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 { + writer + .append_value(entry.resolve(&avro_schema).unwrap()) + .unwrap(); + } + 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); + } + } + + 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 + // 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(); + + 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 + }), + ]; + 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"], + ]; + assert_reads_without_each_field( + &metadata, + &full_schema, + &full_value, + &full_entry, + &optional, + &required, + ); + } + + #[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(); + + 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] + 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_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 + // 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<long>` 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_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!( + "{}/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))]) + ); + } + + #[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 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 = with_default(subscriber, || 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}"); + // The fallback logs only manifests that it reads. + let log = fs::read_to_string(&log_path).unwrap(); + assert!(!log.contains("more than once"), "{log}"); + } } diff --git a/crates/iceberg/src/spec/manifest/reader.rs b/crates/iceberg/src/spec/manifest/reader.rs index e7044a56b..a056d207c 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 15608a63f..a79312dfa 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()?; @@ -580,8 +580,12 @@ mod tests { use tempfile::TempDir; use super::*; + use crate::avro::define_named_types_repeatedly; 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 +823,102 @@ mod tests { ManifestContentType::Deletes, ); } + + /// 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<u8> { + 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 bs = fs::read(&path).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); + } + 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] + 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 25dd5bb93..24a4debf9 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<ManifestList> { 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::<std::result::Result<Vec<Value>, _>>()?); 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 5ce2cfb25..108daa951 100644 --- a/crates/iceberg/src/spec/manifest_list/writer.rs +++ b/crates/iceberg/src/spec/manifest_list/writer.rs @@ -149,7 +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()); + 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) diff --git a/crates/iceberg/src/spec/values/serde.rs b/crates/iceberg/src/spec/values/serde.rs index 896235b1e..a9c611517 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,53 @@ pub(crate) mod _serde { pub fn try_into(self, ty: &Type) -> Result<Option<Literal>, 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. + /// + /// 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<Self, Error> { + 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<String, RawLiteralEnum> = 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::<Result<_, Error>>()? + }; + 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 ff51f318e..c756bc3f8 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<u8>, 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::<ByteBuf>(&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 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(), + 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 test_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 000000000..6b68b71ed --- /dev/null +++ b/crates/iceberg/testdata/manifests/README.md @@ -0,0 +1,99 @@ +<!-- + ~ 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. +--> + + +# 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. + +`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 000000000..c8c686644 Binary files /dev/null and b/crates/iceberg/testdata/manifests/pyiceberg-v2-data.avro differ 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 000000000..598268c03 Binary files /dev/null and b/crates/iceberg/testdata/manifests/repeated-decimal-type-definitions.avro differ
