diff --git a/iceberg-rust-spec/src/spec/manifest.rs b/iceberg-rust-spec/src/spec/manifest.rs index e9791102..afc2de80 100644 --- a/iceberg-rust-spec/src/spec/manifest.rs +++ b/iceberg-rust-spec/src/spec/manifest.rs @@ -24,6 +24,7 @@ use serde_repr::{Deserialize_repr, Serialize_repr}; use crate::{error::Error, partition::BoundPartitionField}; use super::{ + decimal::decimal_scale, partition::PartitionSpec, schema::Schema, table_metadata::FormatVersion, @@ -92,6 +93,94 @@ impl ManifestEntry { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FirstRowIdInheritanceMode { + Assign(i64), + Clear, + Preserve, +} + +/// Applies Iceberg v3 first-row-ID inheritance while reading manifest entries. +/// +/// A committed data manifest with a manifest-level first row ID assigns that ID +/// to live data files that do not already carry one, advancing by each assigned +/// file's record count. A committed manifest without an ID predates row lineage, +/// so per-file IDs are cleared. Uncommitted manifests without an ID preserve +/// their entries because manifest-list assignment has not happened yet. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FirstRowIdInheritance { + mode: FirstRowIdInheritanceMode, +} + +impl FirstRowIdInheritance { + /// Creates inheritance state for a manifest read from a committed manifest list. + pub fn for_committed_manifest(first_row_id: Option) -> Self { + Self { + mode: first_row_id.map_or( + FirstRowIdInheritanceMode::Clear, + FirstRowIdInheritanceMode::Assign, + ), + } + } + + /// Creates validated inheritance state for a committed manifest. + pub fn try_for_committed_manifest(first_row_id: Option) -> Result { + if first_row_id.is_some_and(|first_row_id| first_row_id < 0) { + return Err(Error::InvalidFormat( + "manifest first row id must be non-negative".to_string(), + )); + } + Ok(Self::for_committed_manifest(first_row_id)) + } + + /// Creates inheritance state for a manifest that has not been committed yet. + pub fn for_uncommitted_manifest(first_row_id: Option) -> Self { + Self { + mode: first_row_id.map_or( + FirstRowIdInheritanceMode::Preserve, + FirstRowIdInheritanceMode::Assign, + ), + } + } + + /// Applies inheritance to one entry in manifest order. + pub fn apply(&mut self, entry: &mut ManifestEntry) -> Result<(), Error> { + match &mut self.mode { + FirstRowIdInheritanceMode::Clear => { + entry.data_file.first_row_id = None; + } + FirstRowIdInheritanceMode::Preserve => {} + FirstRowIdInheritanceMode::Assign(next_row_id) => { + if *next_row_id < 0 { + return Err(Error::InvalidFormat( + "manifest first row id must be non-negative".to_string(), + )); + } + if entry.status == Status::Deleted + || entry.data_file.content != Content::Data + || entry.data_file.first_row_id.is_some() + { + return Ok(()); + } + + let record_count = entry.data_file.record_count; + if record_count < 0 { + return Err(Error::InvalidFormat( + "data file record count must be non-negative".to_string(), + )); + } + + let first_row_id = *next_row_id; + *next_row_id = first_row_id.checked_add(record_count).ok_or_else(|| { + Error::InvalidFormat("first row id inheritance overflow".to_string()) + })?; + entry.data_file.first_row_id = Some(first_row_id); + } + } + Ok(()) + } +} + impl ManifestEntry { pub fn try_from_v3( value: ManifestEntryV3, @@ -664,20 +753,30 @@ impl DataFile { .into_iter() .flatten() { - for (field_id, value) in bounds { + bounds.retain(|field_id, value| { let Some(field_type) = field_type_by_id(schema.fields(), *field_id) else { - continue; + return true; }; - match (&*value, field_type) { - (Value::Int(inner), Type::Primitive(PrimitiveType::Long)) => { - *value = Value::LongInt(i64::from(*inner)); - } - (Value::Float(inner), Type::Primitive(PrimitiveType::Double)) => { - *value = Value::Double(ordered_float::OrderedFloat(f64::from(inner.0))); + let source_type = value.datatype(); + if source_type == *field_type + || matches!( + (&*value, field_type), + ( + Value::Decimal(decimal), + Type::Primitive(PrimitiveType::Decimal { scale, .. }) + ) if decimal_scale(decimal) == *scale + ) + { + return true; + } + match value.clone().promote_iceberg(&source_type, field_type) { + Ok(promoted) => { + *value = promoted; + true } - _ => {} + Err(_) => false, } - } + }); } } @@ -875,12 +974,16 @@ pub struct DataFileV3 { /// ID representing sort order for this file pub sort_order_id: Option, /// First row ID stored directly on this data file. + #[serde(default)] pub first_row_id: Option, /// Location of the data file the deletion vector applies to. + #[serde(default)] pub referenced_data_file: Option, /// Byte offset of the deletion-vector blob inside the Puffin file. + #[serde(default)] pub content_offset: Option, /// Length of the deletion-vector blob (compressed if applicable). + #[serde(default)] pub content_size_in_bytes: Option, } @@ -1943,10 +2046,11 @@ impl DataFileV2 { #[cfg(test)] mod tests { use crate::spec::{ + decimal::decimal_from_i128_with_scale, partition::{PartitionField, Transform}, table_metadata::TableMetadataBuilder, types::{PrimitiveType, StructField, Type}, - values::Value, + values::{Struct, Value}, }; use super::*; @@ -1984,6 +2088,23 @@ mod tests { )])), None, ), + StructField::new( + 5, + "decimal_value", + false, + Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }), + None, + ), + StructField::new( + 6, + "string_value", + false, + Type::Primitive(PrimitiveType::String), + None, + ), ]), 1, None, @@ -2004,11 +2125,21 @@ mod tests { (1, Value::Int(7)), (2, Value::Float(OrderedFloat(1.5))), (4, Value::Int(11)), + ( + 5, + Value::Decimal(decimal_from_i128_with_scale(12_345, 2).unwrap()), + ), + (6, Value::String("lower".to_string())), ])), upper_bounds: Some(HashMap::from([ (1, Value::Int(9)), (2, Value::Float(OrderedFloat(2.5))), (4, Value::Int(13)), + ( + 5, + Value::Decimal(decimal_from_i128_with_scale(67_890, 2).unwrap()), + ), + (6, Value::String("upper".to_string())), ])), key_metadata: None, split_offsets: None, @@ -2028,6 +2159,11 @@ mod tests { (1, Value::LongInt(7)), (2, Value::Double(OrderedFloat(1.5))), (4, Value::LongInt(11)), + ( + 5, + Value::Decimal(decimal_from_i128_with_scale(12_345, 2).unwrap()), + ), + (6, Value::String("lower".to_string())), ])) ); assert_eq!( @@ -2036,10 +2172,180 @@ mod tests { (1, Value::LongInt(9)), (2, Value::Double(OrderedFloat(2.5))), (4, Value::LongInt(13)), + ( + 5, + Value::Decimal(decimal_from_i128_with_scale(67_890, 2).unwrap()), + ), + (6, Value::String("upper".to_string())), ])) ); } + fn row_id_entry( + status: Status, + content: Content, + record_count: i64, + first_row_id: Option, + ) -> ManifestEntry { + ManifestEntry { + format_version: FormatVersion::V3, + status, + snapshot_id: Some(1), + sequence_number: Some(1), + data_file: DataFile { + content, + file_path: "/data.parquet".to_string(), + file_format: FileFormat::Parquet, + partition: Struct::from_iter(Vec::<(String, Option)>::new()), + record_count, + file_size_in_bytes: 1, + column_sizes: None, + value_counts: None, + null_value_counts: None, + nan_value_counts: None, + distinct_counts: None, + lower_bounds: None, + upper_bounds: None, + key_metadata: None, + split_offsets: None, + equality_ids: None, + sort_order_id: None, + first_row_id, + referenced_data_file: None, + content_offset: None, + content_size_in_bytes: None, + }, + } + } + + #[test] + fn first_row_id_inheritance_assigns_live_null_data_files_in_manifest_order() { + let mut entries = [ + row_id_entry(Status::Added, Content::Data, 3, None), + row_id_entry(Status::Existing, Content::Data, 2, Some(100)), + row_id_entry(Status::Deleted, Content::Data, 7, None), + row_id_entry(Status::Added, Content::PositionDeletes, 5, None), + row_id_entry(Status::Existing, Content::Data, 4, None), + ]; + let mut inheritance = FirstRowIdInheritance::for_committed_manifest(Some(10)); + + for entry in &mut entries { + inheritance.apply(entry).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(), None); + assert_eq!(*entries[3].data_file().first_row_id(), None); + assert_eq!(*entries[4].data_file().first_row_id(), Some(13)); + } + + #[test] + fn first_row_id_inheritance_clears_committed_pre_upgrade_ids() { + let mut entry = row_id_entry(Status::Existing, Content::Data, 3, Some(100)); + FirstRowIdInheritance::for_committed_manifest(None) + .apply(&mut entry) + .unwrap(); + assert_eq!(*entry.data_file().first_row_id(), None); + } + + #[test] + fn first_row_id_inheritance_preserves_uncommitted_ids() { + let mut explicit = row_id_entry(Status::Existing, Content::Data, 3, Some(100)); + let mut inherited = row_id_entry(Status::Added, Content::Data, 4, None); + let mut inheritance = FirstRowIdInheritance::for_uncommitted_manifest(None); + + inheritance.apply(&mut explicit).unwrap(); + inheritance.apply(&mut inherited).unwrap(); + + assert_eq!(*explicit.data_file().first_row_id(), Some(100)); + assert_eq!(*inherited.data_file().first_row_id(), None); + } + + #[test] + fn first_row_id_inheritance_rejects_counter_overflow() { + let mut entry = row_id_entry(Status::Added, Content::Data, 2, None); + let error = FirstRowIdInheritance::for_committed_manifest(Some(i64::MAX)) + .apply(&mut entry) + .unwrap_err(); + assert!(matches!(error, Error::InvalidFormat(_))); + assert_eq!(*entry.data_file().first_row_id(), None); + } + + #[test] + fn first_row_id_inheritance_rejects_negative_manifest_id() { + let mut entry = row_id_entry(Status::Added, Content::Data, 1, None); + let error = FirstRowIdInheritance::for_committed_manifest(Some(-1)) + .apply(&mut entry) + .unwrap_err(); + + assert!(matches!(error, Error::InvalidFormat(_))); + assert_eq!(*entry.data_file().first_row_id(), None); + } + + #[test] + fn first_row_id_inheritance_rejects_negative_empty_manifest_id() { + let error = FirstRowIdInheritance::try_for_committed_manifest(Some(-1)).unwrap_err(); + + assert!(matches!(error, Error::InvalidFormat(_))); + } + + #[test] + fn data_file_v3_reads_schema_written_before_row_lineage_fields() { + let partition_schema = partition_value_schema(&[]).unwrap(); + let mut schema_json: serde_json::Value = + serde_json::from_str(&DataFileV3::schema(&partition_schema)).unwrap(); + let fields = schema_json + .get_mut("fields") + .and_then(serde_json::Value::as_array_mut) + .unwrap(); + fields.retain(|field| { + !matches!( + field.get("name").and_then(serde_json::Value::as_str), + Some( + "first_row_id" + | "referenced_data_file" + | "content_offset" + | "content_size_in_bytes" + ) + ) + }); + let old_schema = + AvroSchema::parse_str(&serde_json::to_string(&schema_json).unwrap()).unwrap(); + + let data_file: DataFileV3 = row_id_entry(Status::Added, Content::Data, 3, None) + .data_file + .into(); + let mut value = apache_avro::to_value(data_file).unwrap(); + let AvroValue::Record(fields) = &mut value else { + panic!("data file must serialize as an Avro record"); + }; + fields.retain(|(name, _)| { + !matches!( + name.as_str(), + "first_row_id" + | "referenced_data_file" + | "content_offset" + | "content_size_in_bytes" + ) + }); + + let mut writer = apache_avro::Writer::new(&old_schema, Vec::new()); + writer.append(value).unwrap(); + let encoded = writer.into_inner().unwrap(); + let value = apache_avro::Reader::new(&encoded[..]) + .unwrap() + .next() + .unwrap() + .unwrap(); + let restored = apache_avro::from_value::(&value).unwrap(); + + assert_eq!(restored.first_row_id, None); + assert_eq!(restored.referenced_data_file, None); + assert_eq!(restored.content_offset, None); + assert_eq!(restored.content_size_in_bytes, None); + } + #[test] fn manifest_entry() { let table_metadata = TableMetadataBuilder::default() diff --git a/iceberg-rust-spec/src/spec/manifest_list.rs b/iceberg-rust-spec/src/spec/manifest_list.rs index ad335e42..91112c6e 100644 --- a/iceberg-rust-spec/src/spec/manifest_list.rs +++ b/iceberg-rust-spec/src/spec/manifest_list.rs @@ -12,7 +12,7 @@ //! They include summary statistics that can be used to skip reading manifests that //! don't contain relevant data for a query. -use std::sync::OnceLock; +use std::{collections::HashMap, sync::OnceLock}; use apache_avro::{types::Value as AvroValue, Schema as AvroSchema}; use serde::{Deserialize, Serialize}; @@ -196,9 +196,6 @@ mod _serde { pub partitions: Option>, /// Implementation-specific key metadata for encryption pub key_metadata: Option, - /// This field is absent in v2 and remains null when decoding a v2 manifest list. - #[serde(default)] - pub first_row_id: Option, } #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] @@ -310,7 +307,6 @@ mod _serde { .partitions .map(|v| v.into_iter().map(Into::into).collect()), key_metadata: value.key_metadata, - first_row_id: None, } } } @@ -343,6 +339,53 @@ mod _serde { } impl ManifestListEntry { + fn preferred_schema_id(table_metadata: &TableMetadata, added_snapshot_id: i64) -> i32 { + table_metadata + .snapshots + .get(&added_snapshot_id) + .and_then(|snapshot| *snapshot.schema_id()) + .unwrap_or(table_metadata.current_schema_id) + } + + fn partition_type_candidates( + table_metadata: &TableMetadata, + partition_spec_id: i32, + preferred_schema_id: i32, + ) -> Result>, Error> { + let partition_spec = table_metadata + .partition_specs + .get(&partition_spec_id) + .ok_or_else(|| { + Error::NotFound(format!("Partition spec with id {partition_spec_id}")) + })?; + let mut schema_ids = table_metadata.schemas.keys().copied().collect::>(); + schema_ids.sort_unstable_by(|left, right| right.cmp(left)); + schema_ids.retain(|schema_id| *schema_id != preferred_schema_id); + schema_ids.insert(0, preferred_schema_id); + + let mut candidates: Option>> = None; + for schema_id in schema_ids { + let Some(schema) = table_metadata.schemas.get(&schema_id) else { + continue; + }; + let Ok(types) = partition_spec.data_types(schema.fields()) else { + continue; + }; + let candidates = candidates.get_or_insert_with(|| vec![Vec::new(); types.len()]); + for (field_candidates, data_type) in candidates.iter_mut().zip(types) { + if !field_candidates.contains(&data_type) { + field_candidates.push(data_type); + } + } + } + + candidates.ok_or_else(|| { + Error::NotFound(format!( + "Schema containing all source fields for partition spec {partition_spec_id}" + )) + }) + } + pub fn try_from_enum( entry: ManifestListEntryEnum, table_metadata: &TableMetadata, @@ -364,18 +407,20 @@ impl ManifestListEntry { entry: _serde::ManifestListEntryV3, table_metadata: &TableMetadata, ) -> Result { - let partition_types = table_metadata.default_partition_spec()?.data_types( - table_metadata - .current_schema() - .or(table_metadata - .refs - .values() - .next() - .ok_or(Error::NotFound("Current schema".to_string())) - .and_then(|x| table_metadata.schema(x.snapshot_id))) - .unwrap() - .fields(), + let preferred_schema_id = + Self::preferred_schema_id(table_metadata, entry.added_snapshot_id); + let partition_types = Self::partition_type_candidates( + table_metadata, + entry.partition_spec_id, + preferred_schema_id, )?; + Self::try_from_v3_with_partition_types(entry, &partition_types) + } + + fn try_from_v3_with_partition_types( + entry: _serde::ManifestListEntryV3, + partition_types: &[Vec], + ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V3, manifest_path: entry.manifest_path, @@ -396,7 +441,7 @@ impl ManifestListEntry { .map(|v| { v.into_iter() .zip(partition_types.iter()) - .map(|(x, d)| FieldSummary::try_from(x, d)) + .map(|(x, candidates)| FieldSummary::try_from(x, candidates)) .collect::, Error>>() }) .transpose()?, @@ -409,18 +454,20 @@ impl ManifestListEntry { entry: _serde::ManifestListEntryV2, table_metadata: &TableMetadata, ) -> Result { - let partition_types = table_metadata.default_partition_spec()?.data_types( - table_metadata - .current_schema() - .or(table_metadata - .refs - .values() - .next() - .ok_or(Error::NotFound("Current schema".to_string())) - .and_then(|x| table_metadata.schema(x.snapshot_id))) - .unwrap() - .fields(), + let preferred_schema_id = + Self::preferred_schema_id(table_metadata, entry.added_snapshot_id); + let partition_types = Self::partition_type_candidates( + table_metadata, + entry.partition_spec_id, + preferred_schema_id, )?; + Self::try_from_v2_with_partition_types(entry, &partition_types) + } + + fn try_from_v2_with_partition_types( + entry: _serde::ManifestListEntryV2, + partition_types: &[Vec], + ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V2, manifest_path: entry.manifest_path, @@ -441,7 +488,7 @@ impl ManifestListEntry { .map(|v| { v.into_iter() .zip(partition_types.iter()) - .map(|(x, d)| FieldSummary::try_from(x, d)) + .map(|(x, candidates)| FieldSummary::try_from(x, candidates)) .collect::, Error>>() }) .transpose()?, @@ -454,18 +501,20 @@ impl ManifestListEntry { entry: _serde::ManifestListEntryV1, table_metadata: &TableMetadata, ) -> Result { - let partition_types = table_metadata.default_partition_spec()?.data_types( - table_metadata - .current_schema() - .or(table_metadata - .refs - .values() - .next() - .ok_or(Error::NotFound("Current schema".to_string())) - .and_then(|x| table_metadata.schema(x.snapshot_id))) - .unwrap() - .fields(), + let preferred_schema_id = + Self::preferred_schema_id(table_metadata, entry.added_snapshot_id); + let partition_types = Self::partition_type_candidates( + table_metadata, + entry.partition_spec_id, + preferred_schema_id, )?; + Self::try_from_v1_with_partition_types(entry, &partition_types) + } + + fn try_from_v1_with_partition_types( + entry: _serde::ManifestListEntryV1, + partition_types: &[Vec], + ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V1, manifest_path: entry.manifest_path, @@ -486,7 +535,7 @@ impl ManifestListEntry { .map(|v| { v.into_iter() .zip(partition_types.iter()) - .map(|(x, d)| FieldSummary::try_from(x, d)) + .map(|(x, candidates)| FieldSummary::try_from(x, candidates)) .collect::, Error>>() }) .transpose()?, @@ -496,21 +545,104 @@ impl ManifestListEntry { } } +/// Stateful manifest-list decoder that reuses partition type candidates across entries. +pub struct ManifestListEntryDecoder<'a> { + table_metadata: &'a TableMetadata, + partition_type_candidates: HashMap<(i32, i32), Vec>>, +} + +impl<'a> ManifestListEntryDecoder<'a> { + pub fn new(table_metadata: &'a TableMetadata) -> Self { + Self { + table_metadata, + partition_type_candidates: HashMap::new(), + } + } + + fn partition_type_candidates( + &mut self, + partition_spec_id: i32, + added_snapshot_id: i64, + ) -> Result<&[Vec], Error> { + let preferred_schema_id = + ManifestListEntry::preferred_schema_id(self.table_metadata, added_snapshot_id); + let key = (partition_spec_id, preferred_schema_id); + if !self.partition_type_candidates.contains_key(&key) { + let candidates = ManifestListEntry::partition_type_candidates( + self.table_metadata, + partition_spec_id, + preferred_schema_id, + )?; + self.partition_type_candidates.insert(key, candidates); + } + Ok(self.partition_type_candidates.get(&key).unwrap()) + } + + pub fn decode( + &mut self, + value: Result, + format_version: FormatVersion, + ) -> Result { + let entry = value?; + match format_version { + FormatVersion::V1 => { + let entry = apache_avro::from_value::<_serde::ManifestListEntryV1>(&entry)?; + let partition_types = self + .partition_type_candidates(entry.partition_spec_id, entry.added_snapshot_id)?; + ManifestListEntry::try_from_v1_with_partition_types(entry, partition_types) + } + FormatVersion::V2 => { + let entry = apache_avro::from_value::<_serde::ManifestListEntryV2>(&entry)?; + let partition_types = self + .partition_type_candidates(entry.partition_spec_id, entry.added_snapshot_id)?; + ManifestListEntry::try_from_v2_with_partition_types(entry, partition_types) + } + FormatVersion::V3 => { + let entry = apache_avro::from_value::<_serde::ManifestListEntryV3>(&entry)?; + let partition_types = self + .partition_type_candidates(entry.partition_spec_id, entry.added_snapshot_id)?; + ManifestListEntry::try_from_v3_with_partition_types(entry, partition_types) + } + } + } +} + impl FieldSummary { - fn try_from(value: _serde::FieldSummarySerde, data_type: &Type) -> Result { + fn try_from( + value: _serde::FieldSummarySerde, + data_type_candidates: &[Type], + ) -> Result { Ok(FieldSummary { contains_null: value.contains_null, contains_nan: value.contains_nan, lower_bound: value .lower_bound - .map(|x| Value::try_from_bytes(&x, data_type)) + .map(|x| Self::decode_bound(&x, data_type_candidates)) .transpose()?, upper_bound: value .upper_bound - .map(|x| Value::try_from_bytes(&x, data_type)) + .map(|x| Self::decode_bound(&x, data_type_candidates)) .transpose()?, }) } + + fn decode_bound(bytes: &[u8], data_type_candidates: &[Type]) -> Result { + let target_type = data_type_candidates + .first() + .ok_or_else(|| Error::InvalidFormat("partition field type candidates".to_string()))?; + let mut last_error = None; + for data_type in data_type_candidates { + match Value::try_from_bytes(bytes, data_type) { + Ok(value) if data_type == target_type => return Ok(value), + Ok(value) => match value.promote_iceberg(data_type, target_type) { + Ok(value) => return Ok(value), + Err(error) => last_error = Some(error), + }, + Err(error) => last_error = Some(error), + } + } + Err(last_error.unwrap_or_else(|| Error::InvalidFormat("partition field bound".to_string()))) + } } pub fn manifest_list_schema_v1() -> &'static AvroSchema { @@ -788,15 +920,6 @@ pub fn manifest_list_schema_v2() -> &'static AvroSchema { ], "default": null, "field-id": 519 - }, - { - "name": "first_row_id", - "type": [ - "null", - "long" - ], - "default": null, - "field-id": 520 } ] } @@ -806,19 +929,47 @@ pub fn manifest_list_schema_v2() -> &'static AvroSchema { }) } -/// Manifest list Avro schema for V3 tables. Initially identical to the V2 schema. +/// Manifest list Avro schema for V3 tables. pub fn manifest_list_schema_v3() -> &'static AvroSchema { static MANIFEST_LIST_SCHEMA_V3: OnceLock = OnceLock::new(); - MANIFEST_LIST_SCHEMA_V3.get_or_init(|| manifest_list_schema_v2().clone()) + MANIFEST_LIST_SCHEMA_V3.get_or_init(|| { + let mut schema = serde_json::to_value(manifest_list_schema_v2()).unwrap(); + schema["fields"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "name": "first_row_id", + "type": ["null", "long"], + "default": null, + "field-id": 520 + })); + AvroSchema::parse(&schema).unwrap() + }) } /// Convert an avro value result to a manifest list version according to the provided format version pub fn avro_value_to_manifest_list_entry( value: Result, table_metadata: &TableMetadata, +) -> Result { + avro_value_to_manifest_list_entry_for_format_version( + value, + table_metadata, + table_metadata.format_version, + ) +} + +/// Converts an Avro value using the format version of the manifest list writer. +/// +/// This is required when a table has been upgraded: historical manifest lists retain their +/// original writer schema and must not be decoded as the table's current format version. +pub fn avro_value_to_manifest_list_entry_for_format_version( + value: Result, + table_metadata: &TableMetadata, + format_version: FormatVersion, ) -> Result { let entry = value?; - match table_metadata.format_version { + match format_version { FormatVersion::V1 => ManifestListEntry::try_from_v1( apache_avro::from_value::<_serde::ManifestListEntryV1>(&entry)?, table_metadata, @@ -842,8 +993,10 @@ mod tests { use super::*; use crate::spec::{ + decimal::decimal_from_i128_with_scale, partition::{PartitionField, PartitionSpec, Transform}, schema::Schema, + snapshot::{SnapshotBuilder, Summary}, table_metadata::TableMetadataBuilder, types::{PrimitiveType, StructField}, }; @@ -1080,4 +1233,273 @@ mod tests { ); } } + + #[test] + fn historical_manifest_uses_its_partition_spec_and_snapshot_schema() { + let historical_schema = Schema::builder() + .with_schema_id(1) + .with_struct_field(StructField { + id: 1, + name: "historical_id".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Int), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(); + let current_schema = Schema::builder() + .with_schema_id(2) + .with_struct_field(StructField { + id: 2, + name: "current_value".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::String), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(); + let historical_snapshot = SnapshotBuilder::default() + .with_snapshot_id(7) + .with_sequence_number(1) + .with_timestamp_ms(1) + .with_manifest_list("historical-list.avro".to_string()) + .with_summary(Summary::default()) + .with_schema_id(1) + .build() + .unwrap(); + let historical_snapshot_without_schema = SnapshotBuilder::default() + .with_snapshot_id(7) + .with_sequence_number(1) + .with_timestamp_ms(1) + .with_manifest_list("historical-list.avro".to_string()) + .with_summary(Summary::default()) + .build() + .unwrap(); + let table_metadata = |snapshots| { + TableMetadataBuilder::default() + .format_version(FormatVersion::V3) + .location("/") + .current_schema_id(2) + .schemas(HashMap::from([ + (1, historical_schema.clone()), + (2, current_schema.clone()), + ])) + .default_spec_id(1) + .partition_specs(HashMap::from([ + ( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "historical_id", + Transform::Identity, + )) + .build() + .unwrap(), + ), + ( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_partition_field(PartitionField::new( + 2, + 1001, + "current_value", + Transform::Identity, + )) + .build() + .unwrap(), + ), + ])) + .snapshots(snapshots) + .build() + .unwrap() + }; + let historical_entry = ManifestListEntry { + format_version: FormatVersion::V1, + manifest_path: "historical-manifest.avro".to_string(), + manifest_length: 1, + partition_spec_id: 0, + content: Content::Data, + sequence_number: 0, + min_sequence_number: 0, + added_snapshot_id: 7, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: Some(Value::Int(42)), + upper_bound: Some(Value::Int(42)), + }]), + key_metadata: None, + first_row_id: None, + }; + let mut writer = apache_avro::Writer::new(manifest_list_schema_v1(), Vec::new()); + writer.append_ser(historical_entry).unwrap(); + let bytes = writer.into_inner().unwrap(); + for metadata in [ + table_metadata(HashMap::from([(7, historical_snapshot)])), + table_metadata(HashMap::from([(7, historical_snapshot_without_schema)])), + table_metadata(HashMap::new()), + ] { + let record = apache_avro::Reader::new(&bytes[..]) + .unwrap() + .next() + .unwrap(); + let decoded = avro_value_to_manifest_list_entry_for_format_version( + record, + &metadata, + FormatVersion::V1, + ) + .unwrap(); + + assert_eq!( + decoded.partitions.unwrap()[0].lower_bound, + Some(Value::Int(42)) + ); + } + } + + fn decode_historical_bound_after_snapshot_expiration( + source_type: PrimitiveType, + target_type: PrimitiveType, + value: Value, + ) -> Value { + let schema = |schema_id, primitive_type| { + Schema::builder() + .with_schema_id(schema_id) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(primitive_type), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap() + }; + let metadata = TableMetadataBuilder::default() + .format_version(FormatVersion::V3) + .location("/") + .current_schema_id(2) + .schemas(HashMap::from([ + (1, schema(1, source_type)), + (2, schema(2, target_type)), + ])) + .default_spec_id(0) + .partition_specs(HashMap::from([( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "partition_source", + Transform::Identity, + )) + .build() + .unwrap(), + )])) + .build() + .unwrap(); + let entry = ManifestListEntry { + format_version: FormatVersion::V1, + manifest_path: "expired-snapshot-manifest.avro".to_string(), + manifest_length: 1, + partition_spec_id: 0, + content: Content::Data, + sequence_number: 0, + min_sequence_number: 0, + added_snapshot_id: 7, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: Some(value.clone()), + upper_bound: Some(value), + }]), + key_metadata: None, + first_row_id: None, + }; + let mut writer = apache_avro::Writer::new(manifest_list_schema_v1(), Vec::new()); + writer.append_ser(entry).unwrap(); + let bytes = writer.into_inner().unwrap(); + let record = apache_avro::Reader::new(&bytes[..]) + .unwrap() + .next() + .unwrap(); + + avro_value_to_manifest_list_entry_for_format_version(record, &metadata, FormatVersion::V1) + .unwrap() + .partitions + .unwrap()[0] + .lower_bound + .clone() + .unwrap() + } + + #[test] + fn expired_snapshot_bounds_follow_iceberg_schema_promotions() { + assert_eq!( + decode_historical_bound_after_snapshot_expiration( + PrimitiveType::Int, + PrimitiveType::Long, + Value::Int(42), + ), + Value::LongInt(42) + ); + + let float = Value::try_from_bytes( + &1.5_f32.to_le_bytes(), + &Type::Primitive(PrimitiveType::Float), + ) + .unwrap(); + let expected_double = Value::try_from_bytes( + &1.5_f64.to_le_bytes(), + &Type::Primitive(PrimitiveType::Double), + ) + .unwrap(); + assert_eq!( + decode_historical_bound_after_snapshot_expiration( + PrimitiveType::Float, + PrimitiveType::Double, + float, + ), + expected_double + ); + + let decimal = Value::Decimal(decimal_from_i128_with_scale(12_345, 2).unwrap()); + assert_eq!( + decode_historical_bound_after_snapshot_expiration( + PrimitiveType::Decimal { + precision: 7, + scale: 2, + }, + PrimitiveType::Decimal { + precision: 12, + scale: 2, + }, + decimal.clone(), + ), + decimal + ); + } } diff --git a/iceberg-rust-spec/src/spec/values.rs b/iceberg-rust-spec/src/spec/values.rs index 0a85f364..45a0c824 100644 --- a/iceberg-rust-spec/src/spec/values.rs +++ b/iceberg-rust-spec/src/spec/values.rs @@ -269,8 +269,24 @@ impl Struct { let datatype = map .get(name) .ok_or(Error::InvalidFormat("partition_struct".to_string()))?; - // Cast the value to the datatype - let value = field.map(|value| value.cast(datatype)).transpose()?; + // Partition values follow schema promotions, while transform outputs such as + // Int -> Date retain the existing value-cast behavior. + let value = field + .map(|value| { + let source_type = value.datatype(); + match (&source_type, datatype) { + ( + Type::Primitive(PrimitiveType::Int), + Type::Primitive(PrimitiveType::Long), + ) + | ( + Type::Primitive(PrimitiveType::Float), + Type::Primitive(PrimitiveType::Double), + ) => value.promote_iceberg(&source_type, datatype), + _ => value.cast(datatype), + } + }) + .transpose()?; Ok((name.clone(), value)) }) .collect::, Error>>()?, @@ -786,6 +802,40 @@ impl Value { } } } + + /// Applies only schema promotions allowed by the Iceberg specification. + pub fn promote_iceberg(self, source_type: &Type, target_type: &Type) -> Result { + if source_type == target_type { + return Ok(self); + } + + match (self, source_type, target_type) { + ( + Value::Int(input), + Type::Primitive(PrimitiveType::Int), + Type::Primitive(PrimitiveType::Long), + ) => Ok(Value::LongInt(i64::from(input))), + ( + Value::Float(input), + Type::Primitive(PrimitiveType::Float), + Type::Primitive(PrimitiveType::Double), + ) => Ok(Value::Double(OrderedFloat(f64::from(input.0)))), + ( + value @ Value::Decimal(_), + Type::Primitive(PrimitiveType::Decimal { + precision: source_precision, + scale: source_scale, + }), + Type::Primitive(PrimitiveType::Decimal { + precision: target_precision, + scale: target_scale, + }), + ) if source_scale == target_scale && source_precision <= target_precision => Ok(value), + _ => Err(Error::NotSupported(format!( + "Iceberg schema promotion from {source_type} to {target_type}" + ))), + } + } } /// Performs big endian sign extension @@ -1785,6 +1835,41 @@ mod tests { } } + #[test] + fn partition_struct_cast_uses_iceberg_float_promotion() { + let partition = Struct::from_iter([( + "id_partition".to_string(), + Some(Value::Float(OrderedFloat(34.11))), + )]); + let schema = StructType::new(vec![StructField { + id: 1, + name: "id".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Double), + doc: None, + initial_default: None, + write_default: None, + }]); + let partition_spec = [PartitionField::new( + 1, + 1000, + "id_partition", + Transform::Identity, + )]; + + let promoted = partition.cast(&schema, &partition_spec).unwrap(); + assert_eq!( + promoted.get("id_partition"), + Some(&Some(Value::Double(OrderedFloat(f64::from(34.11_f32))))) + ); + assert!( + Value::Float(OrderedFloat(34.11)) + .cast(&Type::Primitive(PrimitiveType::Double)) + .is_err(), + "partition promotion must not widen the generic Value::cast API" + ); + } + fn all_other_primitive_types(excluded: &[PrimitiveType]) -> Vec { let candidates = [ PrimitiveType::Boolean, diff --git a/iceberg-rust/src/catalog/commit.rs b/iceberg-rust/src/catalog/commit.rs index edbaad84..80709062 100644 --- a/iceberg-rust/src/catalog/commit.rs +++ b/iceberg-rust/src/catalog/commit.rs @@ -505,9 +505,9 @@ pub fn apply_table_updates( "v3 row lineage values must be non-negative".to_string(), )); } - if first_row_id != metadata.next_row_id { + if first_row_id < metadata.next_row_id { return Err(Error::InvalidFormat( - "v3 snapshot first-row-id does not match next-row-id".to_string(), + "v3 snapshot first-row-id is lower than next-row-id".to_string(), )); } metadata.next_row_id = first_row_id diff --git a/iceberg-rust/src/table/manifest_list.rs b/iceberg-rust/src/table/manifest_list.rs index 69a2029a..3abddeeb 100644 --- a/iceberg-rust/src/table/manifest_list.rs +++ b/iceberg-rust/src/table/manifest_list.rs @@ -6,21 +6,22 @@ use std::{ collections::{HashMap, HashSet}, future::Future, io::{Cursor, Read}, - iter::{repeat, Map, Repeat, Zip}, sync::Arc, }; -use apache_avro::{ - types::Value as AvroValue, Reader as AvroReader, Schema as AvroSchema, Writer as AvroWriter, -}; -use futures::{future::join_all, TryFutureExt, TryStreamExt}; +use apache_avro::{Reader as AvroReader, Schema as AvroSchema, Writer as AvroWriter}; +use futures::{future::join_all, stream, StreamExt, TryFutureExt, TryStreamExt}; use iceberg_rust_spec::{ manifest::{partition_value_schema, DataFile, ManifestEntry, Status}, - manifest_list::{avro_value_to_manifest_list_entry, Content, ManifestListEntry}, + manifest_list::{ + manifest_list_schema_v1, Content, ManifestListEntry, ManifestListEntryDecoder, + }, snapshot::Snapshot, table_metadata::{FormatVersion, TableMetadata}, util::strip_prefix, }; + +const MANIFEST_UPGRADE_CONCURRENCY: usize = 8; use object_store::{ObjectStore, ObjectStoreExt}; use smallvec::SmallVec; @@ -48,12 +49,6 @@ use super::{ }, }; -type ReaderZip<'a, 'metadata, R> = Zip, Repeat<&'metadata TableMetadata>>; -type ReaderMap<'a, 'metadata, R> = Map< - ReaderZip<'a, 'metadata, R>, - fn((Result, &TableMetadata)) -> Result, ->; - /// A reader for Iceberg manifest list files that provides an iterator over manifest list entries. /// /// ManifestListReader parses manifest list files according to the table's format version (V1/V2) @@ -64,13 +59,43 @@ type ReaderMap<'a, 'metadata, R> = Map< /// * `'metadata` - The lifetime of the table metadata reference /// * `R` - The type implementing `Read` that provides the manifest list data pub(crate) struct ManifestListReader<'a, 'metadata, R: Read> { - reader: ReaderMap<'a, 'metadata, R>, + reader: AvroReader<'a, R>, + decoder: ManifestListEntryDecoder<'metadata>, + writer_format_version: FormatVersion, } impl Iterator for ManifestListReader<'_, '_, R> { type Item = Result; fn next(&mut self) -> Option { - self.reader.next() + self.reader.next().map(|value| { + self.decoder + .decode(value, self.writer_format_version) + .map_err(Error::from) + }) + } +} + +fn manifest_list_format_version( + schema: &AvroSchema, + table_format_version: FormatVersion, +) -> Result { + let AvroSchema::Record(record) = schema else { + return Err(Error::InvalidFormat( + "manifest list writer schema must be a record".to_string(), + )); + }; + + let detected = if record.lookup.contains_key("first_row_id") { + FormatVersion::V3 + } else if record.lookup.contains_key("content") { + FormatVersion::V2 + } else { + FormatVersion::V1 + }; + if table_format_version == FormatVersion::V2 && detected == FormatVersion::V3 { + Ok(FormatVersion::V2) + } else { + Ok(detected) } } @@ -104,16 +129,103 @@ impl<'metadata, R: Read> ManifestListReader<'_, 'metadata, R> { // // TODO: switch back to `AvroReader::with_schema` once all major query engines write // the spec-correct field names. + let reader = AvroReader::new(reader)?; + let writer_format_version = + manifest_list_format_version(reader.writer_schema(), table_metadata.format_version)?; + Ok(Self { - reader: AvroReader::new(reader)?.zip(repeat(table_metadata)).map( - |(avro_value_res, meta)| { - avro_value_to_manifest_list_entry(avro_value_res, meta).map_err(Error::from) - }, - ), + reader, + decoder: ManifestListEntryDecoder::new(table_metadata), + writer_format_version, }) } } +pub(crate) async fn hydrate_v1_manifest_list_row_counts( + bytes: &[u8], + table_metadata: &TableMetadata, + object_store: Arc, +) -> Result>, Error> { + let reader = AvroReader::new(bytes)?; + if manifest_list_format_version(reader.writer_schema(), table_metadata.format_version)? + != FormatVersion::V1 + { + return Ok(None); + } + drop(reader); + + let entries = ManifestListReader::new(bytes, table_metadata)?.collect::, _>>()?; + if entries.iter().all(|entry| { + entry.added_files_count.is_some() + && entry.existing_files_count.is_some() + && entry.deleted_files_count.is_some() + && entry.added_rows_count.is_some() + && entry.existing_rows_count.is_some() + && entry.deleted_rows_count.is_some() + }) { + return Ok(None); + } + + let entries = stream::iter(entries.into_iter().map(|mut entry| { + let object_store = object_store.clone(); + async move { + if entry.added_files_count.is_some() + && entry.existing_files_count.is_some() + && entry.deleted_files_count.is_some() + && entry.added_rows_count.is_some() + && entry.existing_rows_count.is_some() + && entry.deleted_rows_count.is_some() + { + return Ok(entry); + } + + let manifest_bytes = object_store + .get(&strip_prefix(&entry.manifest_path).into()) + .await? + .bytes() + .await?; + let mut file_counts = [0_i32; 3]; + let mut row_counts = [0_i64; 3]; + for manifest_entry in ManifestReader::new(&manifest_bytes[..])? { + let manifest_entry = manifest_entry?; + let record_count = *manifest_entry.data_file().record_count(); + if record_count < 0 { + return Err(Error::InvalidFormat( + "data file record count must be non-negative".to_string(), + )); + } + let index = match *manifest_entry.status() { + Status::Added => 0, + Status::Existing => 1, + Status::Deleted => 2, + }; + file_counts[index] = file_counts[index] + .checked_add(1) + .ok_or_else(|| Error::InvalidFormat("manifest file count overflow".into()))?; + row_counts[index] = row_counts[index] + .checked_add(record_count) + .ok_or_else(|| Error::InvalidFormat("manifest row count overflow".into()))?; + } + entry.added_files_count = Some(file_counts[0]); + entry.existing_files_count = Some(file_counts[1]); + entry.deleted_files_count = Some(file_counts[2]); + entry.added_rows_count = Some(row_counts[0]); + entry.existing_rows_count = Some(row_counts[1]); + entry.deleted_rows_count = Some(row_counts[2]); + Ok(entry) + } + })) + .buffered(MANIFEST_UPGRADE_CONCURRENCY) + .try_collect::>() + .await?; + + let mut writer = AvroWriter::new(manifest_list_schema_v1(), Vec::new()); + for entry in entries { + writer.append_ser(entry)?; + } + Ok(Some(writer.into_inner()?)) +} + /// Reads a snapshot's manifest list file and returns an iterator over its manifest list entries. /// /// This function: @@ -329,6 +441,125 @@ pub async fn snapshot_column_bounds( /// * `selected_manifest` - Optional existing manifest that can be reused for appends /// * `bounding_partition_values` - Computed partition boundaries for the data files /// * `n_existing_files` - Count of existing files for split calculations +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct RowIdAssigner { + next_row_id: i64, +} + +impl RowIdAssigner { + fn new(next_row_id: i64) -> Self { + Self { next_row_id } + } + + /// Applies Iceberg's first-row-ID assignment rule to one manifest-list entry. + fn assign(&mut self, manifest: &mut ManifestListEntry) -> Result<(), Error> { + if manifest.content == Content::Deletes { + manifest.first_row_id = None; + return Ok(()); + } + if manifest + .first_row_id + .is_some_and(|first_row_id| first_row_id < 0) + { + return Err(Error::InvalidFormat(format!( + "manifest {} has a negative first row id", + manifest.manifest_path + ))); + } + if manifest.first_row_id.is_some() { + return Ok(()); + } + + let added_rows = required_non_negative_row_count( + manifest.added_rows_count, + "added_rows_count", + &manifest.manifest_path, + )?; + let existing_rows = required_non_negative_row_count( + manifest.existing_rows_count, + "existing_rows_count", + &manifest.manifest_path, + )?; + let assigned_first_row_id = self.next_row_id; + self.next_row_id = assigned_first_row_id + .checked_add(added_rows) + .and_then(|next| next.checked_add(existing_rows)) + .ok_or_else(|| Error::InvalidFormat("next row id overflow".to_string()))?; + manifest.first_row_id = Some(assigned_first_row_id); + Ok(()) + } +} + +fn required_non_negative_row_count( + row_count: Option, + field: &str, + manifest_path: &str, +) -> Result { + let row_count = row_count.ok_or_else(|| { + Error::InvalidFormat(format!( + "manifest {manifest_path} is missing required {field}" + )) + })?; + if row_count < 0 { + return Err(Error::InvalidFormat(format!( + "manifest {manifest_path} has negative {field}" + ))); + } + Ok(row_count) +} + +pub(crate) fn append_manifest( + writer: &mut AvroWriter<'_, Vec>, + row_id_assigner: Option<&mut RowIdAssigner>, + mut manifest: ManifestListEntry, +) -> Result<(), Error> { + let output_format_version = manifest_list_format_version(writer.schema(), FormatVersion::V3)?; + match output_format_version { + FormatVersion::V3 => { + let row_id_assigner = row_id_assigner.ok_or_else(|| { + Error::InvalidFormat("v3 manifest list requires row id assignment".to_string()) + })?; + manifest.format_version = FormatVersion::V3; + row_id_assigner.assign(&mut manifest)?; + if manifest.content == Content::Data && manifest.first_row_id.is_none() { + return Err(Error::InvalidFormat(format!( + "v3 data manifest {} has no first row id", + manifest.manifest_path + ))); + } + } + FormatVersion::V2 => { + for (field, count) in [ + ( + "added_files_count", + manifest.added_files_count.map(i64::from), + ), + ( + "existing_files_count", + manifest.existing_files_count.map(i64::from), + ), + ( + "deleted_files_count", + manifest.deleted_files_count.map(i64::from), + ), + ("added_rows_count", manifest.added_rows_count), + ("existing_rows_count", manifest.existing_rows_count), + ("deleted_rows_count", manifest.deleted_rows_count), + ] { + required_non_negative_row_count(count, field, &manifest.manifest_path)?; + } + manifest.format_version = FormatVersion::V2; + manifest.first_row_id = None; + } + FormatVersion::V1 => { + manifest.format_version = FormatVersion::V1; + manifest.first_row_id = None; + } + } + writer.append_ser(manifest)?; + Ok(()) +} + pub(crate) struct ManifestListWriter<'schema, 'metadata> { table_metadata: &'metadata TableMetadata, writer: AvroWriter<'schema, Vec>, @@ -338,7 +569,7 @@ pub(crate) struct ManifestListWriter<'schema, 'metadata> { n_existing_files: usize, commit_uuid: String, manifest_count: usize, - next_row_id: Option, + row_id_assigner: Option, } impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { @@ -398,8 +629,8 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: 0, commit_uuid, manifest_count: 0, - next_row_id: (table_metadata.format_version == FormatVersion::V3) - .then_some(table_metadata.next_row_id), + row_id_assigner: (table_metadata.format_version == FormatVersion::V3) + .then(|| RowIdAssigner::new(table_metadata.next_row_id)), }) } @@ -469,6 +700,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { // Preserve old manifests and place appended rows in new manifests instead. if table_metadata.format_version == FormatVersion::V3 { let mut file_count_all_entries = 0usize; + let mut row_id_assigner = RowIdAssigner::new(table_metadata.next_row_id); for manifest in manifest_list_reader { let manifest = manifest?; let file_count = manifest @@ -479,7 +711,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { file_count_all_entries = file_count_all_entries .checked_add(file_count.try_into()?) .ok_or_else(|| Error::InvalidFormat("manifest file count".to_string()))?; - writer.append_ser(manifest)?; + append_manifest(&mut writer, Some(&mut row_id_assigner), manifest)?; } return Ok(Self { @@ -491,7 +723,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: file_count_all_entries, commit_uuid, manifest_count: 0, - next_row_id: Some(table_metadata.next_row_id), + row_id_assigner: Some(row_id_assigner), }); } @@ -500,11 +732,12 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { delete_manifest, file_count_all_entries, } = if partition_column_names.is_empty() { - select_manifest_unpartitioned(manifest_list_reader, &mut writer)? + select_manifest_unpartitioned(manifest_list_reader, &mut writer, None)? } else { select_manifest_partitioned( manifest_list_reader, &mut writer, + None, &bounding_partition_values, )? }; @@ -518,7 +751,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: file_count_all_entries, commit_uuid, manifest_count: 0, - next_row_id: None, + row_id_assigner: None, }) } @@ -590,6 +823,9 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { let mut writer = AvroWriter::new(schema, Vec::new()); + let mut row_id_assigner = (table_metadata.format_version == FormatVersion::V3) + .then(|| RowIdAssigner::new(table_metadata.next_row_id)); + // Rewriting an unaffected v3 manifest is both unnecessary I/O and a // row-lineage hazard. Preserve it by path; affected manifests are // rewritten below after their inherited row IDs are materialized. @@ -610,7 +846,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { if manifests_to_overwrite.contains(&manifest.manifest_path) { manifests.push(manifest); } else { - writer.append_ser(manifest)?; + append_manifest(&mut writer, row_id_assigner.as_mut(), manifest)?; } } @@ -624,12 +860,11 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: file_count_all_entries, commit_uuid, manifest_count: 0, - next_row_id: Some(table_metadata.next_row_id), + row_id_assigner, }, manifests, )); } - let OverwriteManifest { manifest, file_count_all_entries, @@ -638,12 +873,14 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { select_manifest_without_overwrites_unpartitioned( manifest_list_reader, &mut writer, + row_id_assigner.as_mut(), manifests_to_overwrite, )? } else { select_manifest_without_overwrites_partitioned( manifest_list_reader, &mut writer, + row_id_assigner.as_mut(), &bounding_partition_values, manifests_to_overwrite, )? @@ -659,7 +896,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: file_count_all_entries, commit_uuid, manifest_count: 0, - next_row_id: None, + row_id_assigner, }, manifests, )) @@ -679,6 +916,8 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { ) -> Result<(Self, Vec), Error> { let manifest_list_reader = ManifestListReader::new(bytes, table_metadata)?; let mut writer = AvroWriter::new(schema, Vec::new()); + let mut row_id_assigner = (table_metadata.format_version == FormatVersion::V3) + .then(|| RowIdAssigner::new(table_metadata.next_row_id)); let mut manifests = Vec::new(); let mut file_count_all_entries = 0usize; @@ -696,7 +935,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { if manifests_to_overwrite.contains(&manifest.manifest_path) { manifests.push(manifest); } else { - writer.append_ser(manifest)?; + append_manifest(&mut writer, row_id_assigner.as_mut(), manifest)?; } } @@ -710,8 +949,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { n_existing_files: file_count_all_entries, commit_uuid: uuid::Uuid::new_v4().to_string(), manifest_count: 0, - next_row_id: (table_metadata.format_version == FormatVersion::V3) - .then_some(table_metadata.next_row_id), + row_id_assigner, }, manifests, )) @@ -1288,11 +1526,19 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { object_store: Arc, ) -> Result<(String, Option), Error> { if let Some(selected_data_manifest) = self.selected_data_manifest.take() { - self.writer.append_ser(selected_data_manifest)?; + append_manifest( + &mut self.writer, + self.row_id_assigner.as_mut(), + selected_data_manifest, + )?; } if let Some(selected_delete_manifest) = self.selected_delete_manifest.take() { - self.writer.append_ser(selected_delete_manifest)?; + append_manifest( + &mut self.writer, + self.row_id_assigner.as_mut(), + selected_delete_manifest, + )?; } let new_manifest_list_location = new_manifest_list_location( @@ -1311,7 +1557,10 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { ) .await?; - Ok((new_manifest_list_location, self.next_row_id)) + Ok(( + new_manifest_list_location, + self.row_id_assigner.map(|assigner| assigner.next_row_id), + )) } /// Processes manifests for overwrite operations by filtering out specific data files. @@ -1448,25 +1697,8 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { self.selected_data_manifest.as_ref() } - fn append_new_manifest(&mut self, mut manifest: ManifestListEntry) -> Result<(), Error> { - if manifest.content == Content::Data { - if let Some(next_row_id) = self.next_row_id.as_mut() { - let added_rows = manifest.added_rows_count.unwrap_or(0); - let existing_rows = manifest.existing_rows_count.unwrap_or(0); - if added_rows < 0 || existing_rows < 0 { - return Err(Error::InvalidFormat( - "manifest row counts must be non-negative".to_string(), - )); - } - manifest.first_row_id = Some(*next_row_id); - *next_row_id = next_row_id - .checked_add(existing_rows) - .and_then(|value| value.checked_add(added_rows)) - .ok_or_else(|| Error::InvalidFormat("next row id overflow".to_string()))?; - } - } - self.writer.append_ser(manifest)?; - Ok(()) + fn append_new_manifest(&mut self, manifest: ManifestListEntry) -> Result<(), Error> { + append_manifest(&mut self.writer, self.row_id_assigner.as_mut(), manifest) } /// Get the next manifest location, tracking and numbering preceding manifests written by this @@ -1479,3 +1711,202 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { new_manifest_location(&self.table_metadata.location, &self.commit_uuid, next_id) } } + +#[cfg(test)] +mod tests { + use super::*; + use iceberg_rust_spec::{ + manifest_list::{manifest_list_schema_v1, manifest_list_schema_v3}, + spec::{ + schema::SchemaBuilder, + table_metadata::TableMetadataBuilder, + types::{PrimitiveType, StructField, Type}, + }, + }; + + fn test_metadata(format_version: FormatVersion) -> TableMetadata { + let schema = SchemaBuilder::default() + .with_schema_id(0) + .with_struct_field(StructField { + id: 1, + name: "id".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Long), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(); + TableMetadataBuilder::default() + .format_version(format_version) + .location("s3://tests/table".to_string()) + .current_schema_id(0) + .schemas(HashMap::from_iter([(0, schema)])) + .build() + .unwrap() + } + + #[test] + fn upgraded_v3_table_reads_v1_manifest_list_from_writer_schema() { + let metadata = test_metadata(FormatVersion::V3); + let v1_entry = ManifestListEntry { + format_version: FormatVersion::V1, + manifest_path: "s3://tests/table/metadata/v1-manifest.avro".to_string(), + manifest_length: 1, + partition_spec_id: 0, + content: Content::Data, + sequence_number: 0, + min_sequence_number: 0, + added_snapshot_id: 1, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: None, + key_metadata: None, + first_row_id: None, + }; + let mut writer = AvroWriter::new(manifest_list_schema_v1(), Vec::new()); + writer.append_ser(v1_entry.clone()).unwrap(); + let bytes = writer.into_inner().unwrap(); + + let decoded = ManifestListReader::new(bytes.as_slice(), &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + + assert_eq!(decoded, vec![v1_entry]); + } + + #[test] + fn v2_table_reads_legacy_v2_schema_with_first_row_id_field() { + let metadata = test_metadata(FormatVersion::V2); + let legacy_entry = ManifestListEntry { + format_version: FormatVersion::V3, + manifest_path: "s3://tests/table/metadata/v2-manifest.avro".to_string(), + manifest_length: 1, + partition_spec_id: 0, + content: Content::Data, + sequence_number: 1, + min_sequence_number: 1, + added_snapshot_id: 1, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: None, + key_metadata: None, + first_row_id: None, + }; + let mut writer = AvroWriter::new(manifest_list_schema_v3(), Vec::new()); + writer.append_ser(legacy_entry).unwrap(); + let bytes = writer.into_inner().unwrap(); + + let decoded = ManifestListReader::new(bytes.as_slice(), &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + + assert_eq!(decoded.len(), 1); + assert_eq!(decoded[0].format_version, FormatVersion::V2); + assert_eq!(decoded[0].first_row_id, None); + } + + fn manifest( + path: &str, + content: Content, + first_row_id: Option, + added_rows_count: Option, + existing_rows_count: Option, + ) -> ManifestListEntry { + ManifestListEntry { + format_version: FormatVersion::V3, + manifest_path: path.to_string(), + manifest_length: 1, + partition_spec_id: 0, + content, + sequence_number: 1, + min_sequence_number: 1, + added_snapshot_id: 1, + added_files_count: Some(0), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count, + existing_rows_count, + deleted_rows_count: Some(0), + partitions: None, + key_metadata: None, + first_row_id, + } + } + + #[test] + fn row_id_assigner_follows_spec_worked_example() { + let mut assigner = RowIdAssigner::new(1000); + let mut existing = manifest("existing.avro", Content::Data, Some(925), Some(0), Some(75)); + let mut mixed = manifest("mixed.avro", Content::Data, None, Some(100), Some(25)); + let mut added = manifest("added.avro", Content::Data, None, Some(100), Some(0)); + let mut next = manifest("next.avro", Content::Data, None, Some(25), Some(0)); + + assigner.assign(&mut existing).unwrap(); + assigner.assign(&mut mixed).unwrap(); + assigner.assign(&mut added).unwrap(); + assigner.assign(&mut next).unwrap(); + + assert_eq!(existing.first_row_id, Some(925)); + assert_eq!(mixed.first_row_id, Some(1000)); + assert_eq!(added.first_row_id, Some(1125)); + assert_eq!(next.first_row_id, Some(1225)); + assert_eq!(assigner.next_row_id, 1250); + } + + #[test] + fn row_id_assigner_assigns_upgraded_existing_rows_and_skips_deletes() { + let mut assigner = RowIdAssigner::new(1000); + let mut upgraded = manifest("upgraded.avro", Content::Data, None, Some(0), Some(30)); + let mut deletes = manifest( + "deletes.avro", + Content::Deletes, + Some(99), + Some(10), + Some(20), + ); + + assigner.assign(&mut upgraded).unwrap(); + assigner.assign(&mut deletes).unwrap(); + + assert_eq!(upgraded.first_row_id, Some(1000)); + assert_eq!(deletes.first_row_id, None); + assert_eq!(assigner.next_row_id, 1030); + } + + #[test] + fn row_id_assigner_requires_both_row_counts() { + let mut assigner = RowIdAssigner::new(1000); + let mut missing_existing = manifest("missing.avro", Content::Data, None, Some(1), None); + + assert!(matches!( + assigner.assign(&mut missing_existing), + Err(Error::InvalidFormat(_)) + )); + assert_eq!(missing_existing.first_row_id, None); + assert_eq!(assigner.next_row_id, 1000); + } + + #[test] + fn row_id_assigner_rejects_negative_inherited_first_row_id() { + let mut assigner = RowIdAssigner::new(1000); + let mut invalid = manifest("invalid.avro", Content::Data, Some(-1), Some(1), Some(0)); + + assert!(matches!( + assigner.assign(&mut invalid), + Err(Error::InvalidFormat(_)) + )); + assert_eq!(assigner.next_row_id, 1000); + } +} diff --git a/iceberg-rust/src/table/mod.rs b/iceberg-rust/src/table/mod.rs index 941a953f..5072dac0 100644 --- a/iceberg-rust/src/table/mod.rs +++ b/iceberg-rust/src/table/mod.rs @@ -22,8 +22,8 @@ use futures::{stream, Stream, StreamExt, TryFutureExt, TryStreamExt}; use iceberg_rust_spec::util::{self}; use iceberg_rust_spec::{ spec::{ - manifest::{Content, ManifestEntry, Status}, - manifest_list::{Content as ManifestListContent, ManifestListEntry}, + manifest::{Content, FirstRowIdInheritance, ManifestEntry}, + manifest_list::ManifestListEntry, schema::Schema, table_metadata::TableMetadata, }, @@ -396,7 +396,6 @@ async fn datafiles( let manifest_sequence_number = file.sequence_number; let manifest_snapshot_id = file.added_snapshot_id; let manifest_first_row_id = file.first_row_id; - let manifest_content = file.content; async move { // Manifest files are immutable by path. Key by the original // URI so equal store-relative paths cannot alias across stores. @@ -415,40 +414,37 @@ async fn datafiles( }; let bytes = Cursor::new(Vec::from(data)); - let mut entries = ManifestReader::new(bytes)?.collect::, Error>>()?; - assign_first_row_ids( - &manifest_path, - manifest_content, - manifest_first_row_id, - &mut entries, - )?; - Ok::<_, Error>( - entries - .into_iter() - .filter_map(|mut x| { - if x.snapshot_id().is_none() { - *x.snapshot_id_mut() = Some(manifest_snapshot_id); + let mut first_row_id_inheritance = + FirstRowIdInheritance::try_for_committed_manifest(manifest_first_row_id)?; + + ManifestReader::new(bytes)? + .map(move |entry| { + let mut entry = entry?; + first_row_id_inheritance.apply(&mut entry)?; + Ok(entry) + }) + .filter_map_ok(|mut x| { + if x.snapshot_id().is_none() { + *x.snapshot_id_mut() = Some(manifest_snapshot_id); + } + let sequence_number = if let Some(sequence_number) = x.sequence_number() { + *sequence_number + } else { + *x.sequence_number_mut() = Some(manifest_sequence_number); + manifest_sequence_number + }; + + let keep = match sequence_number_range { + (Some(start), Some(end)) => { + start < sequence_number && sequence_number <= end } - let sequence_number = if let Some(sequence_number) = x.sequence_number() - { - *sequence_number - } else { - *x.sequence_number_mut() = Some(manifest_sequence_number); - manifest_sequence_number - }; - - let keep = match sequence_number_range { - (Some(start), Some(end)) => { - start < sequence_number && sequence_number <= end - } - (Some(start), None) => start < sequence_number, - (None, Some(end)) => sequence_number <= end, - _ => true, - }; - keep.then(|| (manifest_path.clone(), x)) - }) - .collect::>(), - ) + (Some(start), None) => start < sequence_number, + (None, Some(end)) => sequence_number <= end, + _ => true, + }; + keep.then(|| (manifest_path.clone(), x)) + }) + .collect::, Error>>() } }) .collect(); @@ -462,54 +458,6 @@ async fn datafiles( .try_flatten()) } -fn assign_first_row_ids( - manifest_path: &str, - content: ManifestListContent, - manifest_first_row_id: Option, - entries: &mut [ManifestEntry], -) -> Result<(), Error> { - if content != ManifestListContent::Data { - return Ok(()); - } - - let Some(mut next_row_id) = manifest_first_row_id else { - for entry in entries { - *entry.data_file_mut().first_row_id_mut() = None; - } - return Ok(()); - }; - - if next_row_id < 0 { - return Err(Error::InvalidFormat(format!( - "Manifest {manifest_path} has a negative first_row_id: {next_row_id}" - ))); - } - - for entry in entries { - if *entry.status() == Status::Deleted { - continue; - } - let data_file = entry.data_file_mut(); - if data_file.first_row_id().is_some() { - continue; - } - let record_count = *data_file.record_count(); - if record_count < 0 { - return Err(Error::InvalidFormat(format!( - "Data file {} has a negative record count: {record_count}", - data_file.file_path() - ))); - } - *data_file.first_row_id_mut() = Some(next_row_id); - next_row_id = next_row_id.checked_add(record_count).ok_or_else(|| { - Error::InvalidFormat(format!( - "Row ID overflow while reading manifest {manifest_path}" - )) - })?; - } - Ok(()) -} - /// delete all datafiles, manifests and metadata files, does not remove table from catalog pub(crate) async fn delete_all_table_files( metadata: &TableMetadata, @@ -568,106 +516,8 @@ pub(crate) async fn delete_all_table_files( #[cfg(test)] mod tests { - use iceberg_rust_spec::spec::{ - manifest::{Content, DataFile, FileFormat, ManifestEntry, Status}, - manifest_list::Content as ManifestListContent, - table_metadata::FormatVersion, - values::{Struct, Value}, - }; use rstest::rstest; - use super::assign_first_row_ids; - - fn data_entry( - status: Status, - path: &str, - record_count: i64, - first_row_id: Option, - ) -> ManifestEntry { - let mut data_file = DataFile::builder(); - data_file - .with_content(Content::Data) - .with_file_path(path.to_owned()) - .with_file_format(FileFormat::Parquet) - .with_partition(Struct::from_iter(Vec::<(String, Option)>::new())) - .with_record_count(record_count) - .with_file_size_in_bytes(100) - .with_column_sizes(None) - .with_value_counts(None) - .with_null_value_counts(None) - .with_nan_value_counts(None) - .with_distinct_counts(None) - .with_lower_bounds(None) - .with_upper_bounds(None) - .with_first_row_id(first_row_id); - ManifestEntry::builder() - .with_format_version(FormatVersion::V3) - .with_status(status) - .with_data_file(data_file.build().expect("build data file")) - .with_sequence_number(1) - .build() - .expect("build manifest entry") - } - - #[test] - fn inherits_v3_first_row_ids_before_scan_filtering() { - let mut entries = vec![ - data_entry(Status::Added, "a.parquet", 2, None), - data_entry(Status::Existing, "explicit.parquet", 1, Some(100)), - data_entry(Status::Added, "b.parquet", 3, None), - data_entry(Status::Deleted, "deleted.parquet", 4, None), - ]; - - assign_first_row_ids( - "manifest.avro", - ManifestListContent::Data, - Some(10), - &mut entries, - ) - .expect("assign first row ids"); - - 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(12)); - assert_eq!(*entries[3].data_file().first_row_id(), None); - } - - #[test] - fn clears_inherited_row_ids_without_manifest_lineage() { - let mut entries = vec![data_entry(Status::Added, "old.parquet", 2, Some(100))]; - - assign_first_row_ids( - "manifest.avro", - ManifestListContent::Data, - None, - &mut entries, - ) - .expect("clear stale first row ids"); - - assert_eq!(*entries[0].data_file().first_row_id(), None); - } - - #[test] - fn rejects_invalid_first_row_id_ranges() { - let mut entries = vec![data_entry(Status::Added, "a.parquet", 2, None)]; - assert!(assign_first_row_ids( - "manifest.avro", - ManifestListContent::Data, - Some(-1), - &mut entries, - ) - .is_err()); - - let mut entries = vec![data_entry(Status::Added, "a.parquet", 2, None)]; - assert!(assign_first_row_ids( - "manifest.avro", - ManifestListContent::Data, - Some(i64::MAX), - &mut entries, - ) - .is_err()); - } - // ----------------------------------------------------------------------- // Placeholders for upstream scan + planning + metadata-table tests. // Scan planning lives partly in iceberg-rust and partly in datafusion_iceberg; diff --git a/iceberg-rust/src/table/transaction/append.rs b/iceberg-rust/src/table/transaction/append.rs index 1c7cf901..aaf476ce 100644 --- a/iceberg-rust/src/table/transaction/append.rs +++ b/iceberg-rust/src/table/transaction/append.rs @@ -8,7 +8,7 @@ use std::collections::HashMap; use crate::{ error::Error, - table::manifest_list::ManifestListReader, + table::manifest_list::{append_manifest, ManifestListReader, RowIdAssigner}, util::{cmp_with_priority, partition_struct_to_vec, summary_to_rectangle, try_sub, Rectangle}, }; @@ -120,6 +120,7 @@ pub(crate) struct SelectedManifest { pub(crate) fn select_manifest_partitioned( manifest_list_reader: ManifestListReader<&[u8]>, manifest_list_writer: &mut apache_avro::Writer>, + mut row_id_assigner: Option<&mut RowIdAssigner>, bounding_partition_values: &Rectangle, ) -> Result { let mut selected_data_state = None; @@ -140,37 +141,65 @@ pub(crate) fn select_manifest_partitioned( match manifest.content { iceberg_rust_spec::manifest_list::Content::Data => { - let Some((selected_bounds, selected_manifest)) = &selected_data_state else { + let Some((selected_bounds, _)) = &selected_data_state else { selected_data_state = Some((bounds, manifest)); continue; }; match selected_bounds.cmp_with_priority(&bounds)? { Ordering::Greater => { - manifest_list_writer.append_ser(selected_manifest)?; - selected_data_state = Some((bounds, manifest)); + let (_, selected_manifest) = selected_data_state + .replace((bounds, manifest)) + .ok_or_else(|| { + Error::InvalidFormat( + "selected data manifest is missing".to_string(), + ) + })?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + selected_manifest, + )?; continue; } _ => { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; continue; } } } iceberg_rust_spec::manifest_list::Content::Deletes => { - let Some((selected_bounds, selected_manifest)) = &selected_delete_state else { + let Some((selected_bounds, _)) = &selected_delete_state else { selected_delete_state = Some((bounds, manifest)); continue; }; match selected_bounds.cmp_with_priority(&bounds)? { Ordering::Greater => { - manifest_list_writer.append_ser(selected_manifest)?; - selected_delete_state = Some((bounds, manifest)); + let (_, selected_manifest) = selected_delete_state + .replace((bounds, manifest)) + .ok_or_else(|| { + Error::InvalidFormat( + "selected delete manifest is missing".to_string(), + ) + })?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + selected_manifest, + )?; continue; } _ => { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; continue; } } @@ -191,6 +220,7 @@ pub(crate) fn select_manifest_partitioned( pub(crate) fn select_manifest_unpartitioned( manifest_list_reader: ManifestListReader<&[u8]>, manifest_list_writer: &mut apache_avro::Writer>, + mut row_id_assigner: Option<&mut RowIdAssigner>, ) -> Result { let mut selected_data_state = None; let mut selected_delete_state = None; @@ -203,7 +233,7 @@ pub(crate) fn select_manifest_unpartitioned( match manifest.content { iceberg_rust_spec::manifest_list::Content::Data => { - let Some((selected_row_count, selected_manifest)) = &selected_data_state else { + let Some((selected_row_count, _)) = &selected_data_state else { selected_data_state = Some((row_count, manifest)); continue; }; @@ -215,16 +245,28 @@ pub(crate) fn select_manifest_unpartitioned( }; if selected_row_count.is_some_and(|x| x > row_count) { - manifest_list_writer.append_ser(selected_manifest)?; - selected_data_state = Some((Some(row_count), manifest)); + let (_, selected_manifest) = selected_data_state + .replace((Some(row_count), manifest)) + .ok_or_else(|| { + Error::InvalidFormat("selected data manifest is missing".to_string()) + })?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + selected_manifest, + )?; continue; } else { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; continue; } } iceberg_rust_spec::manifest_list::Content::Deletes => { - let Some((selected_row_count, selected_manifest)) = &selected_delete_state else { + let Some((selected_row_count, _)) = &selected_delete_state else { selected_delete_state = Some((row_count, manifest)); continue; }; @@ -236,11 +278,23 @@ pub(crate) fn select_manifest_unpartitioned( }; if selected_row_count.is_some_and(|x| x > row_count) { - manifest_list_writer.append_ser(selected_manifest)?; - selected_delete_state = Some((Some(row_count), manifest)); + let (_, selected_manifest) = selected_delete_state + .replace((Some(row_count), manifest)) + .ok_or_else(|| { + Error::InvalidFormat("selected delete manifest is missing".to_string()) + })?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + selected_manifest, + )?; continue; } else { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; continue; } } diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index 33166f41..966fccd8 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -33,7 +33,9 @@ use tokio::task::JoinHandle; use tracing::{debug, instrument}; use crate::table::manifest::{FilteredManifestStats, ManifestWriter}; -use crate::table::manifest_list::{ManifestListReader, ManifestListWriter}; +use crate::table::manifest_list::{ + hydrate_v1_manifest_list_row_counts, ManifestListReader, ManifestListWriter, +}; use crate::table::transaction::append::append_summary; use crate::{ catalog::commit::{TableRequirement, TableUpdate}, @@ -167,8 +169,19 @@ impl Operation { prefetch_manifest_list(old_snapshot, &object_store) { let bytes = manifest_list_bytes.await??; + let hydrated_bytes = if table_metadata.format_version != FormatVersion::V1 { + hydrate_v1_manifest_list_row_counts( + &bytes, + table_metadata, + object_store.clone(), + ) + .await? + } else { + None + }; + let bytes = hydrated_bytes.as_deref().unwrap_or(&bytes); ManifestListWriter::from_existing( - &bytes, + bytes, all_files.iter(), manifest_list_schema, table_metadata, @@ -327,8 +340,19 @@ impl Operation { prefetch_manifest_list(old_snapshot, &object_store) { let bytes = manifest_list_bytes.await??; + let hydrated_bytes = if table_metadata.format_version != FormatVersion::V1 { + hydrate_v1_manifest_list_row_counts( + &bytes, + table_metadata, + object_store.clone(), + ) + .await? + } else { + None + }; + let bytes = hydrated_bytes.as_deref().unwrap_or(&bytes); ManifestListWriter::from_existing( - &bytes, + bytes, data_files_iter, manifest_list_schema, table_metadata, @@ -667,10 +691,21 @@ impl Operation { let bytes = prefetch_manifest_list(Some(old_snapshot), &object_store) .unwrap() .await??; + let hydrated_bytes = if table_metadata.format_version != FormatVersion::V1 { + hydrate_v1_manifest_list_row_counts( + &bytes, + table_metadata, + object_store.clone(), + ) + .await? + } else { + None + }; + let bytes = hydrated_bytes.as_deref().unwrap_or(&bytes); // Validate that all manifests specified in files_to_overwrite actually exist in the current snapshot let current_manifest_paths: HashSet = { - let manifest_list_reader = ManifestListReader::new(&bytes[..], table_metadata)?; + let manifest_list_reader = ManifestListReader::new(bytes, table_metadata)?; manifest_list_reader .map(|entry| entry.map(|e| e.manifest_path.clone())) .collect::, _>>()? @@ -691,14 +726,14 @@ impl Operation { let (mut manifest_list_writer, manifests_to_overwrite) = if n_data_files == 0 { ManifestListWriter::from_existing_for_deletion( - &bytes, + bytes, &manifests_to_overwrite, manifest_list_schema, table_metadata, )? } else { ManifestListWriter::from_existing_without_overwrites( - &bytes, + bytes, data_files_iter, &manifests_to_overwrite, manifest_list_schema, @@ -1258,7 +1293,7 @@ pub fn compute_n_splits( mod tests { use super::*; use crate::table::ManifestReader; - use futures::executor::block_on; + use futures::{executor::block_on, TryStreamExt}; use iceberg_rust_spec::manifest::FileFormat; use iceberg_rust_spec::spec::schema::SchemaBuilder; use iceberg_rust_spec::spec::table_metadata::TableMetadataBuilder; @@ -1377,6 +1412,44 @@ mod tests { .unwrap() } + async fn clear_current_v1_manifest_list_counts( + metadata: &TableMetadata, + store: &Arc, + ) { + let legacy_snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let legacy_manifest_list_path = strip_prefix(legacy_snapshot.manifest_list()).into(); + let legacy_manifest_list = store + .get(&legacy_manifest_list_path) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let mut legacy_manifests = ManifestListReader::new(&legacy_manifest_list[..], metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + for manifest in &mut legacy_manifests { + manifest.added_files_count = None; + manifest.existing_files_count = None; + manifest.deleted_files_count = None; + manifest.added_rows_count = None; + manifest.existing_rows_count = None; + manifest.deleted_rows_count = None; + } + let mut legacy_writer = apache_avro::Writer::new(manifest_list_schema_v1(), Vec::new()); + for manifest in legacy_manifests { + legacy_writer.append_ser(manifest).unwrap(); + } + store + .put( + &legacy_manifest_list_path, + legacy_writer.into_inner().unwrap().into(), + ) + .await + .unwrap(); + } + #[tokio::test] async fn v3_appends_assign_contiguous_row_ranges() { let mut metadata = sample_metadata(&[], None, &[]); @@ -1531,6 +1604,209 @@ mod tests { assert_eq!(metadata.next_row_id, 20); } + #[tokio::test] + async fn v3_append_assigns_row_ids_to_pre_upgrade_manifests() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.format_version = FormatVersion::V1; + let store = Arc::new(InMemory::new()); + + let legacy_append = Operation::Append { + branch: None, + data_files: vec![data_file("s3://tests/table/data/legacy.parquet", 3)], + delete_files: Vec::new(), + additional_summary: None, + }; + let (_, legacy_updates) = legacy_append + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, legacy_updates).unwrap(); + + clear_current_v1_manifest_list_counts(&metadata, &store).await; + + metadata.format_version = FormatVersion::V3; + metadata.next_row_id = 0; + let v3_append = Operation::Append { + branch: None, + data_files: vec![data_file("s3://tests/table/data/v3.parquet", 5)], + delete_files: Vec::new(), + additional_summary: None, + }; + let (_, v3_updates) = v3_append.execute(&metadata, store.clone()).await.unwrap(); + let snapshot = v3_updates + .iter() + .find_map(|update| match update { + TableUpdate::AddSnapshot { snapshot } => Some(snapshot), + _ => None, + }) + .unwrap(); + + assert_eq!(*snapshot.first_row_id(), Some(0)); + assert_eq!(*snapshot.added_rows(), Some(8)); + + let manifest_list_bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&manifest_list_bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + let mut first_row_ids = manifests + .iter() + .map(|manifest| manifest.first_row_id) + .collect::>(); + first_row_ids.sort(); + assert_eq!(first_row_ids, vec![Some(0), Some(3)]); + + let mut file_row_ids = + crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| { + ( + entry.data_file().file_path().clone(), + *entry.data_file().first_row_id(), + ) + }) + .try_collect::>() + .await + .unwrap(); + file_row_ids.sort(); + assert_eq!( + file_row_ids, + vec![ + ("s3://tests/table/data/legacy.parquet".to_string(), Some(0)), + ("s3://tests/table/data/v3.parquet".to_string(), Some(3)), + ] + ); + + crate::catalog::commit::apply_table_updates(&mut metadata, v3_updates).unwrap(); + assert_eq!(metadata.next_row_id, 8); + } + + #[tokio::test] + async fn v2_append_promotes_v1_manifest_list_with_missing_counts() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.format_version = FormatVersion::V1; + let store = Arc::new(InMemory::new()); + + let legacy_append = Operation::Append { + branch: None, + data_files: vec![data_file("s3://tests/table/data/legacy.parquet", 3)], + delete_files: Vec::new(), + additional_summary: None, + }; + let (_, legacy_updates) = legacy_append + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, legacy_updates).unwrap(); + clear_current_v1_manifest_list_counts(&metadata, &store).await; + + metadata.format_version = FormatVersion::V2; + let v2_append = Operation::Append { + branch: None, + data_files: vec![data_file("s3://tests/table/data/v2.parquet", 5)], + delete_files: Vec::new(), + additional_summary: None, + }; + let (_, v2_updates) = v2_append.execute(&metadata, store.clone()).await.unwrap(); + let snapshot = v2_updates + .iter() + .find_map(|update| match update { + TableUpdate::AddSnapshot { snapshot } => Some(snapshot), + _ => None, + }) + .unwrap(); + let manifest_list_bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&manifest_list_bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + + assert!(!manifests.is_empty()); + assert!(manifests.iter().all(|manifest| { + manifest.format_version == FormatVersion::V2 + && manifest.first_row_id.is_none() + && manifest.added_files_count.is_some() + && manifest.existing_files_count.is_some() + && manifest.deleted_files_count.is_some() + && manifest.added_rows_count.is_some() + && manifest.existing_rows_count.is_some() + && manifest.deleted_rows_count.is_some() + })); + let mut paths = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| entry.data_file().file_path().clone()) + .try_collect::>() + .await + .unwrap(); + paths.sort(); + assert_eq!( + paths, + vec![ + "s3://tests/table/data/legacy.parquet".to_string(), + "s3://tests/table/data/v2.parquet".to_string(), + ] + ); + } + + #[test] + fn v3_commit_accepts_forward_row_id_gaps_but_rejects_overlap() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.format_version = FormatVersion::V3; + metadata.next_row_id = 10; + + let gap_snapshot = SnapshotBuilder::default() + .with_snapshot_id(1) + .with_sequence_number(1) + .with_timestamp_ms(1) + .with_manifest_list("s3://tests/table/metadata/gap.avro".to_string()) + .with_summary(Summary::default()) + .with_first_row_id(15) + .with_added_rows(3) + .build() + .unwrap(); + crate::catalog::commit::apply_table_updates( + &mut metadata, + vec![TableUpdate::AddSnapshot { + snapshot: gap_snapshot, + }], + ) + .unwrap(); + assert_eq!(metadata.next_row_id, 18); + + let overlapping_snapshot = SnapshotBuilder::default() + .with_snapshot_id(2) + .with_sequence_number(2) + .with_timestamp_ms(2) + .with_manifest_list("s3://tests/table/metadata/overlap.avro".to_string()) + .with_summary(Summary::default()) + .with_first_row_id(17) + .with_added_rows(1) + .build() + .unwrap(); + let result = crate::catalog::commit::apply_table_updates( + &mut metadata, + vec![TableUpdate::AddSnapshot { + snapshot: overlapping_snapshot, + }], + ); + assert!(matches!(result, Err(Error::InvalidFormat(_)))); + assert_eq!(metadata.next_row_id, 18); + } + #[tokio::test] async fn v3_manifest_rewrites_remain_rejected() { let mut metadata = sample_metadata(&[], None, &[]); diff --git a/iceberg-rust/src/table/transaction/overwrite.rs b/iceberg-rust/src/table/transaction/overwrite.rs index 31a041c3..c7157045 100644 --- a/iceberg-rust/src/table/transaction/overwrite.rs +++ b/iceberg-rust/src/table/transaction/overwrite.rs @@ -4,7 +4,7 @@ use iceberg_rust_spec::manifest_list::ManifestListEntry; use crate::{ error::Error, - table::manifest_list::ManifestListReader, + table::manifest_list::{append_manifest, ManifestListReader, RowIdAssigner}, util::{summary_to_rectangle, Rectangle}, }; @@ -19,6 +19,7 @@ pub(crate) struct OverwriteManifest { pub(crate) fn select_manifest_without_overwrites_partitioned( manifest_list_reader: ManifestListReader<&[u8]>, manifest_list_writer: &mut apache_avro::Writer>, + mut row_id_assigner: Option<&mut RowIdAssigner>, bounding_partition_values: &Rectangle, overwrites: &HashSet, ) -> Result { @@ -48,7 +49,7 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( let old = selected_state.replace((bounds, manifest)); if let Some((_, old)) = old { if !overwrites.contains(&old.manifest_path) { - manifest_list_writer.append_ser(old)?; + append_manifest(manifest_list_writer, row_id_assigner.as_deref_mut(), old)?; } else { manifests_to_overwrite.push(old); } @@ -57,7 +58,11 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( } _ => { if !overwrites.contains(&manifest.manifest_path) { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; } else { manifests_to_overwrite.push(manifest); } @@ -79,6 +84,7 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( pub(crate) fn select_manifest_without_overwrites_unpartitioned( manifest_list_reader: ManifestListReader<&[u8]>, manifest_list_writer: &mut apache_avro::Writer>, + mut row_id_assigner: Option<&mut RowIdAssigner>, overwrites: &HashSet, ) -> Result { let mut selected_state = None; @@ -105,7 +111,7 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( let old = selected_state.replace((Some(row_count), manifest)); if let Some((_, old)) = old { if !overwrites.contains(&old.manifest_path) { - manifest_list_writer.append_ser(old)?; + append_manifest(manifest_list_writer, row_id_assigner.as_deref_mut(), old)?; } else { manifests_to_overwrite.push(old); } @@ -113,7 +119,11 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( continue; } else { if !overwrites.contains(&manifest.manifest_path) { - manifest_list_writer.append_ser(manifest)?; + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; } else { manifests_to_overwrite.push(manifest); }