From 2d95153741813ade9f7ee79955dbc8665b166369 Mon Sep 17 00:00:00 2001 From: Kacper Lisikiewicz <247186352+kacper-li-airspace-intelligence@users.noreply.github.com> Date: Tue, 15 Sep 2026 09:35:33 +0200 Subject: [PATCH 1/3] perf: parse manifest list entries without buffering Avro values --- .../iceberg/src/spec/manifest_list/_serde.rs | 45 +--- crates/iceberg/src/spec/manifest_list/mod.rs | 30 ++- crates/iceberg/tests/manifest_list_memory.rs | 200 ++++++++++++++++++ 3 files changed, 229 insertions(+), 46 deletions(-) create mode 100644 crates/iceberg/tests/manifest_list_memory.rs diff --git a/crates/iceberg/src/spec/manifest_list/_serde.rs b/crates/iceberg/src/spec/manifest_list/_serde.rs index 2a10c39ea3..5bb663f9c5 100644 --- a/crates/iceberg/src/spec/manifest_list/_serde.rs +++ b/crates/iceberg/src/spec/manifest_list/_serde.rs @@ -27,37 +27,28 @@ use crate::Error; use crate::error::Result; use crate::spec::FieldSummary; +#[cfg(test)] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(transparent)] pub(crate) struct ManifestListV3 { entries: Vec, } +#[cfg(test)] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(transparent)] pub(crate) struct ManifestListV2 { entries: Vec, } +#[cfg(test)] #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(transparent)] pub(crate) struct ManifestListV1 { entries: Vec, } -impl ManifestListV3 { - /// Converts the [ManifestListV3] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom for ManifestListV3 { type Error = Error; @@ -72,19 +63,7 @@ impl TryFrom for ManifestListV3 { } } -impl ManifestListV2 { - /// Converts the [ManifestListV2] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom for ManifestListV2 { type Error = Error; @@ -99,19 +78,7 @@ impl TryFrom for ManifestListV2 { } } -impl ManifestListV1 { - /// Converts the [ManifestListV1] into a [ManifestList]. - pub fn try_into(self) -> Result { - Ok(super::ManifestList { - entries: self - .entries - .into_iter() - .map(|v| v.try_into()) - .collect::>>()?, - }) - } -} - +#[cfg(test)] impl TryFrom for ManifestListV1 { type Error = Error; diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index 25dd5bb93e..1fb17b6b09 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/crates/iceberg/src/spec/manifest_list/mod.rs @@ -23,7 +23,6 @@ mod manifest_file; mod reader; mod writer; -use apache_avro::types::Value; use apache_avro::{Reader, from_value}; pub use manifest_file::*; pub use reader::*; @@ -58,22 +57,39 @@ pub struct ManifestList { impl ManifestList { /// Parse manifest list from bytes. + /// Convert each Avro value immediately, retaining only the final typed entries. + /// A decoding or conversion error still discards the entire result. pub fn parse_with_version(bs: &[u8], version: FormatVersion) -> Result { match version { FormatVersion::V1 => { let reader = Reader::with_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1, bs)?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV1>(&values)?.try_into() + let entries = reader + .map(|value| { + let value = value?; + from_value::<_serde::ManifestFileV1>(&value)?.try_into() + }) + .collect::>>()?; + Ok(ManifestList { entries }) } FormatVersion::V2 => { let reader = Reader::new(bs)?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV2>(&values)?.try_into() + let entries = reader + .map(|value| { + let value = value?; + from_value::<_serde::ManifestFileV2>(&value)?.try_into() + }) + .collect::>>()?; + Ok(ManifestList { entries }) } FormatVersion::V3 => { let reader = Reader::new(bs)?; - let values = Value::Array(reader.collect::, _>>()?); - from_value::<_serde::ManifestListV3>(&values)?.try_into() + let entries = reader + .map(|value| { + let value = value?; + from_value::<_serde::ManifestFileV3>(&value)?.try_into() + }) + .collect::>>()?; + Ok(ManifestList { entries }) } } } diff --git a/crates/iceberg/tests/manifest_list_memory.rs b/crates/iceberg/tests/manifest_list_memory.rs new file mode 100644 index 0000000000..9da19eed51 --- /dev/null +++ b/crates/iceberg/tests/manifest_list_memory.rs @@ -0,0 +1,200 @@ +// 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-list compatibility and peak live-allocation regression. + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering::Relaxed; +use std::time::Instant; + +use apache_avro::types::Value; +use apache_avro::{Codec, Reader, Writer}; +use iceberg::io::FileIO; +use iceberg::spec::{ + FieldSummary, FormatVersion, ManifestContentType, ManifestFile, ManifestList, + ManifestListWriter, +}; + +struct CountingAllocator; +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + +fn allocated(bytes: usize) { + PEAK.fetch_max(LIVE.fetch_add(bytes, Relaxed) + bytes, Relaxed); +} + +// The allocator is confined to this test executable. Every allocation is tracked, +// including those made before measurement, so deallocation cannot underflow LIVE. +unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc(layout) }; + if !ptr.is_null() { + allocated(layout.size()); + } + ptr + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc_zeroed(layout) }; + if !ptr.is_null() { + allocated(layout.size()); + } + ptr + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + unsafe { System.dealloc(ptr, layout) }; + LIVE.fetch_sub(layout.size(), Relaxed); + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { + let new_ptr = unsafe { System.realloc(ptr, layout, size) }; + if !new_ptr.is_null() { + LIVE.fetch_sub(layout.size(), Relaxed); + allocated(size); + } + new_ptr + } +} + +#[global_allocator] +static ALLOCATOR: CountingAllocator = CountingAllocator; + +fn entry(version: FormatVersion, index: i64) -> ManifestFile { + ManifestFile { + manifest_path: format!("s3://test/metadata/manifest-{index}.avro"), + manifest_length: 1234, + partition_spec_id: index as i32, + content: ManifestContentType::Data, + sequence_number: if version == FormatVersion::V1 { 0 } else { 7 }, + min_sequence_number: if version == FormatVersion::V1 { 0 } else { 3 }, + added_snapshot_id: 42, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(9), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: true, + contains_nan: None, + lower_bound: Some(vec![1, 0, 0, 0].into()), + upper_bound: None, + }]), + key_metadata: Some(vec![1, 2, 3]), + first_row_id: (version == FormatVersion::V3).then_some(100 + index as u64 * 9), + } +} + +async fn encode(version: FormatVersion, entries: Vec) -> Vec { + let io = FileIO::new_with_memory(); + let output = io.new_output("memory:///manifest-list.avro").unwrap(); + let writer = output.writer().await.unwrap(); + let mut writer = match version { + FormatVersion::V1 => ManifestListWriter::v1(writer, 42, None), + FormatVersion::V2 => ManifestListWriter::v2(writer, 42, None, 7), + FormatVersion::V3 => ManifestListWriter::v3(writer, 42, None, 7, Some(100)), + }; + writer.add_manifests(entries.into_iter()).unwrap(); + writer.close().await.unwrap(); + io.new_input("memory:///manifest-list.avro") + .unwrap() + .read() + .await + .unwrap() + .to_vec() +} + +// A single test keeps the allocation measurement isolated from parallel tests. +#[test] +fn manifest_parser_preserves_formats_and_bounds_intermediate_allocations() { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + for version in [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3] { + let expected = vec![entry(version, 0), entry(version, 1)]; + let encoded = runtime.block_on(encode(version, expected.clone())); + assert_eq!( + ManifestList::parse_with_version(&encoded, version) + .unwrap() + .entries(), + expected + ); + let empty = runtime.block_on(encode(version, vec![])); + assert!( + ManifestList::parse_with_version(&empty, version) + .unwrap() + .entries() + .is_empty() + ); + // Truncating the final sync marker must not return a partial list. + assert!(ManifestList::parse_with_version(&encoded[..encoded.len() - 1], version).is_err()); + if version == FormatVersion::V1 { + // Reading a V1 file as V2 must retain the version-default behavior. + assert_eq!( + ManifestList::parse_with_version(&encoded, FormatVersion::V2) + .unwrap() + .entries(), + expected + ); + } + } + + let seed = runtime.block_on(encode(FormatVersion::V2, vec![entry(FormatVersion::V2, 0)])); + let mut reader = Reader::new(seed.as_slice()).unwrap(); + let schema = reader.writer_schema().clone(); + let mut value = reader.next().unwrap().unwrap(); + for codec in [ + Codec::Null, + Codec::Snappy, + Codec::Zstandard(Default::default()), + ] { + let mut writer = Writer::with_codec(&schema, Vec::new(), codec); + for index in 0..10_000 { + let Value::Record(fields) = &mut value else { + panic!("expected a manifest record"); + }; + let (_, path) = fields + .iter_mut() + .find(|(name, _)| name == "manifest_path") + .unwrap(); + *path = Value::String(format!("s3://test/metadata/manifest-{index:05}.avro")); + writer.append(value.clone()).unwrap(); + } + let encoded = writer.into_inner().unwrap(); + let baseline = LIVE.load(Relaxed); + PEAK.store(baseline, Relaxed); + let started = Instant::now(); + let parsed = ManifestList::parse_with_version(&encoded, FormatVersion::V2).unwrap(); + let elapsed = started.elapsed(); + let peak = PEAK.load(Relaxed).saturating_sub(baseline); + assert_eq!(parsed.entries().len(), 10_000); + assert!(parsed.entries().iter().enumerate().all(|(index, actual)| { + let mut expected = entry(FormatVersion::V2, 0); + expected.manifest_path = format!("s3://test/metadata/manifest-{index:05}.avro"); + actual == &expected + })); + // The original whole-Value-array parser exceeds this budget. This bounds + // live allocations in the parser; it is deliberately not an RSS promise. + eprintln!( + "10,000 entries, {codec:?}: {} encoded bytes, {peak} bytes peak live allocations, {elapsed:?}", + encoded.len() + ); + assert!(peak < 16 * 1024 * 1024, "parser peak: {peak} bytes"); + } +} From 3db218ec660658ca33acfebcf849006d727065c9 Mon Sep 17 00:00:00 2001 From: Kacper Lisikiewicz Date: Tue, 13 Oct 2026 16:06:10 +0200 Subject: [PATCH 2/3] Move manifest list parser tests into unit tests Cover round trips, empty lists, truncated input and V1 read as V2 as separate unit tests, and drop the allocation-budget integration test. --- crates/iceberg/src/spec/manifest_list/mod.rs | 87 ++++++++ crates/iceberg/tests/manifest_list_memory.rs | 200 ------------------- 2 files changed, 87 insertions(+), 200 deletions(-) delete mode 100644 crates/iceberg/tests/manifest_list_memory.rs diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index d9d5ce9d71..c0e5b086bc 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/crates/iceberg/src/spec/manifest_list/mod.rs @@ -155,6 +155,93 @@ mod test { assert_eq!(manifest_list, parsed_manifest_list); } + const VERSIONS: [FormatVersion; 3] = [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3]; + + fn entry(version: FormatVersion, index: i64) -> ManifestFile { + ManifestFile { + manifest_path: format!("s3://test/metadata/manifest-{index}.avro"), + manifest_length: 1234, + partition_spec_id: index as i32, + content: ManifestContentType::Data, + sequence_number: if version == FormatVersion::V1 { 0 } else { 7 }, + min_sequence_number: if version == FormatVersion::V1 { 0 } else { 3 }, + added_snapshot_id: 42, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(9), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: true, + contains_nan: None, + lower_bound: Some(vec![1, 0, 0, 0].into()), + upper_bound: None, + }]), + key_metadata: Some(vec![1, 2, 3]), + first_row_id: (version == FormatVersion::V3).then_some(100 + index as u64 * 9), + } + } + + fn two_entries(version: FormatVersion) -> Vec { + vec![entry(version, 0), entry(version, 1)] + } + + async fn encode(version: FormatVersion, entries: Vec) -> Vec { + let io = FileIO::new_with_memory(); + let output = io.new_output("memory:///manifest-list.avro").unwrap(); + let writer = output.writer().await.unwrap(); + let mut writer = match version { + FormatVersion::V1 => ManifestListWriter::v1(writer, 42, None), + FormatVersion::V2 => ManifestListWriter::v2(writer, 42, None, 7), + FormatVersion::V3 => ManifestListWriter::v3(writer, 42, None, 7, Some(100)), + }; + writer.add_manifests(entries.into_iter()).unwrap(); + writer.close().await.unwrap(); + io.new_input("memory:///manifest-list.avro") + .unwrap() + .read() + .await + .unwrap() + .to_vec() + } + + #[tokio::test] + async fn test_parse_manifest_list_round_trip() { + for version in VERSIONS { + let expected = two_entries(version); + let bs = encode(version, expected.clone()).await; + let parsed = ManifestList::parse_with_version(&bs, version).unwrap(); + assert_eq!(parsed.entries(), expected, "{version:?}"); + } + } + + #[tokio::test] + async fn test_parse_empty_manifest_list() { + for version in VERSIONS { + let bs = encode(version, vec![]).await; + let parsed = ManifestList::parse_with_version(&bs, version).unwrap(); + assert!(parsed.entries().is_empty(), "{version:?}"); + } + } + + #[tokio::test] + async fn test_parse_truncated_manifest_list() { + for version in VERSIONS { + let bs = encode(version, two_entries(version)).await; + let result = ManifestList::parse_with_version(&bs[..bs.len() - 1], version); + assert!(result.is_err(), "{version:?}"); + } + } + + #[tokio::test] + async fn test_parse_manifest_list_v1_as_v2() { + let expected = two_entries(FormatVersion::V1); + let bs = encode(FormatVersion::V1, expected.clone()).await; + let parsed = ManifestList::parse_with_version(&bs, FormatVersion::V2).unwrap(); + assert_eq!(parsed.entries(), expected); + } + #[tokio::test] async fn test_parse_manifest_list_v2() { let manifest_list = ManifestList { diff --git a/crates/iceberg/tests/manifest_list_memory.rs b/crates/iceberg/tests/manifest_list_memory.rs deleted file mode 100644 index 57dd857c1c..0000000000 --- a/crates/iceberg/tests/manifest_list_memory.rs +++ /dev/null @@ -1,200 +0,0 @@ -// 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-list compatibility and peak live-allocation regression. - -use std::alloc::{GlobalAlloc, Layout, System}; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering::Relaxed; -use std::time::Instant; - -use apache_avro::types::Value; -use apache_avro::{Codec, Reader, Writer}; -use iceberg::io::FileIO; -use iceberg::spec::{ - FieldSummary, FormatVersion, ManifestContentType, ManifestFile, ManifestList, - ManifestListWriter, -}; - -struct CountingAllocator; -static LIVE: AtomicUsize = AtomicUsize::new(0); -static PEAK: AtomicUsize = AtomicUsize::new(0); - -fn allocated(bytes: usize) { - PEAK.fetch_max(LIVE.fetch_add(bytes, Relaxed) + bytes, Relaxed); -} - -// The allocator is confined to this test executable. Every allocation is tracked, -// including those made before measurement, so deallocation cannot underflow LIVE. -unsafe impl GlobalAlloc for CountingAllocator { - unsafe fn alloc(&self, layout: Layout) -> *mut u8 { - let ptr = unsafe { System.alloc(layout) }; - if !ptr.is_null() { - allocated(layout.size()); - } - ptr - } - - unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { - let ptr = unsafe { System.alloc_zeroed(layout) }; - if !ptr.is_null() { - allocated(layout.size()); - } - ptr - } - - unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { - unsafe { System.dealloc(ptr, layout) }; - LIVE.fetch_sub(layout.size(), Relaxed); - } - - unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { - let new_ptr = unsafe { System.realloc(ptr, layout, size) }; - if !new_ptr.is_null() { - LIVE.fetch_sub(layout.size(), Relaxed); - allocated(size); - } - new_ptr - } -} - -#[global_allocator] -static ALLOCATOR: CountingAllocator = CountingAllocator; - -fn entry(version: FormatVersion, index: i64) -> ManifestFile { - ManifestFile { - manifest_path: format!("s3://test/metadata/manifest-{index}.avro"), - manifest_length: 1234, - partition_spec_id: index as i32, - content: ManifestContentType::Data, - sequence_number: if version == FormatVersion::V1 { 0 } else { 7 }, - min_sequence_number: if version == FormatVersion::V1 { 0 } else { 3 }, - added_snapshot_id: 42, - added_files_count: Some(1), - existing_files_count: Some(0), - deleted_files_count: Some(0), - added_rows_count: Some(9), - existing_rows_count: Some(0), - deleted_rows_count: Some(0), - partitions: Some(vec![FieldSummary { - contains_null: true, - contains_nan: None, - lower_bound: Some(vec![1, 0, 0, 0].into()), - upper_bound: None, - }]), - key_metadata: Some(vec![1, 2, 3]), - first_row_id: (version == FormatVersion::V3).then_some(100 + index as u64 * 9), - } -} - -async fn encode(version: FormatVersion, entries: Vec) -> Vec { - let io = FileIO::new_with_memory(); - let output = io.new_output("memory:///manifest-list.avro").unwrap(); - let writer = output.writer().await.unwrap(); - let mut writer = match version { - FormatVersion::V1 => ManifestListWriter::v1(writer, 42, None), - FormatVersion::V2 => ManifestListWriter::v2(writer, 42, None, 7), - FormatVersion::V3 => ManifestListWriter::v3(writer, 42, None, 7, Some(100)), - }; - writer.add_manifests(entries.into_iter()).unwrap(); - writer.close().await.unwrap(); - io.new_input("memory:///manifest-list.avro") - .unwrap() - .read() - .await - .unwrap() - .to_vec() -} - -// A single test keeps the allocation measurement isolated from parallel tests. -#[test] -fn manifest_parser_preserves_formats_and_bounds_intermediate_allocations() { - let runtime = tokio::runtime::Builder::new_current_thread() - .build() - .unwrap(); - for version in [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3] { - let expected = vec![entry(version, 0), entry(version, 1)]; - let encoded = runtime.block_on(encode(version, expected.clone())); - assert_eq!( - ManifestList::parse_with_version(&encoded, version) - .unwrap() - .entries(), - expected - ); - let empty = runtime.block_on(encode(version, vec![])); - assert!( - ManifestList::parse_with_version(&empty, version) - .unwrap() - .entries() - .is_empty() - ); - // Truncating the final sync marker must not return a partial list. - assert!(ManifestList::parse_with_version(&encoded[..encoded.len() - 1], version).is_err()); - if version == FormatVersion::V1 { - // Reading a V1 file as V2 must retain the version-default behavior. - assert_eq!( - ManifestList::parse_with_version(&encoded, FormatVersion::V2) - .unwrap() - .entries(), - expected - ); - } - } - - let seed = runtime.block_on(encode(FormatVersion::V2, vec![entry(FormatVersion::V2, 0)])); - let mut reader = Reader::new(seed.as_slice()).unwrap(); - let schema = reader.writer_schema().clone(); - let mut value = reader.next().unwrap().unwrap(); - for codec in [ - Codec::Null, - Codec::Snappy, - Codec::Zstandard(Default::default()), - ] { - let mut writer = Writer::with_codec(&schema, Vec::new(), codec).unwrap(); - for index in 0..10_000 { - let Value::Record(fields) = &mut value else { - panic!("expected a manifest record"); - }; - let (_, path) = fields - .iter_mut() - .find(|(name, _)| name == "manifest_path") - .unwrap(); - *path = Value::String(format!("s3://test/metadata/manifest-{index:05}.avro")); - writer.append_value(value.clone()).unwrap(); - } - let encoded = writer.into_inner().unwrap(); - let baseline = LIVE.load(Relaxed); - PEAK.store(baseline, Relaxed); - let started = Instant::now(); - let parsed = ManifestList::parse_with_version(&encoded, FormatVersion::V2).unwrap(); - let elapsed = started.elapsed(); - let peak = PEAK.load(Relaxed).saturating_sub(baseline); - assert_eq!(parsed.entries().len(), 10_000); - assert!(parsed.entries().iter().enumerate().all(|(index, actual)| { - let mut expected = entry(FormatVersion::V2, 0); - expected.manifest_path = format!("s3://test/metadata/manifest-{index:05}.avro"); - actual == &expected - })); - // The original whole-Value-array parser exceeds this budget. This bounds - // live allocations in the parser; it is deliberately not an RSS promise. - eprintln!( - "10,000 entries, {codec:?}: {} encoded bytes, {peak} bytes peak live allocations, {elapsed:?}", - encoded.len() - ); - assert!(peak < 16 * 1024 * 1024, "parser peak: {peak} bytes"); - } -} From 7015b312007776982825d6e13b71f9dcad179a0f Mon Sep 17 00:00:00 2001 From: Kacper Lisikiewicz Date: Wed, 14 Oct 2026 07:35:46 +0200 Subject: [PATCH 3/3] Fold manifest list round-trip cases into existing tests Drop the round-trip and V1-as-V2 tests, which duplicate test_parse_manifest_list_v* and test_manifest_list_writer_v1_as_v2. Set a non-empty key_metadata and null contains_nan and upper_bound on existing entries to keep that coverage. --- crates/iceberg/src/spec/manifest_list/mod.rs | 42 +++++--------------- 1 file changed, 10 insertions(+), 32 deletions(-) diff --git a/crates/iceberg/src/spec/manifest_list/mod.rs b/crates/iceberg/src/spec/manifest_list/mod.rs index c0e5b086bc..9c599102c3 100644 --- a/crates/iceberg/src/spec/manifest_list/mod.rs +++ b/crates/iceberg/src/spec/manifest_list/mod.rs @@ -119,7 +119,7 @@ mod test { existing_rows_count: Some(0), deleted_rows_count: Some(0), partitions: Some(vec![]), - key_metadata: None, + key_metadata: Some(vec![1, 2, 3]), first_row_id: None, } ] @@ -157,11 +157,11 @@ mod test { const VERSIONS: [FormatVersion; 3] = [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3]; - fn entry(version: FormatVersion, index: i64) -> ManifestFile { + fn entry(version: FormatVersion) -> ManifestFile { ManifestFile { - manifest_path: format!("s3://test/metadata/manifest-{index}.avro"), + manifest_path: "s3://test/metadata/manifest.avro".to_string(), manifest_length: 1234, - partition_spec_id: index as i32, + partition_spec_id: 0, content: ManifestContentType::Data, sequence_number: if version == FormatVersion::V1 { 0 } else { 7 }, min_sequence_number: if version == FormatVersion::V1 { 0 } else { 3 }, @@ -179,14 +179,10 @@ mod test { upper_bound: None, }]), key_metadata: Some(vec![1, 2, 3]), - first_row_id: (version == FormatVersion::V3).then_some(100 + index as u64 * 9), + first_row_id: (version == FormatVersion::V3).then_some(100), } } - fn two_entries(version: FormatVersion) -> Vec { - vec![entry(version, 0), entry(version, 1)] - } - async fn encode(version: FormatVersion, entries: Vec) -> Vec { let io = FileIO::new_with_memory(); let output = io.new_output("memory:///manifest-list.avro").unwrap(); @@ -206,16 +202,6 @@ mod test { .to_vec() } - #[tokio::test] - async fn test_parse_manifest_list_round_trip() { - for version in VERSIONS { - let expected = two_entries(version); - let bs = encode(version, expected.clone()).await; - let parsed = ManifestList::parse_with_version(&bs, version).unwrap(); - assert_eq!(parsed.entries(), expected, "{version:?}"); - } - } - #[tokio::test] async fn test_parse_empty_manifest_list() { for version in VERSIONS { @@ -228,20 +214,12 @@ mod test { #[tokio::test] async fn test_parse_truncated_manifest_list() { for version in VERSIONS { - let bs = encode(version, two_entries(version)).await; + let bs = encode(version, vec![entry(version)]).await; let result = ManifestList::parse_with_version(&bs[..bs.len() - 1], version); assert!(result.is_err(), "{version:?}"); } } - #[tokio::test] - async fn test_parse_manifest_list_v1_as_v2() { - let expected = two_entries(FormatVersion::V1); - let bs = encode(FormatVersion::V1, expected.clone()).await; - let parsed = ManifestList::parse_with_version(&bs, FormatVersion::V2).unwrap(); - assert_eq!(parsed.entries(), expected); - } - #[tokio::test] async fn test_parse_manifest_list_v2() { let manifest_list = ManifestList { @@ -281,9 +259,9 @@ mod test { existing_rows_count: Some(0), deleted_rows_count: Some(0), partitions: Some( - vec![FieldSummary { contains_null: false, contains_nan: Some(false), lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: Some(Datum::float(2.1_f32).to_bytes().unwrap())}] + vec![FieldSummary { contains_null: false, contains_nan: None, lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: None}] ), - key_metadata: None, + key_metadata: Some(vec![1, 2, 3]), first_row_id: None, } ] @@ -399,9 +377,9 @@ mod test { existing_rows_count: Some(0), deleted_rows_count: Some(0), partitions: Some( - vec![FieldSummary { contains_null: false, contains_nan: Some(false), lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: Some(Datum::float(2.1_f32).to_bytes().unwrap())}] + vec![FieldSummary { contains_null: false, contains_nan: None, lower_bound: Some(Datum::float(1.1_f32).to_bytes().unwrap()), upper_bound: None}] ), - key_metadata: None, + key_metadata: Some(vec![1, 2, 3]), first_row_id: Some(13), } ]