Skip to content

perf: parse manifest list entries without buffering Avro values #3224

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kacper-li-airspace-intelligence wants to merge 6 commits into
base: main
Choose a base branch
from
Open

perf: parse manifest list entries without buffering Avro values #3224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
2d95153
perf: parse manifest list entries without buffering Avro values
kacper-li-airspace-intelligence Sep 15, 2026
b124830
Merge branch 'main' into reduce-manifest-list-parser-memory
kacper-li-airspace-intelligence Oct 6, 2026
e448f0f
Merge branch 'main' into reduce-manifest-list-parser-memory
kacper-li-airspace-intelligence Oct 13, 2026
3db218e
Move manifest list parser tests into unit tests
kacper-li-airspace-intelligence Oct 13, 2026
7015b31
Fold manifest list round-trip cases into existing tests
kacper-li-airspace-intelligence Oct 14, 2026
2002867
Merge remote-tracking branch 'origin/main' into reduce-manifest-list-…
kacper-li-airspace-intelligence Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 6 additions & 39 deletions crates/iceberg/src/spec/manifest_list/_serde.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,37 +27,28 @@ use crate::Error;
use crate::error::{Result, invalid_data};
use crate::spec::FieldSummary;

#[cfg(test)]
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(transparent)]
pub(crate) struct ManifestListV3 {
entries: Vec<ManifestFileV3>,
}

#[cfg(test)]
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(transparent)]
pub(crate) struct ManifestListV2 {
entries: Vec<ManifestFileV2>,
}

#[cfg(test)]
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(transparent)]
pub(crate) struct ManifestListV1 {
entries: Vec<ManifestFileV1>,
}

impl ManifestListV3 {
/// Converts the [ManifestListV3] into a [ManifestList].
pub fn try_into(self) -> Result<super::ManifestList> {
Ok(super::ManifestList {
entries: self
.entries
.into_iter()
.map(|v| v.try_into())
.collect::<Result<Vec<_>>>()?,
})
}
}

#[cfg(test)]
impl TryFrom<super::ManifestList> for ManifestListV3 {
type Error = Error;

Expand All @@ -72,19 +63,7 @@ impl TryFrom<super::ManifestList> for ManifestListV3 {
}
}

impl ManifestListV2 {
/// Converts the [ManifestListV2] into a [ManifestList].
pub fn try_into(self) -> Result<super::ManifestList> {
Ok(super::ManifestList {
entries: self
.entries
.into_iter()
.map(|v| v.try_into())
.collect::<Result<Vec<_>>>()?,
})
}
}

#[cfg(test)]
impl TryFrom<super::ManifestList> for ManifestListV2 {
type Error = Error;

Expand All @@ -99,19 +78,7 @@ impl TryFrom<super::ManifestList> for ManifestListV2 {
}
}

impl ManifestListV1 {
/// Converts the [ManifestListV1] into a [ManifestList].
pub fn try_into(self) -> Result<super::ManifestList> {
Ok(super::ManifestList {
entries: self
.entries
.into_iter()
.map(|v| v.try_into())
.collect::<Result<Vec<_>>>()?,
})
}
}

