You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by li...@apache.org on 2026/08/20 10:19:18 UTC
(iceberg-rust) branch main updated: refactor: Add manifest reader (#2884)
This is an automated email from the ASF dual-hosted git repository.
blackmwk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1ecd49777 refactor: Add manifest reader (#2884)
1ecd49777 is described below
commit 1ecd497774612405f0af0dc9e1dc2bd6da3c3311
Author: Xander <za...@googlemail.com>
AuthorDate: Thu Aug 20 11:19:13 2026 +0100
refactor: Add manifest reader (#2884)
* Refactor manifest reader into its own struct
* pub api
* make crate pub
* remove doc
* remove load method
* fix up
---
crates/iceberg/public-api.txt | 6 +-
crates/iceberg/src/catalog/utils.rs | 4 +-
crates/iceberg/src/io/object_cache.rs | 12 +-
crates/iceberg/src/spec/manifest/mod.rs | 2 +
crates/iceberg/src/spec/manifest/reader.rs | 424 +++++++++++++++++++++
.../src/spec/manifest_list/manifest_file.rs | 225 +----------
crates/iceberg/src/table.rs | 9 +-
crates/iceberg/src/transaction/append.rs | 17 +-
crates/iceberg/src/transaction/snapshot.rs | 7 +-
.../datafusion/src/physical_plan/commit.rs | 5 +-
10 files changed, 474 insertions(+), 237 deletions(-)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index 77b736b97..a023ad515 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -2100,8 +2100,6 @@ impl iceberg::spec::ManifestFile
pub fn iceberg::spec::ManifestFile::has_added_files(&self) -> bool
pub fn iceberg::spec::ManifestFile::has_deleted_files(&self) -> bool
pub fn iceberg::spec::ManifestFile::has_existing_files(&self) -> bool
-impl iceberg::spec::ManifestFile
-pub async fn iceberg::spec::ManifestFile::load_manifest(&self, file_io: &iceberg::io::FileIO) -> iceberg::Result<iceberg::spec::Manifest>
impl core::clone::Clone for iceberg::spec::ManifestFile
pub fn iceberg::spec::ManifestFile::clone(&self) -> iceberg::spec::ManifestFile
impl core::cmp::Eq for iceberg::spec::ManifestFile
@@ -2160,6 +2158,9 @@ pub fn iceberg::spec::ManifestMetadata::fmt(&self, f: &mut core::fmt::Formatter<
impl core::marker::StructuralPartialEq for iceberg::spec::ManifestMetadata
impl iceberg::spec::ManifestMetadata
pub fn iceberg::spec::ManifestMetadata::builder() -> ManifestMetadataBuilder<((), (), (), (), ())>
+pub struct iceberg::spec::ManifestReader
+impl iceberg::spec::ManifestReader
+pub async fn iceberg::spec::ManifestReader::read(self, manifest_file: &iceberg::spec::ManifestFile) -> iceberg::Result<iceberg::spec::Manifest>
pub struct iceberg::spec::ManifestWriter
impl iceberg::spec::ManifestWriter
pub fn iceberg::spec::ManifestWriter::add_delete_file(&mut self, data_file: iceberg::spec::DataFile, sequence_number: i64, file_sequence_number: core::option::Option<i64>) -> iceberg::Result<()>
@@ -3215,6 +3216,7 @@ pub fn iceberg::table::Table::file_io(&self) -> &iceberg::io::FileIO
pub fn iceberg::table::Table::identifier(&self) -> &iceberg::TableIdent
pub fn iceberg::table::Table::inspect(&self) -> iceberg::inspect::MetadataTable<'_>
pub fn iceberg::table::Table::manifest_list_reader(&self, snapshot: &iceberg::spec::SnapshotRef) -> iceberg::spec::ManifestListReader
+pub fn iceberg::table::Table::manifest_reader(&self) -> iceberg::spec::ManifestReader
pub fn iceberg::table::Table::metadata(&self) -> &iceberg::spec::TableMetadata
pub fn iceberg::table::Table::metadata_location(&self) -> core::option::Option<&str>
pub fn iceberg::table::Table::metadata_location_result(&self) -> iceberg::Result<&str>
diff --git a/crates/iceberg/src/catalog/utils.rs b/crates/iceberg/src/catalog/utils.rs
index 8e743e7d7..70c94dcfb 100644
--- a/crates/iceberg/src/catalog/utils.rs
+++ b/crates/iceberg/src/catalog/utils.rs
@@ -23,7 +23,7 @@ use futures::{TryStreamExt, stream};
use crate::Result;
use crate::io::FileIO;
-use crate::spec::ManifestFile;
+use crate::spec::{ManifestFile, ManifestReader};
use crate::table::Table;
const DELETE_CONCURRENCY: usize = 10;
@@ -111,7 +111,7 @@ async fn delete_data_files(
) -> Result<()> {
stream::iter(manifest_files.values().map(Ok))
.try_for_each_concurrent(DELETE_CONCURRENCY, |manifest_file| async move {
- let manifest = manifest_file.load_manifest(io).await?;
+ let manifest = ManifestReader::new(io.clone()).read(manifest_file).await?;
let data_file_paths = manifest
.entries()
.iter()
diff --git a/crates/iceberg/src/io/object_cache.rs b/crates/iceberg/src/io/object_cache.rs
index eafc65054..d780e60c1 100644
--- a/crates/iceberg/src/io/object_cache.rs
+++ b/crates/iceberg/src/io/object_cache.rs
@@ -21,8 +21,8 @@ use std::sync::Arc;
use crate::encryption::EncryptionManager;
use crate::io::FileIO;
use crate::spec::{
- FormatVersion, Manifest, ManifestFile, ManifestList, ManifestListReader, SchemaId, SnapshotRef,
- TableMetadataRef,
+ FormatVersion, Manifest, ManifestFile, ManifestList, ManifestListReader, ManifestReader,
+ SchemaId, SnapshotRef, TableMetadataRef,
};
use crate::{Error, ErrorKind, Result};
@@ -103,8 +103,8 @@ impl ObjectCache {
/// or retrieves one from FileIO and parses it if not present
pub(crate) async fn get_manifest(&self, manifest_file: &ManifestFile) -> Result<Arc<Manifest>> {
if self.cache_disabled {
- return manifest_file
- .load_manifest(&self.file_io)
+ return ManifestReader::new(self.file_io.clone())
+ .read(manifest_file)
.await
.map(Arc::new);
}
@@ -187,7 +187,9 @@ impl ObjectCache {
}
async fn fetch_and_parse_manifest(&self, manifest_file: &ManifestFile) -> Result<CachedItem> {
- let manifest = manifest_file.load_manifest(&self.file_io).await?;
+ let manifest = ManifestReader::new(self.file_io.clone())
+ .read(manifest_file)
+ .await?;
Ok(CachedItem::Manifest(Arc::new(manifest)))
}
diff --git a/crates/iceberg/src/spec/manifest/mod.rs b/crates/iceberg/src/spec/manifest/mod.rs
index 308a4738c..9156914aa 100644
--- a/crates/iceberg/src/spec/manifest/mod.rs
+++ b/crates/iceberg/src/spec/manifest/mod.rs
@@ -23,6 +23,8 @@ mod entry;
pub use entry::*;
mod metadata;
pub use metadata::*;
+mod reader;
+pub use reader::*;
mod writer;
use std::sync::Arc;
diff --git a/crates/iceberg/src/spec/manifest/reader.rs b/crates/iceberg/src/spec/manifest/reader.rs
new file mode 100644
index 000000000..499ff8fc2
--- /dev/null
+++ b/crates/iceberg/src/spec/manifest/reader.rs
@@ -0,0 +1,424 @@
+// 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.
+
+use super::Manifest;
+use crate::encryption::{EncryptedInputFile, StandardKeyMetadata};
+use crate::error::Result;
+use crate::io::FileIO;
+use crate::spec::{ManifestContentType, ManifestEntry, ManifestFile};
+use crate::{Error, ErrorKind};
+
+/// Reads a manifest file referenced by a manifest list entry, transparently
+/// decrypting it when the entry records key metadata.
+pub struct ManifestReader {
+ file_io: FileIO,
+}
+
+impl ManifestReader {
+ /// Create a manifest reader.
+ pub(crate) fn new(file_io: FileIO) -> Self {
+ Self { file_io }
+ }
+
+ /// Read, decrypt, parse and return the manifest described by
+ /// `manifest_file`.
+ pub async fn read(self, manifest_file: &ManifestFile) -> Result<Manifest> {
+ let input_file = self.file_io.new_input(&manifest_file.manifest_path)?;
+ let key_metadata = manifest_file
+ .key_metadata
+ .as_deref()
+ .map(StandardKeyMetadata::decode)
+ .transpose()?;
+ let bytes = match key_metadata {
+ Some(key_metadata) => {
+ EncryptedInputFile::new(input_file, key_metadata)
+ .read()
+ .await?
+ }
+ None => input_file.read().await?,
+ };
+
+ let (metadata, mut entries) = Manifest::try_from_avro_bytes(&bytes)?;
+
+ for entry in &mut entries {
+ entry.inherit_data(manifest_file);
+ }
+
+ self.assign_first_row_ids(manifest_file, &mut entries)?;
+
+ Ok(Manifest::new(metadata, entries))
+ }
+
+ /// Assigns `first_row_id` to the live data-file entries, following the
+ /// row-lineage inheritance rules in
+ /// <https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
+ ///
+ /// With a manifest-level `first_row_id`, each live entry lacking one is
+ /// assigned the running id, which then advances by that entry's record
+ /// count; entries that already carry a `first_row_id` keep it and do not
+ /// advance the counter. Without a manifest-level `first_row_id`, any
+ /// inherited per-entry value is cleared so callers never observe a stale id.
+ fn assign_first_row_ids(
+ &self,
+ manifest_file: &ManifestFile,
+ entries: &mut [ManifestEntry],
+ ) -> Result<()> {
+ // A `first_row_id` is only valid on data manifests. Delete files always
+ // have a null `first_row_id`, so there is nothing to assign or clear; a
+ // stray value on a delete manifest is a spec violation by the writer,
+ // which we surface without failing the read.
+ if manifest_file.content != ManifestContentType::Data {
+ if let Some(manifest_first_row_id) = manifest_file.first_row_id {
+ tracing::warn!(
+ "Ignoring first_row_id {manifest_first_row_id} on delete manifest {}",
+ manifest_file.manifest_path
+ );
+ }
+
+ return Ok(());
+ }
+
+ let Some(manifest_first_row_id) = manifest_file.first_row_id else {
+ // A data manifest with no manifest-level `first_row_id` predates row
+ // lineage; clear any per-entry value inherited from an earlier read.
+ for entry in entries {
+ entry.data_file.first_row_id = None;
+ }
+
+ return Ok(());
+ };
+
+ let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid first_row_id: {manifest_first_row_id} (exceeds i64::MAX)"),
+ )
+ })?;
+
+ for entry in entries {
+ if !entry.is_alive() {
+ continue;
+ }
+
+ if entry.data_file.first_row_id.is_none() {
+ let file_first_row_id = next_row_id;
+ entry.data_file.first_row_id = Some(file_first_row_id);
+ let record_count = entry.data_file.record_count;
+ next_row_id = file_first_row_id.checked_add_unsigned(record_count).ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Row ID overflow assigning first_row_id in {}. File first_row_id: {file_first_row_id}, record count: {record_count}",
+ manifest_file.manifest_path
+ ),
+ )
+ })?;
+ }
+ }
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use super::*;
+ use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata};
+ use crate::io::FileIO;
+ use crate::spec::{
+ DataContentType, DataFile, DataFileBuilder, DataFileFormat, ManifestEntry, ManifestStatus,
+ ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, Struct, Type,
+ };
+
+ #[tokio::test]
+ async fn test_read_plaintext_manifest_inherits_entries() {
+ let schema = test_schema();
+ let partition_spec = PartitionSpec::builder(schema.clone())
+ .with_spec_id(0)
+ .build()
+ .unwrap();
+
+ let io = FileIO::new_with_memory();
+ let path = "memory:///table/metadata/plain.avro";
+ let mut writer = ManifestWriterBuilder::new(
+ io.new_output(path).unwrap(),
+ Some(1),
+ schema.clone(),
+ partition_spec,
+ )
+ .build_v2_data();
+ writer.add_entry(test_entry()).unwrap();
+ // Writing the manifest yields the manifest list entry describing it.
+ let manifest_file = writer.write_manifest_file().await.unwrap();
+
+ let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap();
+ assert_eq!(manifest.entries().len(), 1);
+ assert_eq!(
+ manifest.entries()[0].data_file().file_path(),
+ "memory:///table/data/00000.parquet"
+ );
+ // Entries must inherit values from the manifest list entry.
+ assert_eq!(
+ manifest.entries()[0].sequence_number(),
+ Some(manifest_file.sequence_number)
+ );
+ assert_eq!(
+ manifest.entries()[0].snapshot_id(),
+ Some(manifest_file.added_snapshot_id)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_read_encrypted_manifest_roundtrip() {
+ let schema = test_schema();
+ let partition_spec = PartitionSpec::builder(schema.clone())
+ .with_spec_id(0)
+ .build()
+ .unwrap();
+
+ let io = FileIO::new_with_memory();
+ let path = "memory:///table/metadata/encrypted.avro";
+ let encrypted_output =
+ EncryptedOutputFile::new(io.new_output(path).unwrap(), key_metadata());
+
+ let mut writer = ManifestWriterBuilder::new_from_encrypted(
+ encrypted_output,
+ Some(1),
+ schema.clone(),
+ partition_spec,
+ )
+ .unwrap()
+ .build_v3_data();
+ writer.add_entry(test_entry()).unwrap();
+ // The returned manifest list entry records the key metadata.
+ let manifest_file = writer.write_manifest_file().await.unwrap();
+ assert!(manifest_file.key_metadata.is_some());
+
+ // Reading with the recorded key metadata must recover the entry.
+ let manifest = ManifestReader::new(io.clone())
+ .read(&manifest_file)
+ .await
+ .unwrap();
+ assert_eq!(manifest.entries().len(), 1);
+ assert_eq!(
+ manifest.entries()[0].data_file().file_path(),
+ "memory:///table/data/00000.parquet"
+ );
+
+ // Without the key metadata the encrypted bytes must not read as plaintext.
+ let mut plaintext_entry = manifest_file.clone();
+ plaintext_entry.key_metadata = None;
+ assert!(
+ ManifestReader::new(io)
+ .read(&plaintext_entry)
+ .await
+ .is_err(),
+ "encrypted manifest must not parse as plaintext"
+ );
+ }
+
+ fn test_schema() -> Arc<Schema> {
+ Arc::new(
+ Schema::builder()
+ .with_fields(vec![Arc::new(NestedField::optional(
+ 1,
+ "id",
+ Type::Primitive(PrimitiveType::Long),
+ ))])
+ .build()
+ .unwrap(),
+ )
+ }
+
+ fn test_entry() -> ManifestEntry {
+ ManifestEntry {
+ status: ManifestStatus::Added,
+ snapshot_id: None,
+ sequence_number: None,
+ file_sequence_number: None,
+ data_file: DataFile {
+ content: DataContentType::Data,
+ file_path: "memory:///table/data/00000.parquet".to_string(),
+ file_format: DataFileFormat::Parquet,
+ partition: Struct::empty(),
+ record_count: 1,
+ file_size_in_bytes: 4096,
+ 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,
+ },
+ }
+ }
+
+ fn key_metadata() -> StandardKeyMetadata {
+ StandardKeyMetadata::try_new(b"0123456789abcdef").unwrap()
+ }
+
+ /// A reader with an unused in-memory `FileIO`, for exercising the pure
+ /// row-id assignment logic without touching storage.
+ fn test_reader() -> ManifestReader {
+ ManifestReader::new(FileIO::new_with_memory())
+ }
+
+ /// Builds a data-file manifest entry with the given status, record count,
+ /// and pre-existing `first_row_id`.
+ fn data_entry(
+ status: ManifestStatus,
+ record_count: u64,
+ first_row_id: Option<i64>,
+ ) -> ManifestEntry {
+ let data_file = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("s3://bucket/table/data/00000.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(4096)
+ .record_count(record_count)
+ .first_row_id(first_row_id)
+ .build()
+ .unwrap();
+
+ ManifestEntry::builder()
+ .status(status)
+ .data_file(data_file)
+ .build()
+ }
+
+ /// Builds a manifest file with the given content type and manifest-level
+ /// `first_row_id`. Other fields are irrelevant to row-id assignment.
+ fn manifest_file(content: ManifestContentType, first_row_id: Option<u64>) -> ManifestFile {
+ ManifestFile {
+ manifest_path: "memory:///m.avro".to_string(),
+ manifest_length: 0,
+ partition_spec_id: 0,
+ content,
+ sequence_number: 0,
+ min_sequence_number: 0,
+ added_snapshot_id: 0,
+ added_files_count: None,
+ existing_files_count: None,
+ deleted_files_count: None,
+ added_rows_count: None,
+ existing_rows_count: None,
+ deleted_rows_count: None,
+ partitions: None,
+ key_metadata: None,
+ first_row_id,
+ }
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_interleaved() {
+ let manifest = manifest_file(ManifestContentType::Data, Some(10));
+ let mut entries = vec![
+ data_entry(ManifestStatus::Added, 3, None),
+ // A pre-assigned entry between two assigned ones: it keeps its id and
+ // must not advance the running counter.
+ data_entry(ManifestStatus::Added, 5, Some(100)),
+ // A deleted entry with a pre-set id: it is skipped, so the id is
+ // preserved verbatim and does not advance the counter.
+ data_entry(ManifestStatus::Deleted, 7, Some(999)),
+ data_entry(ManifestStatus::Existing, 2, None),
+ ];
+
+ test_reader()
+ .assign_first_row_ids(&manifest, &mut entries)
+ .unwrap();
+
+ assert_eq!(entries[0].data_file.first_row_id, Some(10));
+ assert_eq!(entries[1].data_file.first_row_id, Some(100));
+ assert_eq!(entries[2].data_file.first_row_id, Some(999));
+ // 10 + 3 = 13; the preserved and deleted entries in between do not move it.
+ assert_eq!(entries[3].data_file.first_row_id, Some(13));
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_clears_without_manifest_first_row_id() {
+ // A data manifest with no manifest-level first_row_id predates row
+ // lineage: any per-entry value inherited from an earlier read is cleared
+ // so callers never observe a stale id.
+ let manifest = manifest_file(ManifestContentType::Data, None);
+ let mut entries = vec![
+ data_entry(ManifestStatus::Added, 3, None),
+ data_entry(ManifestStatus::Existing, 5, Some(100)),
+ ];
+
+ test_reader()
+ .assign_first_row_ids(&manifest, &mut entries)
+ .unwrap();
+
+ assert_eq!(entries[0].data_file.first_row_id, None);
+ assert_eq!(entries[1].data_file.first_row_id, None);
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_ignores_delete_manifest() {
+ // A stray first_row_id on a delete manifest is a writer-side spec
+ // violation; the read ignores it rather than failing, and does not
+ // assign ids to the entries.
+ let manifest = manifest_file(ManifestContentType::Deletes, Some(10));
+ let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
+
+ test_reader()
+ .assign_first_row_ids(&manifest, &mut entries)
+ .unwrap();
+
+ assert_eq!(entries[0].data_file.first_row_id, None);
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_rejects_oversized_manifest_first_row_id() {
+ // A manifest-level first_row_id above i64::MAX cannot be represented as the
+ // signed running counter and must be rejected.
+ let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64 + 1));
+ let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
+
+ let err = test_reader()
+ .assign_first_row_ids(&manifest, &mut entries)
+ .expect_err("an oversized manifest first_row_id must be rejected");
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("Invalid first_row_id"));
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_rejects_counter_overflow() {
+ // Advancing the running counter past i64::MAX must be rejected rather than
+ // wrapping to a negative value that would corrupt subsequent assignments.
+ let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64));
+ let mut entries = vec![data_entry(ManifestStatus::Added, 1, None)];
+
+ let err = test_reader()
+ .assign_first_row_ids(&manifest, &mut entries)
+ .expect_err("counter overflow past i64::MAX must be rejected");
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("Row ID overflow"));
+ }
+}
diff --git a/crates/iceberg/src/spec/manifest_list/manifest_file.rs b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
index a05377a02..567e78e3a 100644
--- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs
+++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
@@ -20,10 +20,7 @@ use std::str::FromStr;
use serde_derive::{Deserialize, Serialize};
use super::ByteBuf;
-use crate::encryption::{EncryptedInputFile, StandardKeyMetadata};
use crate::error::Result;
-use crate::io::FileIO;
-use crate::spec::{Manifest, ManifestEntry};
use crate::{Error, ErrorKind};
/// Entry in a manifest list.
@@ -174,99 +171,6 @@ impl TryFrom<i32> for ManifestContentType {
}
}
-impl ManifestFile {
- /// Load [`Manifest`].
- ///
- /// This method will also initialize inherited values of [`ManifestEntry`](crate::spec::ManifestEntry), such as `sequence_number`.
- pub async fn load_manifest(&self, file_io: &FileIO) -> Result<Manifest> {
- let input = file_io.new_input(&self.manifest_path)?;
- let avro = match &self.key_metadata {
- Some(key_metadata_bytes) => {
- let key_metadata = StandardKeyMetadata::decode(key_metadata_bytes)?;
- EncryptedInputFile::new(input, key_metadata).read().await?
- }
- None => input.read().await?,
- };
-
- let (metadata, mut entries) = Manifest::try_from_avro_bytes(&avro)?;
-
- // Let entries inherit values from the manifest list entry.
- for entry in &mut entries {
- entry.inherit_data(self);
- }
-
- self.assign_first_row_ids(&mut entries)?;
-
- Ok(Manifest::new(metadata, entries))
- }
-
- /// Assigns `first_row_id` to the live data-file entries, following the
- /// row-lineage inheritance rules in
- /// <https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
- ///
- /// With a manifest-level `first_row_id`, each live entry lacking one is
- /// assigned the running id, which then advances by that entry's record
- /// count; entries that already carry a `first_row_id` keep it and do not
- /// advance the counter. Without a manifest-level `first_row_id`, any
- /// inherited per-entry value is cleared so callers never observe a stale id.
- fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) -> Result<()> {
- // A `first_row_id` is only valid on data manifests. Delete files always
- // have a null `first_row_id`, so there is nothing to assign or clear; a
- // stray value on a delete manifest is a spec violation by the writer,
- // which we surface without failing the read.
- if self.content != ManifestContentType::Data {
- if let Some(manifest_first_row_id) = self.first_row_id {
- tracing::warn!(
- "Ignoring first_row_id {manifest_first_row_id} on delete manifest {}",
- self.manifest_path
- );
- }
-
- return Ok(());
- }
-
- let Some(manifest_first_row_id) = self.first_row_id else {
- // A data manifest with no manifest-level `first_row_id` predates row
- // lineage; clear any per-entry value inherited from an earlier read.
- for entry in entries {
- entry.data_file.first_row_id = None;
- }
-
- return Ok(());
- };
-
- let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_| {
- Error::new(
- ErrorKind::DataInvalid,
- format!("Invalid first_row_id: {manifest_first_row_id} (exceeds i64::MAX)"),
- )
- })?;
-
- for entry in entries {
- if !entry.is_alive() {
- continue;
- }
-
- if entry.data_file.first_row_id.is_none() {
- let file_first_row_id = next_row_id;
- entry.data_file.first_row_id = Some(file_first_row_id);
- let record_count = entry.data_file.record_count;
- next_row_id = file_first_row_id.checked_add_unsigned(record_count).ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- format!(
- "Row ID overflow assigning first_row_id in {}. File first_row_id: {file_first_row_id}, record count: {record_count}",
- self.manifest_path
- ),
- )
- })?;
- }
- }
-
- Ok(())
- }
-}
-
/// Field summary for partition field in the spec.
///
/// Each field in the list corresponds to a field in the manifest file’s partition spec.
@@ -300,8 +204,9 @@ mod test {
use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata};
use crate::io::FileIO;
use crate::spec::{
- DataContentType, DataFileBuilder, DataFileFormat, ManifestEntry, ManifestStatus,
- ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type,
+ DataContentType, DataFileBuilder, DataFileFormat, ManifestEntry, ManifestReader,
+ ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema,
+ SchemaRef, Type,
};
#[test]
@@ -392,7 +297,7 @@ mod test {
let manifest_file = write_encrypted_manifest(&io, path, key_metadata).await;
assert_eq!(manifest_file.key_metadata, Some(encoded_key_metadata));
- let manifest = manifest_file.load_manifest(&io).await.unwrap();
+ let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(
manifest.entries()[0].file_path(),
@@ -420,10 +325,10 @@ mod test {
.with_aad_prefix(b"test-aad-prefix!");
manifest_file.key_metadata = Some(wrong_key_metadata.encode().unwrap().to_vec());
- let err = manifest_file
- .load_manifest(&io)
+ let err = ManifestReader::new(io)
+ .read(&manifest_file)
.await
- .expect_err("load_manifest must fail when decrypting with the wrong key");
+ .expect_err("read must fail when decrypting with the wrong key");
assert_eq!(err.kind(), ErrorKind::Unexpected);
}
@@ -445,10 +350,10 @@ mod test {
.with_aad_prefix(b"wrong-aad-prefix");
manifest_file.key_metadata = Some(wrong_aad_metadata.encode().unwrap().to_vec());
- let err = manifest_file
- .load_manifest(&io)
+ let err = ManifestReader::new(io)
+ .read(&manifest_file)
.await
- .expect_err("load_manifest must fail when decrypting with the wrong AAD prefix");
+ .expect_err("read must fail when decrypting with the wrong AAD prefix");
assert_eq!(err.kind(), ErrorKind::Unexpected);
}
@@ -475,110 +380,6 @@ mod test {
.build()
}
- /// Builds a manifest file with the given content type and manifest-level
- /// `first_row_id`. Other fields are irrelevant to row-id assignment.
- fn manifest_file(content: ManifestContentType, first_row_id: Option<u64>) -> ManifestFile {
- ManifestFile {
- manifest_path: "memory:///m.avro".to_string(),
- manifest_length: 0,
- partition_spec_id: 0,
- content,
- sequence_number: 0,
- min_sequence_number: 0,
- added_snapshot_id: 0,
- added_files_count: None,
- existing_files_count: None,
- deleted_files_count: None,
- added_rows_count: None,
- existing_rows_count: None,
- deleted_rows_count: None,
- partitions: None,
- key_metadata: None,
- first_row_id,
- }
- }
-
- #[test]
- fn test_assign_first_row_ids_interleaved() {
- let manifest = manifest_file(ManifestContentType::Data, Some(10));
- let mut entries = vec![
- data_entry(ManifestStatus::Added, 3, None),
- // A pre-assigned entry between two assigned ones: it keeps its id and
- // must not advance the running counter.
- data_entry(ManifestStatus::Added, 5, Some(100)),
- // A deleted entry with a pre-set id: it is skipped, so the id is
- // preserved verbatim and does not advance the counter.
- data_entry(ManifestStatus::Deleted, 7, Some(999)),
- data_entry(ManifestStatus::Existing, 2, None),
- ];
-
- manifest.assign_first_row_ids(&mut entries).unwrap();
-
- assert_eq!(entries[0].data_file.first_row_id, Some(10));
- assert_eq!(entries[1].data_file.first_row_id, Some(100));
- assert_eq!(entries[2].data_file.first_row_id, Some(999));
- // 10 + 3 = 13; the preserved and deleted entries in between do not move it.
- assert_eq!(entries[3].data_file.first_row_id, Some(13));
- }
-
- #[test]
- fn test_assign_first_row_ids_clears_without_manifest_first_row_id() {
- // A data manifest with no manifest-level first_row_id predates row
- // lineage: any per-entry value inherited from an earlier read is cleared
- // so callers never observe a stale id.
- let manifest = manifest_file(ManifestContentType::Data, None);
- let mut entries = vec![
- data_entry(ManifestStatus::Added, 3, None),
- data_entry(ManifestStatus::Existing, 5, Some(100)),
- ];
-
- manifest.assign_first_row_ids(&mut entries).unwrap();
-
- assert_eq!(entries[0].data_file.first_row_id, None);
- assert_eq!(entries[1].data_file.first_row_id, None);
- }
-
- #[test]
- fn test_assign_first_row_ids_ignores_delete_manifest() {
- // A stray first_row_id on a delete manifest is a writer-side spec
- // violation; the read ignores it rather than failing, and does not
- // assign ids to the entries.
- let manifest = manifest_file(ManifestContentType::Deletes, Some(10));
- let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
-
- manifest.assign_first_row_ids(&mut entries).unwrap();
-
- assert_eq!(entries[0].data_file.first_row_id, None);
- }
-
- #[test]
- fn test_assign_first_row_ids_rejects_oversized_manifest_first_row_id() {
- // A manifest-level first_row_id above i64::MAX cannot be represented as the
- // signed running counter and must be rejected.
- let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64 + 1));
- let mut entries = vec![data_entry(ManifestStatus::Added, 3, None)];
-
- let err = manifest
- .assign_first_row_ids(&mut entries)
- .expect_err("an oversized manifest first_row_id must be rejected");
- assert_eq!(err.kind(), ErrorKind::DataInvalid);
- assert!(err.message().contains("Invalid first_row_id"));
- }
-
- #[test]
- fn test_assign_first_row_ids_rejects_counter_overflow() {
- // Advancing the running counter past i64::MAX must be rejected rather than
- // wrapping to a negative value that would corrupt subsequent assignments.
- let manifest = manifest_file(ManifestContentType::Data, Some(i64::MAX as u64));
- let mut entries = vec![data_entry(ManifestStatus::Added, 1, None)];
-
- let err = manifest
- .assign_first_row_ids(&mut entries)
- .expect_err("counter overflow past i64::MAX must be rejected");
- assert_eq!(err.kind(), ErrorKind::DataInvalid);
- assert!(err.message().contains("Row ID overflow"));
- }
-
#[tokio::test]
async fn test_load_manifest_reads_written_entries() {
let io = FileIO::new_with_memory();
@@ -586,7 +387,7 @@ mod test {
let manifest_file = write_manifest(&io, path).await;
assert_eq!(manifest_file.key_metadata, None);
- let manifest = manifest_file.load_manifest(&io).await.unwrap();
+ let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(
manifest.entries()[0].file_path(),
@@ -597,7 +398,7 @@ mod test {
/// End-to-end: writing a v3 data manifest, stamping a manifest-level
/// `first_row_id`, and loading it must assign inherited `first_row_id`s to
- /// the entries. This exercises the wiring in `load_manifest` and the
+ /// the entries. This exercises the wiring in [`ManifestReader`] and the
/// write/read round-trip that leaves per-file `first_row_id` as `None`.
#[tokio::test]
async fn test_load_manifest_assigns_first_row_ids() {
@@ -608,7 +409,7 @@ mod test {
// Stamp a manifest-level first_row_id, as the manifest-list writer would.
manifest_file.first_row_id = Some(1000);
- let manifest = manifest_file.load_manifest(&io).await.unwrap();
+ let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(manifest.entries()[0].data_file().first_row_id(), Some(1000));
}
diff --git a/crates/iceberg/src/table.rs b/crates/iceberg/src/table.rs
index ecb523b68..21cd8681a 100644
--- a/crates/iceberg/src/table.rs
+++ b/crates/iceberg/src/table.rs
@@ -27,7 +27,9 @@ use crate::io::FileIO;
use crate::io::object_cache::ObjectCache;
use crate::runtime::Runtime;
use crate::scan::TableScanBuilder;
-use crate::spec::{ManifestListReader, SchemaRef, SnapshotRef, TableMetadata, TableMetadataRef};
+use crate::spec::{
+ ManifestListReader, ManifestReader, SchemaRef, SnapshotRef, TableMetadata, TableMetadataRef,
+};
use crate::{Error, ErrorKind, Result, TableIdent};
/// Builder to create table scan.
@@ -311,6 +313,11 @@ impl Table {
)
}
+ /// Creates a [`ManifestReader`] for loading manifests referenced by this table.
+ pub fn manifest_reader(&self) -> ManifestReader {
+ ManifestReader::new(self.file_io.clone())
+ }
+
/// Create a reader for the table.
pub fn reader_builder(&self) -> ArrowReaderBuilder {
ArrowReaderBuilder::new(self.file_io.clone(), self.runtime().clone())
diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs
index a19e297ac..872a1d89c 100644
--- a/crates/iceberg/src/transaction/append.rs
+++ b/crates/iceberg/src/transaction/append.rs
@@ -377,8 +377,9 @@ mod tests {
.await
.unwrap();
assert_eq!(1, manifest_list.entries().len());
- manifest_list.entries()[0]
- .load_manifest(table.file_io())
+ table
+ .manifest_reader()
+ .read(&manifest_list.entries()[0])
.await
.unwrap()
.entries()
@@ -430,7 +431,6 @@ mod tests {
.find(|m| m.added_files_count.unwrap_or(0) > 0)
.expect("new snapshot should carry the appended data manifest");
- // ManifestReader once it exists.
// The manifest list entry must carry decodable key metadata.
let key_metadata_bytes = manifest_file
.key_metadata
@@ -439,11 +439,11 @@ mod tests {
StandardKeyMetadata::decode(key_metadata_bytes)
.expect("recorded key metadata must decode as StandardKeyMetadata");
- // load_manifest self-decrypts using the recorded key metadata and must
- // recover the entry we appended. Because the read goes through
+ // The reader self-decrypts using the recorded key metadata and must
+ // recover the entry we appended. Because the read goes through the
// decryption path, this succeeding also proves the bytes
// on disk were genuinely encrypted (not silently written as plaintext).
- let manifest = manifest_file.load_manifest(table.file_io()).await.unwrap();
+ let manifest = table.manifest_reader().read(manifest_file).await.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(
manifest.entries()[0].data_file().file_path(),
@@ -903,8 +903,9 @@ mod tests {
);
// check manifest
- let manifest = manifest_list.entries()[0]
- .load_manifest(table.file_io())
+ let manifest = table
+ .manifest_reader()
+ .read(&manifest_list.entries()[0])
.await
.unwrap();
assert_eq!(1, manifest.entries().len());
diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index b4f1946a7..1f09a79a0 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -172,7 +172,6 @@ impl<'a> SnapshotProducer<'a> {
.collect();
let runtime = self.table.runtime();
- let file_io = self.table.file_io();
let manifest_list = self
.table
.manifest_list_reader(current_snapshot)
@@ -184,10 +183,8 @@ impl<'a> SnapshotProducer<'a> {
.consume_entries()
.into_iter()
.map(|entry| {
- let file_io = file_io.clone();
- runtime
- .io()
- .spawn(async move { entry.load_manifest(&file_io).await })
+ let reader = self.table.manifest_reader();
+ runtime.io().spawn(async move { reader.read(&entry).await })
})
.collect::<FuturesUnordered<_>>()
.try_fold(Vec::new(), |mut acc, manifest| async move {
diff --git a/crates/integrations/datafusion/src/physical_plan/commit.rs b/crates/integrations/datafusion/src/physical_plan/commit.rs
index 9ae8b845c..f1d93679e 100644
--- a/crates/integrations/datafusion/src/physical_plan/commit.rs
+++ b/crates/integrations/datafusion/src/physical_plan/commit.rs
@@ -503,8 +503,9 @@ mod tests {
assert!(!manifest_list.entries().is_empty());
// Load the first manifest and verify it contains our data files
- let manifest = manifest_list.entries()[0]
- .load_manifest(updated_table.file_io())
+ let manifest = updated_table
+ .manifest_reader()
+ .read(&manifest_list.entries()[0])
.await?;
// Verify that the manifest contains our data files