Repository navigation
perf: parse manifest list entries without buffering Avro values - #3224
kacper-li-airspace-intelligence wants to merge 6 commits into
Conversation
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @kacper-li-airspace-intelligence for this, and for the allocation measurements in the description.
| /// 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<ManifestList> { | ||
| match version { | ||
| FormatVersion::V1 => { | ||
| let reader = Reader::with_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1, bs)?; | ||
| let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?); | ||
| from_value::<_serde::ManifestListV1>(&values)?.try_into() | ||
| let entries = reader | ||
| .map(|value| { | ||
| let value = value?; | ||
| from_value::<_serde::ManifestFileV1>(&value)?.try_into() | ||
| }) | ||
| .collect::<Result<Vec<_>>>()?; | ||
| Ok(ManifestList { entries }) | ||
| } | ||
| 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() | ||
| let entries = reader | ||
| .map(|value| { | ||
| let value = value?; | ||
| from_value::<_serde::ManifestFileV2>(&value)?.try_into() | ||
| }) | ||
| .collect::<Result<Vec<_>>>()?; | ||
| Ok(ManifestList { entries }) | ||
| } | ||
| 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 entries = reader | ||
| .map(|value| { | ||
| let value = value?; | ||
| from_value::<_serde::ManifestFileV3>(&value)?.try_into() | ||
| }) | ||
| .collect::<Result<Vec<_>>>()?; | ||
| Ok(ManifestList { entries }) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
#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.
There was a problem hiding this comment.
Done. I merged main and use your into_deser_iter version. Peak is now 4,355,177 bytes, against 31,194,672 on main.
There was a problem hiding this comment.
Done. I merged
mainand use yourinto_deser_iterversion. Peak is now 4,355,177 bytes, against 31,194,672 onmain.
Awesome, and thank you for updating the PR description with the new numbers!
| 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; |
There was a problem hiding this comment.
This would be the first integration test in crates/iceberg. It brings a new test binary and an unsafe global allocator, and it folds the format checks into one test so they can share the allocator. Could the round trips, the empty lists, the truncated input, and the V1 file read as V2 move into the unit tests in spec/manifest_list/mod.rs, next to test_parse_manifest_list_v1? Those cases aren't covered there today, and as separate tests a failure in one wouldn't hide the others.
For the allocation budget, what do you think about leaving it out of the suite and keeping the before and after numbers in the PR description? The test also times the parse with Instant and prints it without checking it.
There was a problem hiding this comment.
I moved the round trip, empty, truncated and V1-as-V2 cases into separate unit tests in spec/manifest_list/mod.rs. I removed the integration test. The before/after numbers are now in the description.
There was a problem hiding this comment.
Those cases aren't covered there today, and as separate tests a failure in one wouldn't hide the others.
I got part of this wrong, sorry for the churn. test_parse_manifest_list_v1, test_parse_manifest_list_v2 and test_parse_manifest_list_v3 in the same module already write a list with ManifestListWriter and parse it back. test_manifest_list_writer_v1_as_v2 in writer.rs writes a V1 list and parses it as V2, which is what test_parse_manifest_list_v1_as_v2 checks. Only the empty and truncated cases are new.
What do you think about dropping test_parse_manifest_list_round_trip and test_parse_manifest_list_v1_as_v2? The round trip does add inputs the existing tests don't have: a non-empty key_metadata, and a null contains_nan and upper_bound, which go through the union handling in Resolved. Setting those on an entry in the existing test_parse_manifest_list_v* tests would keep that coverage without a second set of round trips.
There was a problem hiding this comment.
Thanks for pointing it out I missed this. I folded the tests.
Resolve the parse_with_version conflict by reading entries with into_deser_iter and the Resolved adapter from apache#3354.
Cover round trips, empty lists, truncated input and V1 read as V2 as separate unit tests, and drop the allocation-budget integration test.
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @kacper-li-airspace-intelligence for switching to into_deser_iter and for moving the tests so quickly. The parser change reads well to me.
| FormatVersion::V1 => reader | ||
| .into_deser_iter::<Resolved<_serde::ManifestFileV1>>() | ||
| .map(|entry| entry?.0.try_into()) | ||
| .collect::<Result<Vec<_>>>()?, |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Great find, opened #3445
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.
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @kacper-li-airspace-intelligence for folding the tests and for opening #3445 so quickly. I have nothing new on the code beyond the existing threads.
Could you add Part of #3262 under "Which issue does this PR close?" in the description? The PR template asks for a linked issue because it feeds the release changelog, and this PR is the "Read manifest lists without Value" task there.
|
@mbutrovich thank you for all your comments and the review. I linked the issue. |
Which issue does this PR close?
Task
Read manifest lists without ValueWhat changes are included in this PR?
ManifestList::parse_with_versioncollects the whole list into aValue::Arraybefore converting it, so the generic and typed representations coexist.FastAppendAction::existing_manifestloads the current manifest list on every append, so this cost grows with the table, not the append.Read each entry with
Reader::into_deser_iterthrough theResolvedadapter from #3354, as the manifest entry reader does. NoValueis built, for the list or per record. Field aliases and V1 defaults onManifestFileV2/V3are serde attributes and carry over. V1 now reads against the writer schema instead ofMANIFEST_LIST_AVRO_SCHEMA_V1, as V1 manifests already do.Parsing still returns an error without a partial list. The unused whole-list conversion methods are removed; the serialization wrappers used by existing tests stay behind
cfg(test).Are these changes tested?
New unit tests in
spec/manifest_list/mod.rscover empty and truncated V1/V2/V3 lists. The existingtest_parse_manifest_list_v*tests now include a non-emptykey_metadataand nullcontains_nan/upper_bound. Allicebergunit tests, clippy and fmt pass.Parsing a 10,000-entry uncompressed V2 list (692,273 bytes), measured with a counting global allocator outside the suite, release mode, aarch64 macOS:
mainatb154603aAI Disclosure
AI assisted with the implementation, test and description.