#[cfg(test)]
impl TryFrom<super::ManifestList> for ManifestListV1 {
type Error = Error;

Expand Down
115 changes: 88 additions & 27 deletions crates/iceberg/src/spec/manifest_list/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,14 @@ mod manifest_file;
mod reader;
mod writer;

use apache_avro::types::Value;
use apache_avro::{Reader, from_value};
use apache_avro::Reader;
pub use manifest_file::*;
pub use reader::*;
pub use serde_bytes::ByteBuf;
pub use writer::*;

use self::_const_schema::MANIFEST_LIST_AVRO_SCHEMA_V1;
use super::FormatVersion;
use crate::avro::Resolved;
use crate::error::Result;

/// Placeholder for sequence number. The field with this value must be replaced with the actual sequence number before it write.
Expand Down Expand Up @@ -59,25 +58,22 @@ pub struct ManifestList {
impl ManifestList {
/// Parse manifest list from bytes.
pub fn parse_with_version(bs: &[u8], version: FormatVersion) -> Result<ManifestList> {
match version {
FormatVersion::V1 => {
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()
}
FormatVersion::V2 => {
let reader = Reader::new(bs)?;
let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
from_value::<_serde::ManifestListV2>(&values)?.try_into()
}
FormatVersion::V3 => {
let reader = Reader::new(bs)?;
let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
from_value::<_serde::ManifestListV3>(&values)?.try_into()
}
}
let reader = Reader::new(bs)?;
let entries = match version {
FormatVersion::V1 => reader
.into_deser_iter::<Resolved<_serde::ManifestFileV1>>()
.map(|entry| entry?.0.try_into())
.collect::<Result<Vec<_>>>()?,
Comment on lines +63 to +66

@mbutrovich mbutrovich Oct 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java writes V1 manifest lists with the field names added_files_count, existing_files_count and deleted_files_count (see V1Metadata.MANIFEST_LIST_SCHEMA and ManifestFile), which are the names in the spec's manifest list fields. ManifestFileV1 only knows the older added_data_files_count names and has no aliases. So for a V1 list written by current Java, those three counts parse as None.

I checked this by writing a one-entry list with Java's V1 Avro schema and parsing it as V1. added_files_count and existing_files_count come back None, while added_rows_count (same name in both) comes back Some(9). main at b154603a returns the same thing, so this isn't from this PR. Before, resolution against MANIFEST_LIST_AVRO_SCHEMA_V1 filled added_data_files_count with its null default, so serde never saw Java's names. Now that V1 reads against the writer schema, the #[serde(alias = ...)] approach ManifestFileV2 already uses works for V1 too. Adding the three aliases to ManifestFileV1 returns Some(3) and Some(0) for that list.

The spec treats a null count as non-zero, so planning stays correct, but the counts are lost. As I read existing_manifest, a fast append on such a table copies these entries into the new manifest list, so the nulls get written back. Since this is a separate bug, could you open an issue for it? The fix plus a test with Java's V1 names would make a small follow-up PR.

@kacper-li-airspace-intelligence kacper-li-airspace-intelligence Oct 14, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great find, opened #3445

FormatVersion::V2 => reader
.into_deser_iter::<Resolved<_serde::ManifestFileV2>>()
.map(|entry| entry?.0.try_into())
.collect::<Result<Vec<_>>>()?,
FormatVersion::V3 => reader
.into_deser_iter::<Resolved<_serde::ManifestFileV3>>()
.map(|entry| entry?.0.try_into())
.collect::<Result<Vec<_>>>()?,
};
Ok(ManifestList { entries })
}
Comment on lines 59 to 77

@mbutrovich mbutrovich Oct 9, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#3354 merged today. It upgraded to apache-avro 0.22 and reads manifest entries with Reader::into_deser_iter, which decodes straight into the serde types without building an apache_avro::types::Value. The Resolved adapter it added handles writer record names, unions, and numeric widening against the writer schema.

What do you think about rebasing and reading manifest lists the same way as the manifest entry reader? Something like:

    /// Parse manifest list from bytes.
    pub fn parse_with_version(bs: &[u8], version: FormatVersion) -> Result<ManifestList> {
        let reader = Reader::new(bs)?;
        let entries = match version {
            FormatVersion::V1 => reader
                .into_deser_iter::<Resolved<_serde::ManifestFileV1>>()
                .map(|entry| entry?.0.try_into())
                .collect::<Result<Vec<_>>>()?,
            FormatVersion::V2 => reader
                .into_deser_iter::<Resolved<_serde::ManifestFileV2>>()
                .map(|entry| entry?.0.try_into())
                .collect::<Result<Vec<_>>>()?,
            FormatVersion::V3 => reader
                .into_deser_iter::<Resolved<_serde::ManifestFileV3>>()
                .map(|entry| entry?.0.try_into())
                .collect::<Result<Vec<_>>>()?,
        };
        Ok(ManifestList { entries })
    }

This removes the per-record Value as well as the whole-list one. With it, your manifest_list_memory test reports a peak of 4,355,177 bytes for the uncompressed list on my machine, against 6,040,920 at the head commit. The field aliases and V1 defaults on ManifestFileV2 and ManifestFileV3 are serde attributes, so they carry over. V1 would read against the writer schema instead of MANIFEST_LIST_AVRO_SCHEMA_V1, as the manifest entry reader already does for V1 manifests. The two doc lines about converting each Avro value would go too.

This is also the "Read manifest lists without Value" task in #3262.

@kacper-li-airspace-intelligence kacper-li-airspace-intelligence Oct 13, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. I merged main and use your into_deser_iter version. Peak is now 4,355,177 bytes, against 31,194,672 on main.

@mbutrovich mbutrovich Oct 13, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. I merged main and use your into_deser_iter version. Peak is now 4,355,177 bytes, against 31,194,672 on main.

Awesome, and thank you for updating the PR description with the new numbers!


/// Get the entries in the manifest list.
Expand Down Expand Up @@ -123,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,
}
]
Expand Down Expand Up @@ -159,6 +155,71 @@ mod test {
assert_eq!(manifest_list, parsed_manifest_list);
}

const VERSIONS: [FormatVersion; 3] = [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3];

fn entry(version: FormatVersion) -> ManifestFile {
ManifestFile {
manifest_path: "s3://test/metadata/manifest.avro".to_string(),
manifest_length: 1234,
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 },
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),
}
}

async fn encode(version: FormatVersion, entries: Vec<ManifestFile>) -> Vec<u8> {
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_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, 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_v2() {
let manifest_list = ManifestList {
Expand Down Expand Up @@ -198,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,
}
]
Expand Down Expand Up @@ -316,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),
}
]
Expand Down
Loading