diff --git a/crates/iceberg/src/io/object_cache.rs b/crates/iceberg/src/io/object_cache.rs index 9d7815569b..eafc650542 100644 --- a/crates/iceberg/src/io/object_cache.rs +++ b/crates/iceberg/src/io/object_cache.rs @@ -37,7 +37,11 @@ pub(crate) enum CachedItem { #[derive(Clone, Debug, Hash, Eq, PartialEq)] pub(crate) enum CachedObjectKey { ManifestList((String, FormatVersion, SchemaId)), - Manifest(String), + // The manifest-level `first_row_id` is part of the key because the parsed + // manifest inherits it onto its entries: the same physical manifest can be + // referenced with different offsets across snapshots and branches, so it + // cannot be shared under the path alone. + Manifest((String, Option)), } /// Caches metadata objects deserialized from immutable files @@ -105,7 +109,10 @@ impl ObjectCache { .map(Arc::new); } - let key = CachedObjectKey::Manifest(manifest_file.manifest_path.clone()); + let key = CachedObjectKey::Manifest(( + manifest_file.manifest_path.clone(), + manifest_file.first_row_id, + )); let cache_entry = self .cache @@ -434,4 +441,73 @@ mod tests { "1.parquet" ); } + + #[tokio::test] + async fn test_get_manifest_keys_on_first_row_id() { + use crate::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type}; + + let io = FileIO::new_with_memory(); + let path = "memory:///metadata/first_row_id_manifest.avro"; + + let schema: SchemaRef = Schema::builder() + .with_fields(vec![ + NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap() + .into(); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .build() + .unwrap(); + + let mut writer = ManifestWriterBuilder::new( + io.new_output(path).unwrap(), + Some(1), + schema, + partition_spec, + ) + .build_v3_data(); + writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let manifest_file = writer.write_manifest_file().await.unwrap(); + + // Two manifest-list references to the same physical manifest carrying + // different offsets, as time travel or two branches would produce. + let mut manifest_file_a = manifest_file.clone(); + manifest_file_a.first_row_id = Some(1000); + let mut manifest_file_b = manifest_file; + manifest_file_b.first_row_id = Some(2000); + + let object_cache = ObjectCache::new(io, None); + + let manifest_a = object_cache.get_manifest(&manifest_file_a).await.unwrap(); + let manifest_b = object_cache.get_manifest(&manifest_file_b).await.unwrap(); + + // A cache keyed on path alone would serve manifest_a's inherited ids for + // the second read; keying on first_row_id keeps them distinct. + assert_eq!( + manifest_a.entries()[0].data_file().first_row_id(), + Some(1000) + ); + assert_eq!( + manifest_b.entries()[0].data_file().first_row_id(), + Some(2000) + ); + } } diff --git a/crates/iceberg/src/spec/manifest_list/manifest_file.rs b/crates/iceberg/src/spec/manifest_list/manifest_file.rs index 9518a11e82..a05377a024 100644 --- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs +++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs @@ -23,7 +23,7 @@ use super::ByteBuf; use crate::encryption::{EncryptedInputFile, StandardKeyMetadata}; use crate::error::Result; use crate::io::FileIO; -use crate::spec::Manifest; +use crate::spec::{Manifest, ManifestEntry}; use crate::{Error, ErrorKind}; /// Entry in a manifest list. @@ -195,8 +195,76 @@ impl ManifestFile { 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 + /// . + /// + /// 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. @@ -225,7 +293,6 @@ pub struct FieldSummary { #[cfg(test)] mod test { - use std::collections::HashMap; use std::sync::Arc; use super::{ManifestContentType, ManifestFile}; @@ -233,8 +300,8 @@ mod test { use crate::encryption::{EncryptedOutputFile, StandardKeyMetadata}; use crate::io::FileIO; use crate::spec::{ - DataContentType, DataFile, DataFileFormat, ManifestEntry, ManifestStatus, - ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, Struct, Type, + DataContentType, DataFileBuilder, DataFileFormat, ManifestEntry, ManifestStatus, + ManifestWriterBuilder, NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type, }; #[test] @@ -247,15 +314,9 @@ mod test { assert_eq!(ManifestContentType::default() as i32, 0); } - /// Writes a single-entry manifest, encrypting it with `key_metadata`, and - /// returns the resulting [`ManifestFile`]. The manifest is stored in `io` - /// at `path`. - async fn write_encrypted_manifest( - io: &FileIO, - path: &str, - key_metadata: StandardKeyMetadata, - ) -> ManifestFile { - let schema = Arc::new( + /// A single-field schema used by the manifest-writing test helpers. + fn test_schema() -> SchemaRef { + Arc::new( Schema::builder() .with_fields(vec![Arc::new(NestedField::optional( 1, @@ -264,8 +325,37 @@ mod test { ))]) .build() .unwrap(), - ); + ) + } + /// Writes a single-entry v3 data manifest to `io` at `path`, without + /// encryption, and returns the resulting [`ManifestFile`]. + async fn write_manifest(io: &FileIO, path: &str) -> ManifestFile { + let schema = test_schema(); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(0) + .build() + .unwrap(); + + let output_file = io.new_output(path).unwrap(); + let mut writer = ManifestWriterBuilder::new(output_file, Some(1), schema, partition_spec) + .build_v3_data(); + + writer + .add_entry(data_entry(ManifestStatus::Added, 100, None)) + .unwrap(); + + writer.write_manifest_file().await.unwrap() + } + + /// Writes a single-entry v3 data manifest to `io` at `path`, encrypting it + /// with `key_metadata`, and returns the resulting [`ManifestFile`]. + async fn write_encrypted_manifest( + io: &FileIO, + path: &str, + key_metadata: StandardKeyMetadata, + ) -> ManifestFile { + let schema = test_schema(); let partition_spec = PartitionSpec::builder(schema.clone()) .with_spec_id(0) .build() @@ -277,42 +367,14 @@ mod test { let mut writer = ManifestWriterBuilder::new_from_encrypted( encrypted_output, Some(1), - schema.clone(), - partition_spec.clone(), + schema, + partition_spec, ) .expect("Expected a valid writer") .build_v3_data(); writer - .add_entry(ManifestEntry { - status: ManifestStatus::Added, - snapshot_id: None, - sequence_number: None, - file_sequence_number: None, - data_file: DataFile { - content: DataContentType::Data, - file_path: "s3://bucket/table/data/00000.parquet".to_string(), - file_format: DataFileFormat::Parquet, - partition: Struct::empty(), - record_count: 100, - 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, - }, - }) + .add_entry(data_entry(ManifestStatus::Added, 100, None)) .unwrap(); writer.write_manifest_file().await.unwrap() @@ -389,4 +451,165 @@ mod test { .expect_err("load_manifest must fail when decrypting with the wrong AAD prefix"); assert_eq!(err.kind(), ErrorKind::Unexpected); } + + /// 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, + ) -> 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) -> 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(); + let path = "memory:///test/plaintext_manifest.avro"; + let manifest_file = write_manifest(&io, path).await; + assert_eq!(manifest_file.key_metadata, None); + + let manifest = manifest_file.load_manifest(&io).await.unwrap(); + assert_eq!(manifest.entries().len(), 1); + assert_eq!( + manifest.entries()[0].file_path(), + "s3://bucket/table/data/00000.parquet" + ); + assert_eq!(manifest.entries()[0].data_file.record_count, 100); + } + + /// 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 + /// write/read round-trip that leaves per-file `first_row_id` as `None`. + #[tokio::test] + async fn test_load_manifest_assigns_first_row_ids() { + let io = FileIO::new_with_memory(); + let path = "memory:///test/first_row_id_manifest.avro"; + let mut manifest_file = write_manifest(&io, path).await; + + // 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(); + assert_eq!(manifest.entries().len(), 1); + assert_eq!(manifest.entries()[0].data_file().first_row_id(), Some(1000)); + } }