diff --git a/.typos.toml b/.typos.toml index e9fa0028f5..796e778861 100644 --- a/.typos.toml +++ b/.typos.toml @@ -21,6 +21,7 @@ extend-ignore-identifiers-re = ["^bimap$"] [default.extend-words] AGS = "AGS" ags = "ags" +mor = "mor" [files] extend-exclude = ["**/testdata", "CHANGELOG.md", "**/public-api.txt"] diff --git a/Cargo.lock b/Cargo.lock index 87b228a871..a0966ca837 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3773,6 +3773,7 @@ dependencies = [ "bimap", "bytes", "chrono", + "crc32fast", "derive_builder", "expect-test", "fastnum", @@ -3886,6 +3887,7 @@ dependencies = [ "chrono", "http 1.4.2", "iceberg", + "iceberg-storage-opendal", "iceberg_test_utils", "itertools 0.13.0", "mockito", diff --git a/Cargo.toml b/Cargo.toml index a394ec107c..1d2307c5e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,6 +76,7 @@ bytes = "1.11" cfg-if = "1" chrono = "0.4.41" clap = { version = "4.5.48", features = ["derive", "cargo"] } +crc32fast = "1.5" dashmap = "6" datafusion = "54.1.0" datafusion-cli = "54.1.0" diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml index 247709efd4..1c7bf8b678 100644 --- a/crates/catalog/rest/Cargo.toml +++ b/crates/catalog/rest/Cargo.toml @@ -44,6 +44,7 @@ typed-builder = { workspace = true } uuid = { workspace = true, features = ["v4"] } [dev-dependencies] +iceberg-storage-opendal = { workspace = true } iceberg_test_utils = { path = "../../test_utils", features = ["tests"] } mockito = { workspace = true } tokio = { workspace = true } diff --git a/crates/catalog/rest/examples/pulse_dv_realdata.rs b/crates/catalog/rest/examples/pulse_dv_realdata.rs new file mode 100644 index 0000000000..3ed6fb3054 --- /dev/null +++ b/crates/catalog/rest/examples/pulse_dv_realdata.rs @@ -0,0 +1,158 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Real-data, cross-engine validation of V3 deletion-vector writes. +//! +//! Writes a `deletion-vector-v1` to a REAL Iceberg table via a REST catalog +//! (Polaris), so an INDEPENDENT engine (Doris / Spark / DuckDB) can read the +//! table back and confirm the deletes were applied. This is the gate before +//! opening the upstream RowDelta MoR DV-write PR (#2203). +//! +//! Prereqs: a clean V3 table with one data file and NO pre-existing deletion +//! vector (create it via Doris/Spark/DuckDB first), and a port-forward to +//! Polaris. Run from a host with S3 access for the warehouse bucket. +//! +//! ```bash +//! kubectl port-forward svc/polaris -n pulse-data 8181:8181 & +//! POLARIS_URI=http://localhost:8181/api/catalog \ +//! POLARIS_CREDENTIAL="$(kubectl get secret -n pulse-compute polaris-svc-spark \ +//! -o jsonpath='{.data.client-id}' | base64 -d):$(kubectl get secret -n \ +//! pulse-compute polaris-svc-spark -o jsonpath='{.data.client-secret}' | base64 -d)" \ +//! POLARIS_WAREHOUSE=bronze_dbnew \ +//! DV_NAMESPACE=zz_compactbench DV_TABLE=zz_rust_dv_test DV_DELETE_COUNT=3 \ +//! cargo run -p iceberg-catalog-rest --example pulse_dv_realdata +//! ``` +//! +//! Then verify in Doris: `SELECT COUNT(*) FROM bronze_dbnew.zz_compactbench.zz_rust_dv_test;` +//! should drop by `DV_DELETE_COUNT`. + +use std::collections::HashMap; +use std::sync::Arc; + +use iceberg::delete_vector::DeleteVector; +use iceberg::spec::DataContentType; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::{Catalog, CatalogBuilder, TableIdent}; +use iceberg_catalog_rest::RestCatalogBuilder; +use iceberg_storage_opendal::OpenDalResolvingStorageFactory; +use uuid::Uuid; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let uri = std::env::var("POLARIS_URI")?; + let credential = std::env::var("POLARIS_CREDENTIAL")?; + let warehouse = std::env::var("POLARIS_WAREHOUSE").unwrap_or_else(|_| "bronze_dbnew".into()); + let namespace = std::env::var("DV_NAMESPACE").unwrap_or_else(|_| "zz_compactbench".into()); + let table_name = std::env::var("DV_TABLE").unwrap_or_else(|_| "zz_rust_dv_test".into()); + let delete_count: u64 = std::env::var("DV_DELETE_COUNT") + .unwrap_or_else(|_| "3".into()) + .parse()?; + + // --- connect to Polaris (REST + OAuth2; S3 creds vended by the catalog) --- + let mut props = HashMap::new(); + props.insert("uri".to_string(), uri.clone()); + props.insert("warehouse".to_string(), warehouse); + props.insert("credential".to_string(), credential); + props.insert("scope".to_string(), "PRINCIPAL_ROLE:ALL".to_string()); + props.insert( + "oauth2-server-uri".to_string(), + format!("{}/v1/oauth/tokens", uri.trim_end_matches('/')), + ); + let catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(OpenDalResolvingStorageFactory::new())) + .load("polaris", props) + .await?; + + let ident = TableIdent::from_strs([namespace.as_str(), table_name.as_str()])?; + let table = catalog.load_table(&ident).await?; + println!( + "loaded {ident:?} (format_version={:?})", + table.metadata().format_version() + ); + + // --- find a live Data file in the current snapshot --- + let snapshot = table + .metadata() + .current_snapshot() + .ok_or("table has no current snapshot")?; + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + let mut chosen = None; + for manifest_file in manifest_list.entries() { + let manifest = manifest_file.load_manifest(table.file_io()).await?; + for entry in manifest.entries() { + if entry.is_alive() && entry.data_file().content_type() == DataContentType::Data { + chosen = Some(entry.data_file().clone()); + break; + } + } + if chosen.is_some() { + break; + } + } + let data_file = chosen.ok_or("no live data file found in the table")?; + let referenced = data_file.file_path().to_string(); + let total_rows = data_file.record_count(); + println!("target data file: {referenced} ({total_rows} rows)"); + if delete_count > total_rows { + return Err(format!("DV_DELETE_COUNT {delete_count} > rows in file {total_rows}").into()); + } + + // --- build a DV deleting the first `delete_count` positions --- + let mut dv = DeleteVector::default(); + for pos in 0..delete_count { + dv.insert(pos); + } + + // --- write the DV to a Puffin file and build the PositionDeletes DataFile --- + let dv_location = format!( + "{}/data/rust-dv-{}.puffin", + table.metadata().location().trim_end_matches('/'), + Uuid::now_v7() + ); + let t_write_start = std::time::Instant::now(); + let dv_data_file = dv + .write_to_puffin_file( + table.file_io(), + dv_location.clone(), + referenced.clone(), + data_file.partition().clone(), + table.metadata().default_partition_spec_id(), + ) + .await?; + let t_write = t_write_start.elapsed(); + + // --- commit via RowDelta -> content=Deletes manifest + Operation::Delete --- + let t_commit_start = std::time::Instant::now(); + let tx = Transaction::new(&table); + let action = tx.row_delta().add_delete_files(vec![dv_data_file]); + let tx = action.apply(tx)?; + let updated = tx.commit(&catalog).await?; + let t_commit = t_commit_start.elapsed(); + println!( + "BENCH K={delete_count} write_to_puffin={:.3}s commit={:.3}s total={:.3}s", + t_write.as_secs_f64(), + t_commit.as_secs_f64(), + (t_write + t_commit).as_secs_f64() + ); + + println!( + "COMMITTED. new snapshot_id={:?}. Now verify with an independent engine \ + (Doris/Spark): COUNT(*) should be (previous count - {delete_count}).", + updated.metadata().current_snapshot_id() + ); + Ok(()) +} diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index c33445c64c..733779d741 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -52,6 +52,7 @@ base64 = { workspace = true } bimap = { workspace = true } bytes = { workspace = true } chrono = { workspace = true } +crc32fast = { workspace = true } derive_builder = { workspace = true } expect-test = { workspace = true } fastnum = { workspace = true } diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 610675c83d..f518a86781 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -170,6 +170,30 @@ impl serde_core::ser::Serialize for iceberg::compression::CompressionCodec pub fn iceberg::compression::CompressionCodec::serialize(&self, serializer: S) -> core::result::Result<::Ok, ::Error> impl<'de> serde_core::de::Deserialize<'de> for iceberg::compression::CompressionCodec pub fn iceberg::compression::CompressionCodec::deserialize>(deserializer: D) -> core::result::Result::Error> +pub mod iceberg::delete_vector +pub struct iceberg::delete_vector::DeleteVector +impl iceberg::delete_vector::DeleteVector +pub fn iceberg::delete_vector::DeleteVector::from_puffin_blob(blob: iceberg::puffin::Blob) -> iceberg::Result +pub fn iceberg::delete_vector::DeleteVector::insert(&mut self, pos: u64) -> bool +pub fn iceberg::delete_vector::DeleteVector::insert_positions(&mut self, positions: &[u64]) -> iceberg::Result +pub fn iceberg::delete_vector::DeleteVector::is_empty(&self) -> bool +pub fn iceberg::delete_vector::DeleteVector::iter(&self) -> iceberg::delete_vector::DeleteVectorIterator<'_> +pub fn iceberg::delete_vector::DeleteVector::len(&self) -> u64 +pub fn iceberg::delete_vector::DeleteVector::new(roaring_treemap: roaring::treemap::RoaringTreemap) -> iceberg::delete_vector::DeleteVector +pub fn iceberg::delete_vector::DeleteVector::to_puffin_blob(&self, properties: std::collections::hash::map::HashMap) -> iceberg::Result +pub async fn iceberg::delete_vector::DeleteVector::write_to_puffin_file(&self, file_io: &iceberg::io::FileIO, location: alloc::string::String, referenced_data_file: alloc::string::String, partition: iceberg::spec::Struct, partition_spec_id: i32) -> iceberg::Result +impl core::default::Default for iceberg::delete_vector::DeleteVector +pub fn iceberg::delete_vector::DeleteVector::default() -> iceberg::delete_vector::DeleteVector +impl core::fmt::Debug for iceberg::delete_vector::DeleteVector +pub fn iceberg::delete_vector::DeleteVector::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::ops::bit::BitOrAssign for iceberg::delete_vector::DeleteVector +pub fn iceberg::delete_vector::DeleteVector::bitor_assign(&mut self, other: Self) +pub struct iceberg::delete_vector::DeleteVectorIterator<'a> +impl iceberg::delete_vector::DeleteVectorIterator<'_> +pub fn iceberg::delete_vector::DeleteVectorIterator<'_>::advance_to(&mut self, pos: u64) +impl core::iter::traits::iterator::Iterator for iceberg::delete_vector::DeleteVectorIterator<'_> +pub type iceberg::delete_vector::DeleteVectorIterator<'_>::Item = u64 +pub fn iceberg::delete_vector::DeleteVectorIterator<'_>::next(&mut self) -> core::option::Option pub mod iceberg::encryption pub mod iceberg::encryption::kms pub struct iceberg::encryption::kms::GeneratedKey @@ -1249,10 +1273,18 @@ impl iceberg::puffin::PuffinReader pub async fn iceberg::puffin::PuffinReader::blob(&self, blob_metadata: &iceberg::puffin::BlobMetadata) -> iceberg::Result pub async fn iceberg::puffin::PuffinReader::file_metadata(&self) -> iceberg::Result<&iceberg::puffin::FileMetadata> pub fn iceberg::puffin::PuffinReader::new(input_file: iceberg::io::InputFile) -> Self +pub struct iceberg::puffin::PuffinWriteResult +pub iceberg::puffin::PuffinWriteResult::blobs_metadata: alloc::vec::Vec +pub iceberg::puffin::PuffinWriteResult::file_size_in_bytes: u64 +impl core::clone::Clone for iceberg::puffin::PuffinWriteResult +pub fn iceberg::puffin::PuffinWriteResult::clone(&self) -> iceberg::puffin::PuffinWriteResult +impl core::fmt::Debug for iceberg::puffin::PuffinWriteResult +pub fn iceberg::puffin::PuffinWriteResult::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::puffin::PuffinWriter impl iceberg::puffin::PuffinWriter pub async fn iceberg::puffin::PuffinWriter::add(&mut self, blob: iceberg::puffin::Blob, compression_codec: iceberg::compression::CompressionCodec) -> iceberg::Result<()> pub async fn iceberg::puffin::PuffinWriter::close(self) -> iceberg::Result<()> +pub async fn iceberg::puffin::PuffinWriter::close_with_metadata(self) -> iceberg::Result pub async fn iceberg::puffin::PuffinWriter::new(output_file: &iceberg::io::OutputFile, properties: std::collections::hash::map::HashMap, compress_footer: bool) -> iceberg::Result pub const iceberg::puffin::APACHE_DATASKETCHES_THETA_V1: &str pub const iceberg::puffin::CREATED_BY_PROPERTY: &str @@ -3154,6 +3186,7 @@ pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::tr pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::append::FastAppendAction pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::sort_order::ReplaceSortOrderAction +pub fn iceberg::transaction::Transaction::row_delta(&self) -> iceberg::transaction::row_delta::RowDeltaAction pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::update_location::UpdateLocationAction pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::update_schema::UpdateSchemaAction pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::update_statistics::UpdateStatisticsAction diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index df8a10193c..1ad074a731 100644 --- a/crates/iceberg/src/delete_vector.rs +++ b/crates/iceberg/src/delete_vector.rs @@ -15,20 +15,42 @@ // specific language governing permissions and limitations // under the License. +//! Iceberg V3 deletion vectors (`deletion-vector-v1`): a roaring-bitmap-backed set +//! of deleted row positions, serialized to and from Puffin blobs and files. + +use std::collections::HashMap; +use std::io::Cursor; use std::ops::BitOrAssign; +use crc32fast::Hasher; use roaring::RoaringTreemap; use roaring::bitmap::Iter; use roaring::treemap::BitmapIter; +use crate::io::FileIO; +use crate::puffin::{Blob, CompressionCodec, DELETION_VECTOR_V1, PuffinWriter}; +use crate::spec::{DataContentType, DataFile, DataFileBuilder, DataFileFormat, Struct}; use crate::{Error, ErrorKind, Result}; +/// Iceberg `deletion-vector-v1` Puffin blob magic bytes (Iceberg Puffin spec; +/// ported from risingwavelabs/iceberg-rust #113 — design reference only). +const DELETION_VECTOR_MAGIC_BYTES: [u8; 4] = [0xD1, 0xD3, 0x39, 0x64]; +/// Minimum blob size: u32 length (4) + magic (4) + u32 crc (4). +const MIN_SERIALIZED_DELETION_VECTOR_BLOB: usize = 12; +/// Puffin blob property: deletion vector cardinality (number of deleted positions). +pub(crate) const DELETION_VECTOR_PROPERTY_CARDINALITY: &str = "cardinality"; +/// Puffin blob property: referenced data file path the DV applies to. +pub(crate) const DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE: &str = "referenced-data-file"; + +/// A set of deleted row positions backed by a 64-bit roaring bitmap — the in-memory +/// form of an Iceberg V3 `deletion-vector-v1`. #[derive(Debug, Default)] pub struct DeleteVector { inner: RoaringTreemap, } impl DeleteVector { + /// Creates a delete vector that wraps an existing roaring treemap of positions. #[allow(unused)] pub fn new(roaring_treemap: RoaringTreemap) -> DeleteVector { DeleteVector { @@ -36,11 +58,13 @@ impl DeleteVector { } } + /// Returns an iterator over the deleted row positions in ascending order. pub fn iter(&self) -> DeleteVectorIterator<'_> { let outer = self.inner.bitmaps(); DeleteVectorIterator { outer, inner: None } } + /// Marks row position `pos` as deleted; returns `true` if it was newly added. pub fn insert(&mut self, pos: u64) -> bool { self.inner.insert(pos) } @@ -64,10 +88,210 @@ impl DeleteVector { Ok(positions.len()) } + /// Returns the number of deleted row positions. #[allow(unused)] pub fn len(&self) -> u64 { self.inner.len() } + + /// Returns `true` if there are no deleted positions in this vector. + pub fn is_empty(&self) -> bool { + self.inner.is_empty() + } + + /// Serialize this delete vector into an Iceberg V3 `deletion-vector-v1` Puffin blob. + /// + /// Blob layout (Iceberg Puffin spec): `length(u32 BE = magic + bitmap) ‖ magic ‖ + /// portable 64-bit RoaringTreemap ‖ crc32(u32 BE over magic + bitmap)`. + /// `properties` must contain `cardinality` + `referenced-data-file`. + /// Ported from risingwavelabs/iceberg-rust #113 (design reference only). + pub fn to_puffin_blob(&self, properties: HashMap) -> Result { + Self::check_properties(&properties)?; + + let serialized_bitmap_size = self.inner.serialized_size(); + let combined_length = (DELETION_VECTOR_MAGIC_BYTES.len() + serialized_bitmap_size) as u32; + let mut data = Vec::with_capacity( + std::mem::size_of_val(&combined_length) + + DELETION_VECTOR_MAGIC_BYTES.len() + + serialized_bitmap_size + + 4, + ); + + data.extend_from_slice(&combined_length.to_be_bytes()); + data.extend_from_slice(&DELETION_VECTOR_MAGIC_BYTES); + + let bitmap_start = data.len(); + data.resize(bitmap_start + serialized_bitmap_size, 0); + { + let mut cursor = Cursor::new(&mut data[bitmap_start..]); + self.inner.serialize_into(&mut cursor).map_err(|err| { + Error::new( + ErrorKind::Unexpected, + "failed to serialize deletion vector bitmap".to_string(), + ) + .with_source(err) + })?; + } + + let mut hasher = Hasher::new(); + hasher.update(&data[4..]); + let crc = hasher.finalize(); + data.extend_from_slice(&crc.to_be_bytes()); + + Ok(Blob::builder() + .r#type(DELETION_VECTOR_V1.to_string()) + .fields(vec![]) + .snapshot_id(-1) + .sequence_number(-1) + .data(data) + .properties(properties) + .build()) + } + + /// Deserialize a delete vector from an Iceberg `deletion-vector-v1` Puffin blob. + pub fn from_puffin_blob(blob: Blob) -> Result { + if blob.blob_type() != DELETION_VECTOR_V1 { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("unsupported puffin blob type: {}", blob.blob_type()), + )); + } + + let data = blob.data(); + if data.len() < MIN_SERIALIZED_DELETION_VECTOR_BLOB { + return Err(Error::new( + ErrorKind::DataInvalid, + "serialized deletion vector blob too small".to_string(), + )); + } + + let magic = &data[4..8]; + if magic != DELETION_VECTOR_MAGIC_BYTES { + return Err(Error::new( + ErrorKind::DataInvalid, + "invalid deletion vector magic bytes".to_string(), + )); + } + + let combined_length = u32::from_be_bytes([data[0], data[1], data[2], data[3]]); + let expected_len = std::mem::size_of_val(&combined_length) + combined_length as usize + 4; + if expected_len != data.len() { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "serialized deletion vector length mismatch: expected {expected_len}, actual {}", + data.len() + ), + )); + } + + let bitmap_end = data.len() - 4; + let bitmap_data = &data[8..bitmap_end]; + + let mut hasher = Hasher::new(); + hasher.update(&data[4..bitmap_end]); + let expected_crc = hasher.finalize(); + let stored_crc = u32::from_be_bytes([ + data[data.len() - 4], + data[data.len() - 3], + data[data.len() - 2], + data[data.len() - 1], + ]); + if expected_crc != stored_crc { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("deletion vector crc mismatch: expected {expected_crc}, got {stored_crc}"), + )); + } + + let bitmap = + RoaringTreemap::deserialize_from(&mut Cursor::new(bitmap_data)).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + "failed to deserialize deletion vector bitmap".to_string(), + ) + .with_source(err) + })?; + + Ok(DeleteVector::new(bitmap)) + } + + fn check_properties(properties: &HashMap) -> Result<()> { + if !properties.contains_key(DELETION_VECTOR_PROPERTY_CARDINALITY) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector blob missing required property: {DELETION_VECTOR_PROPERTY_CARDINALITY}" + ), + )); + } + if !properties.contains_key(DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector blob missing required property: {DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE}" + ), + )); + } + Ok(()) + } + + /// Write this delete vector to a `deletion-vector-v1` Puffin file at `location` and + /// return the V3 `DataFile{content=PositionDeletes, …}` to feed + /// `RowDeltaAction::add_delete_files`. Connects DV serialization → Puffin file → + /// delete-file metadata (offset/length) in one step (cf. RW `deletion_vector_writer.rs`). + pub async fn write_to_puffin_file( + &self, + file_io: &FileIO, + location: String, + referenced_data_file: String, + partition: Struct, + partition_spec_id: i32, + ) -> Result { + let cardinality = self.len(); + let properties = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + cardinality.to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + referenced_data_file.clone(), + ), + ]); + let blob = self.to_puffin_blob(properties)?; + + let output_file = file_io.new_output(&location)?; + let mut writer = PuffinWriter::new(&output_file, HashMap::new(), false).await?; + writer.add(blob, CompressionCodec::None).await?; + let result = writer.close_with_metadata().await?; + let file_size = result.file_size_in_bytes; + let blob_metadata = result.blobs_metadata.first().ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + "puffin metadata is empty after writing deletion vector", + ) + })?; + + DataFileBuilder::default() + .content(DataContentType::PositionDeletes) + .file_path(location) + .file_format(DataFileFormat::Puffin) + .partition(partition) + .partition_spec_id(partition_spec_id) + .record_count(cardinality) + .file_size_in_bytes(file_size) + .referenced_data_file(Some(referenced_data_file)) + .content_offset(Some(blob_metadata.offset() as i64)) + .content_size_in_bytes(Some(blob_metadata.length() as i64)) + .build() + .map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("failed to build deletion vector data file: {err}"), + ) + }) + } } // Ideally, we'd just wrap `roaring::RoaringTreemap`'s iterator, `roaring::treemap::Iter` here. @@ -76,6 +300,7 @@ impl DeleteVector { // There is a PR open on roaring to add this (https://github.com/RoaringBitmap/roaring-rs/pull/314) // and if that gets merged then we can simplify `DeleteVectorIterator` here, refactoring `advance_to` // to just a wrapper around the underlying iterator's method. +/// Iterator over the deleted row positions of a [`DeleteVector`], in ascending order. pub struct DeleteVectorIterator<'a> { // NB: `BitMapIter` was only exposed publicly in https://github.com/RoaringBitmap/roaring-rs/pull/316 // which is not yet released. As a consequence our Cargo.toml temporarily uses a git reference for @@ -113,6 +338,7 @@ impl Iterator for DeleteVectorIterator<'_> { } impl DeleteVectorIterator<'_> { + /// Advances the iterator so the next yielded position is `>= pos`. pub fn advance_to(&mut self, pos: u64) { let hi = (pos >> 32) as u32; let lo = pos as u32; @@ -198,4 +424,319 @@ mod tests { let res = dv.insert_positions(&positions); assert!(res.is_err()); } + + fn dv_props() -> HashMap { + HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "0".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "s3://bucket/data/f.parquet".to_string(), + ), + ]) + } + + /// Self round-trip: serialize → Puffin blob → deserialize recovers the positions, + /// validating the frame (length, magic, crc) and serialize/deserialize symmetry. + #[test] + fn test_dv_puffin_blob_roundtrip() { + let positions = [1u64, 5, 42, 100, 1 << 33, (1u64 << 33) + 7]; + let mut dv = DeleteVector::default(); + for p in positions { + dv.insert(p); + } + let blob = dv.to_puffin_blob(dv_props()).unwrap(); + assert_eq!(blob.blob_type(), DELETION_VECTOR_V1); + + let restored = DeleteVector::from_puffin_blob(blob).unwrap(); + let mut got: Vec = restored.iter().collect(); + got.sort(); + assert_eq!(got, positions.to_vec()); + } + + /// Spark-compatibility proxy: parse the serialized bitmap with the EXACT algorithm + /// pyiceberg's `_deserialize_bitmap` uses — `[u64 LE bucket count]` then per bucket + /// `[u32 LE high-key + 32-bit portable RoaringBitmap]`. If this recovers the + /// positions, the bytes are Iceberg/Spark portable (no Spark needed for the signal). + #[test] + fn test_dv_blob_is_iceberg_portable() { + use std::io::Read; + + use roaring::RoaringBitmap; + + let positions = [3u64, 7, 100, (1u64 << 33) + 5]; + let mut dv = DeleteVector::default(); + for p in positions { + dv.insert(p); + } + let blob = dv.to_puffin_blob(dv_props()).unwrap(); + let data = blob.data(); + + // Frame (certain): [u32 BE len][magic][bitmap][u32 BE crc] + assert_eq!(&data[4..8], &DELETION_VECTOR_MAGIC_BYTES); + let bitmap = &data[8..data.len() - 4]; + + // pyiceberg portable parse + let mut cur = Cursor::new(bitmap); + let mut count_buf = [0u8; 8]; + cur.read_exact(&mut count_buf).unwrap(); + let n_buckets = u64::from_le_bytes(count_buf); + + let mut recovered: Vec = Vec::new(); + for _ in 0..n_buckets { + let mut key_buf = [0u8; 4]; + cur.read_exact(&mut key_buf).unwrap(); + let hi = u32::from_le_bytes(key_buf) as u64; + let bm = RoaringBitmap::deserialize_from(&mut cur).unwrap(); + for lo in bm.iter() { + recovered.push((hi << 32) | u64::from(lo)); + } + } + recovered.sort(); + assert_eq!( + recovered, + positions.to_vec(), + "serialized bitmap is NOT Iceberg-portable — roaring serialize_into header \ + differs from pyiceberg layout; switch to hand-rolled portable framing" + ); + } + + /// Piece 2 — full Puffin-FILE round-trip in Rust: write a DV blob to a real + /// Puffin file via `PuffinWriter`, read it back via `PuffinReader`, and recover + /// the deleted positions. Proves the Puffin file framing, not just the blob bytes. + #[tokio::test] + async fn test_dv_puffin_file_roundtrip() { + use tempfile::TempDir; + + use crate::io::FileIO; + use crate::puffin::{CompressionCodec, PuffinReader, PuffinWriter}; + + let positions = [2u64, 9, 256, (1u64 << 33) + 11]; + let mut dv = DeleteVector::default(); + for p in positions { + dv.insert(p); + } + assert!(!dv.is_empty()); + + let mut props = dv_props(); + props.insert( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + dv.len().to_string(), + ); + let blob = dv.to_puffin_blob(props).unwrap(); + + let tmp = TempDir::new().unwrap(); + let path_buf = tmp.path().join("dv.puffin"); + let path = path_buf.to_str().unwrap(); + + let file_io = FileIO::new_with_fs(); + let output = file_io.new_output(path).unwrap(); + let mut writer = PuffinWriter::new(&output, HashMap::new(), false) + .await + .unwrap(); + writer.add(blob, CompressionCodec::None).await.unwrap(); + writer.close().await.unwrap(); + + let input = output.to_input_file(); + let reader = PuffinReader::new(input); + let meta = reader.file_metadata().await.unwrap().clone(); + assert_eq!(meta.blobs.len(), 1); + let read_blob = reader.blob(meta.blobs.first().unwrap()).await.unwrap(); + + let restored = DeleteVector::from_puffin_blob(read_blob).unwrap(); + let mut got: Vec = restored.iter().collect(); + got.sort(); + assert_eq!(got, positions.to_vec()); + } + + /// Piece 2.5 — glue: write a DV to a Puffin file and get back a V3 + /// `DataFile{PositionDeletes}` (offset/size/referenced-file filled), then read the + /// written file back and recover the positions. + #[tokio::test] + async fn test_dv_write_to_puffin_file() { + use tempfile::TempDir; + + use crate::puffin::PuffinReader; + + let positions = [4u64, 11, 512, (1u64 << 33) + 3]; + let mut dv = DeleteVector::default(); + for p in positions { + dv.insert(p); + } + + let tmp = TempDir::new().unwrap(); + let path_buf = tmp.path().join("dv2.puffin"); + let location = path_buf.to_str().unwrap().to_string(); + let file_io = FileIO::new_with_fs(); + + let data_file = dv + .write_to_puffin_file( + &file_io, + location.clone(), + "s3://bucket/data/x.parquet".to_string(), + Struct::empty(), + 0, + ) + .await + .unwrap(); + + assert_eq!(data_file.content_type(), DataContentType::PositionDeletes); + assert_eq!( + data_file.referenced_data_file().as_deref(), + Some("s3://bucket/data/x.parquet") + ); + assert!(data_file.content_offset().is_some()); + assert!(data_file.content_size_in_bytes().is_some()); + + // The written Puffin file reads back to the same positions. + let input = file_io.new_input(&location).unwrap(); + let reader = PuffinReader::new(input); + let meta = reader.file_metadata().await.unwrap().clone(); + let blob = reader.blob(meta.blobs.first().unwrap()).await.unwrap(); + let restored = DeleteVector::from_puffin_blob(blob).unwrap(); + let mut got: Vec = restored.iter().collect(); + got.sort(); + assert_eq!(got, positions.to_vec()); + } + + /// Cross-implementation byte-parity with Apache Iceberg-Java: the serialized + /// `deletion-vector-v1` payload for positions {1,3,5,7,9} must be byte-identical + /// to the Java-produced golden fixture (lifted from apache/iceberg test resources + /// `small-alternating-values-position-index.bin` via apache/iceberg-go). Proves our + /// roaring serialization + framing exactly match the Iceberg-Java reference. + #[test] + fn test_dv_payload_byte_identical_to_java_golden() { + let golden: &[u8] = include_bytes!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/testdata/puffin/deletion-vector-v1-payload.bin" + )); + + let mut dv = DeleteVector::default(); + for p in [1u64, 3, 5, 7, 9] { + dv.insert(p); + } + let props = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "5".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "data/test.parquet".to_string(), + ), + ]); + let blob = dv.to_puffin_blob(props).unwrap(); + + assert_eq!( + blob.data(), + golden, + "DV payload must be byte-identical to the apache/iceberg Java golden fixture" + ); + } + + /// Empty deletion vector round-trips (mirrors iceberg-go `TestSerializeDVEmpty`). + #[test] + fn test_dv_empty_roundtrip() { + let dv = DeleteVector::default(); + let props = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "0".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "data/empty.parquet".to_string(), + ), + ]); + let blob = dv.to_puffin_blob(props).unwrap(); + let restored = DeleteVector::from_puffin_blob(blob).unwrap(); + assert!(restored.is_empty()); + assert_eq!(restored.len(), 0); + } + + /// Positions straddling the 2^31 (Java-signed) and 2^32 (roaring bucket) + /// boundaries round-trip (mirrors iceberg-go `TestSerializeDVLargePositions`). + #[test] + fn test_dv_boundary_positions_roundtrip() { + let positions = [100u64, 101, 2_147_483_747, 2_147_483_748, (1u64 << 32) | 42]; + let mut dv = DeleteVector::default(); + for p in positions { + dv.insert(p); + } + let props = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "5".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "data/boundary.parquet".to_string(), + ), + ]); + let blob = dv.to_puffin_blob(props).unwrap(); + let restored = DeleteVector::from_puffin_blob(blob).unwrap(); + let mut got: Vec = restored.iter().collect(); + got.sort(); + assert_eq!(got, positions.to_vec()); + } + + /// Byte-parity with Apache Iceberg-Java: the **empty** DV payload + /// (`empty-position-index.bin` from apache/iceberg test resources). + #[test] + fn test_dv_payload_byte_identical_to_java_empty() { + let golden: &[u8] = include_bytes!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/testdata/puffin/empty-position-index.bin" + )); + let dv = DeleteVector::default(); + let props = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "0".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "data/test.parquet".to_string(), + ), + ]); + let blob = dv.to_puffin_blob(props).unwrap(); + assert_eq!( + blob.data(), + golden, + "empty DV payload must be byte-identical to the Java golden fixture" + ); + } + + /// Byte-parity with Apache Iceberg-Java: small + large positions spanning two + /// 16-bit roaring containers — {100, 101, INT_MAX+100, INT_MAX+101} per + /// `small-and-large-values-position-index.bin` from apache/iceberg test resources. + #[test] + fn test_dv_payload_byte_identical_to_java_small_and_large() { + let golden: &[u8] = include_bytes!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/testdata/puffin/small-and-large-values-position-index.bin" + )); + let mut dv = DeleteVector::default(); + for p in [100u64, 101, 2_147_483_747, 2_147_483_748] { + dv.insert(p); + } + let props = HashMap::from([ + ( + DELETION_VECTOR_PROPERTY_CARDINALITY.to_string(), + "4".to_string(), + ), + ( + DELETION_VECTOR_PROPERTY_REFERENCED_DATA_FILE.to_string(), + "data/test.parquet".to_string(), + ), + ]); + let blob = dv.to_puffin_blob(props).unwrap(); + assert_eq!( + blob.data(), + golden, + "small+large DV payload must be byte-identical to the Java golden fixture" + ); + } } diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs index 4e346460f5..4fd02f79e7 100644 --- a/crates/iceberg/src/lib.rs +++ b/crates/iceberg/src/lib.rs @@ -98,7 +98,7 @@ pub mod encryption; pub mod test_utils; pub mod writer; -mod delete_vector; +pub mod delete_vector; pub mod metadata_columns; pub mod puffin; /// Utility functions and modules. diff --git a/crates/iceberg/src/puffin/mod.rs b/crates/iceberg/src/puffin/mod.rs index 0e054cac51..a660e25b59 100644 --- a/crates/iceberg/src/puffin/mod.rs +++ b/crates/iceberg/src/puffin/mod.rs @@ -51,7 +51,7 @@ mod reader; pub use reader::PuffinReader; mod writer; -pub use writer::PuffinWriter; +pub use writer::{PuffinWriteResult, PuffinWriter}; #[cfg(test)] mod test_utils; diff --git a/crates/iceberg/src/puffin/writer.rs b/crates/iceberg/src/puffin/writer.rs index 4af4970b04..d77c533be0 100644 --- a/crates/iceberg/src/puffin/writer.rs +++ b/crates/iceberg/src/puffin/writer.rs @@ -26,6 +26,16 @@ use crate::io::{FileWrite, OutputFile}; use crate::puffin::blob::Blob; use crate::puffin::metadata::{BlobMetadata, FileMetadata, Flag}; +/// Result of finalizing a Puffin file: total bytes written + per-blob metadata +/// (offsets/lengths), needed to build delete-file `DataFile`s for MoR commits. +#[derive(Debug, Clone)] +pub struct PuffinWriteResult { + /// Total size of the written Puffin file in bytes. + pub file_size_in_bytes: u64, + /// Metadata (incl. offset + length) for each blob written into the file. + pub blobs_metadata: Vec, +} + /// Puffin writer pub struct PuffinWriter { writer: Box, @@ -87,12 +97,21 @@ impl PuffinWriter { Ok(()) } - /// Finalizes the Puffin file - pub async fn close(mut self) -> Result<()> { + /// Finalizes the Puffin file. + pub async fn close(self) -> Result<()> { + self.close_with_metadata().await.map(|_| ()) + } + + /// Finalizes the Puffin file and returns the written size + per-blob metadata + /// (offsets/lengths) — needed to build a `DataFile` for an added MoR delete file. + pub async fn close_with_metadata(mut self) -> Result { self.write_header_once().await?; self.write_footer().await?; self.writer.close().await?; - Ok(()) + Ok(PuffinWriteResult { + file_size_in_bytes: self.num_bytes_written, + blobs_metadata: self.written_blobs_metadata, + }) } async fn write(&mut self, bytes: Bytes) -> Result<()> { diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 799abdeffd..d1644edceb 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -125,7 +125,7 @@ impl SnapshotProduceOperation for FastAppendOperation { async fn existing_manifest( &self, - snapshot_produce: &SnapshotProducer<'_>, + snapshot_produce: &mut SnapshotProducer<'_>, ) -> Result> { let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { return Ok(vec![]); diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 7b8e02da36..0eab4ee9b3 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -55,6 +55,7 @@ mod action; pub use action::*; mod append; mod expire_snapshots; +mod row_delta; mod snapshot; mod sort_order; mod update_location; @@ -75,6 +76,7 @@ use crate::table::Table; use crate::transaction::action::BoxedTransactionAction; use crate::transaction::append::FastAppendAction; use crate::transaction::expire_snapshots::ExpireSnapshotsAction; +use crate::transaction::row_delta::RowDeltaAction; use crate::transaction::sort_order::ReplaceSortOrderAction; use crate::transaction::update_location::UpdateLocationAction; use crate::transaction::update_properties::UpdatePropertiesAction; @@ -151,6 +153,16 @@ impl Transaction { FastAppendAction::new() } + /// Creates a row delta action for row-level modifications. + /// + /// RowDelta supports: + /// - Adding new data files (inserts) + /// - Removing data files (deletes in Copy-on-Write (COW) mode) + /// - Both operations in a single transaction (updates/merges) + pub fn row_delta(&self) -> RowDeltaAction { + RowDeltaAction::new() + } + /// Creates replace sort order action. pub fn replace_sort_order(&self) -> ReplaceSortOrderAction { ReplaceSortOrderAction::new() diff --git a/crates/iceberg/src/transaction/row_delta.rs b/crates/iceberg/src/transaction/row_delta.rs new file mode 100644 index 0000000000..44d167b8ba --- /dev/null +++ b/crates/iceberg/src/transaction/row_delta.rs @@ -0,0 +1,674 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::Result; +use crate::spec::{DataFile, ManifestContentType, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// Transaction action for Copy-on-Write row-level modifications (UPDATE, DELETE, MERGE INTO). +/// +/// Corresponds to `org.apache.iceberg.RowDelta` in the Java implementation. +pub struct RowDeltaAction { + added_data_files: Vec, + removed_data_files: Vec, + /// MoR delete files (position/equality deletes, incl. V3 deletion vectors) to add. + added_delete_files: Vec, + commit_uuid: Option, + snapshot_properties: HashMap, + starting_snapshot_id: Option, +} + +impl RowDeltaAction { + pub(crate) fn new() -> Self { + Self { + added_data_files: vec![], + removed_data_files: vec![], + added_delete_files: vec![], + commit_uuid: None, + snapshot_properties: HashMap::default(), + starting_snapshot_id: None, + } + } + + /// Add new data files (INSERT rows or Copy-on-Write rewritten files). + pub fn add_data_files(mut self, data_files: impl IntoIterator) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Mark existing data files as deleted (Copy-on-Write mode). + /// + /// Corresponds to `removeRows(DataFile)` in the Java implementation. + pub fn remove_data_files(mut self, data_files: impl IntoIterator) -> Self { + self.removed_data_files.extend(data_files); + self + } + + /// Add Merge-on-Read delete files (position/equality deletes, incl. V3 deletion + /// vectors). Written into a content=Deletes manifest at commit time. + pub fn add_delete_files(mut self, delete_files: impl IntoIterator) -> Self { + self.added_delete_files.extend(delete_files); + self + } + + /// Set the commit UUID used for manifest file naming. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Attach custom key/value metadata to the snapshot summary. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap) -> Self { + self.snapshot_properties = snapshot_properties; + self + } + + /// Reject the commit if the table has advanced past `snapshot_id` (optimistic concurrency). + pub fn validate_from_snapshot(mut self, snapshot_id: i64) -> Self { + self.starting_snapshot_id = Some(snapshot_id); + self + } +} + +#[async_trait] +impl TransactionAction for RowDeltaAction { + async fn commit(self: Arc, table: &Table) -> Result { + if let Some(expected_snapshot_id) = self.starting_snapshot_id + && table.metadata().current_snapshot_id() != Some(expected_snapshot_id) + { + return Err(crate::Error::new( + crate::ErrorKind::DataInvalid, + format!( + "Cannot commit RowDelta based on stale snapshot. Expected: {}, Current: {:?}", + expected_snapshot_id, + table.metadata().current_snapshot_id() + ), + )); + } + + let mut snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + ); + + // Validate newly added data files (partition value type-checks, etc.). + // removed_data_files are not re-validated: they are existing table files that were + // already validated when originally committed. This matches Java's MergingSnapshotProducer. + snapshot_producer.validate_added_data_files()?; + + // MoR delete files (position/equality deletes, incl. V3 deletion vectors) are + // written into a separate content=Deletes manifest by the snapshot producer. + snapshot_producer.set_added_delete_files(self.added_delete_files.clone()); + + let operation = RowDeltaOperation { + removed_data_files: self.removed_data_files.clone(), + has_added_data_files: !self.added_data_files.is_empty(), + has_added_delete_files: !self.added_delete_files.is_empty(), + }; + + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } +} + +struct RowDeltaOperation { + removed_data_files: Vec, + has_added_data_files: bool, + has_added_delete_files: bool, +} + +impl SnapshotProduceOperation for RowDeltaOperation { + /// Operation type (mirrors Java `BaseRowDelta.operation()`): + /// - Any data files removed → `Overwrite` + /// - MoR delete files added → `Overwrite` if data files also added, else `Delete` + /// - Only data files added (or nothing) → `Append` + fn operation(&self) -> Operation { + if !self.removed_data_files.is_empty() { + Operation::Overwrite + } else if self.has_added_delete_files { + if self.has_added_data_files { + Operation::Overwrite + } else { + Operation::Delete + } + } else { + Operation::Append + } + } + + /// Delete entries are handled inside `existing_manifest` by rewriting the manifest. + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result> { + Ok(vec![]) + } + + /// Returns manifest files for the new snapshot. + /// + /// For each manifest in the previous snapshot: + /// - If it contains any file being removed: rewrite it with DELETED entries for removed files + /// and EXISTING entries for survivors, preserving original sequence numbers. + /// - Otherwise: carry it forward unchanged. + /// + /// This matches Java's `ManifestFilterManager.filterManifestWithDeletedFiles` logic. + async fn existing_manifest( + &self, + snapshot_produce: &mut SnapshotProducer<'_>, + ) -> Result> { + let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { + return Ok(vec![]); + }; + + let manifest_list = snapshot_produce + .table + .manifest_list_reader(snapshot) + .load() + .await?; + + let deleted_paths: HashSet<&str> = self + .removed_data_files + .iter() + .map(|f| f.file_path()) + .collect(); + + let mut result = Vec::new(); + for manifest_file in manifest_list.entries() { + if !manifest_file.has_added_files() && !manifest_file.has_existing_files() { + continue; + } + + let manifest = manifest_file + .load_manifest(snapshot_produce.table.file_io()) + .await?; + + let needs_rewrite = manifest + .entries() + .iter() + .any(|e| e.is_alive() && deleted_paths.contains(e.data_file().file_path())); + + if !needs_rewrite { + result.push(manifest_file.clone()); + continue; + } + + // Rewrite: deleted files → DELETED (new snapshot_id, original seq nums preserved), + // surviving files → EXISTING (all original fields preserved). + let mut writer = snapshot_produce.new_manifest_writer(ManifestContentType::Data)?; + for entry in manifest.entries() { + if deleted_paths.contains(entry.data_file().file_path()) { + writer.add_delete_entry((**entry).clone())?; + } else if entry.is_alive() { + writer.add_existing_entry((**entry).clone())?; + } + // else: an already-DELETED (status=2) entry not removed by this + // operation — DROP it. Carrying it forward as EXISTING would + // RESURRECT a superseded file: for deletion vectors a data file + // accumulates one DELETED DV entry per prior rewrite, so + // resurrecting them yields multiple "live" DVs for one data + // file and readers fail with "Can't index multiple DVs". The + // deletion stays recorded in its originating snapshot's + // manifest; it does not belong in this one. + } + result.push(writer.write_manifest_file().await?); + } + + Ok(result) + } + + fn removed_data_files(&self) -> &[DataFile] { + &self.removed_data_files + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use crate::spec::{ + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, + ManifestStatus, Struct, TableMetadataBuilder, + }; + use crate::table::Table; + use crate::transaction::tests::make_v2_minimal_table; + use crate::transaction::{Transaction, TransactionAction}; + use crate::{TableIdent, TableUpdate}; + + fn make_data_file(table: &Table, path: &str, size: u64) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(size) + .record_count(10) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(100))])) + .build() + .unwrap() + } + + /// Build a table that has `snapshot` as its current snapshot, backed by the same FileIO. + async fn table_with_snapshot(base: &Table, snapshot: crate::spec::Snapshot) -> Table { + let updated_metadata = + TableMetadataBuilder::new_from_metadata(base.metadata_ref().as_ref().clone(), None) + .set_branch_snapshot(snapshot, MAIN_BRANCH) + .unwrap() + .build() + .unwrap() + .metadata; + + Table::builder() + .metadata(updated_metadata) + .metadata_location("s3://bucket/test/location/metadata/v2.json".to_string()) + .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap()) + .file_io(base.file_io().clone()) + .runtime(crate::test_utils::test_runtime()) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_row_delta_add_only() { + let table = make_v2_minimal_table(); + let data_file = make_data_file(&table, "test/1.parquet", 100); + let action = Transaction::new(&table) + .row_delta() + .add_data_files(vec![data_file]); + + let mut commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = commit.take_updates(); + + if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + assert_eq!(snapshot.summary().operation, crate::spec::Operation::Append); + } else { + panic!("expected AddSnapshot"); + } + } + + #[tokio::test] + async fn test_row_delta_with_snapshot_properties() { + let table = make_v2_minimal_table(); + let data_file = make_data_file(&table, "test/1.parquet", 100); + let mut props = std::collections::HashMap::new(); + props.insert("key".to_string(), "value".to_string()); + let action = Transaction::new(&table) + .row_delta() + .set_snapshot_properties(props) + .add_data_files(vec![data_file]); + + let mut commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = commit.take_updates(); + + if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + assert_eq!( + snapshot.summary().additional_properties.get("key").unwrap(), + "value" + ); + } else { + panic!("expected AddSnapshot"); + } + } + + #[tokio::test] + async fn test_row_delta_validate_from_snapshot() { + let table = make_v2_minimal_table(); + let data_file = make_data_file(&table, "test/1.parquet", 100); + let action = Transaction::new(&table) + .row_delta() + .validate_from_snapshot(99999) + .add_data_files(vec![data_file]); + + let result = Arc::new(action).commit(&table).await; + match result { + Ok(_) => panic!("expected DataInvalid error for stale snapshot"), + Err(e) => assert_eq!(e.kind(), crate::ErrorKind::DataInvalid), + } + } + + #[tokio::test] + async fn test_row_delta_empty_action() { + let table = make_v2_minimal_table(); + assert!( + Arc::new(Transaction::new(&table).row_delta()) + .commit(&table) + .await + .is_err() + ); + } + + #[tokio::test] + async fn test_row_delta_incompatible_partition_value() { + let table = make_v2_minimal_table(); + let bad_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("test/bad.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(10) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::string("wrong"))])) + .build() + .unwrap(); + let action = Transaction::new(&table) + .row_delta() + .add_data_files(vec![bad_file]); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + /// MoR: adding a position-delete file via RowDelta commits a content=Deletes + /// manifest and an `Operation::Delete` snapshot (replaces the old "errors" test + /// now that `add_delete_files` is implemented). + #[tokio::test] + async fn test_row_delta_add_delete_files_mor() { + let base = make_v2_minimal_table(); + + // S1: append a data file. + let data_file = make_data_file(&base, "test/data.parquet", 100); + let mut c1 = Arc::new( + Transaction::new(&base) + .fast_append() + .add_data_files(vec![data_file]), + ) + .commit(&base) + .await + .unwrap(); + let snap_s1 = if let TableUpdate::AddSnapshot { snapshot } = + c1.take_updates().into_iter().next().unwrap() + { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + let table_s1 = table_with_snapshot(&base, snap_s1).await; + + // S2: add a MoR position-delete file referencing the data file. + let delete_file = DataFileBuilder::default() + .content(DataContentType::PositionDeletes) + .file_path("test/pos-delete.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(50) + .record_count(3) + .partition_spec_id(table_s1.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(100))])) + .referenced_data_file(Some("test/data.parquet".to_string())) + .build() + .unwrap(); + let mut c2 = Arc::new( + Transaction::new(&table_s1) + .row_delta() + .add_delete_files(vec![delete_file]), + ) + .commit(&table_s1) + .await + .unwrap(); + let updates2 = c2.take_updates(); + let snap_s2 = if let TableUpdate::AddSnapshot { ref snapshot } = updates2[0] { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + + // Only delete files added (no data adds/removes) → Operation::Delete. + assert_eq!(snap_s2.summary().operation, crate::spec::Operation::Delete); + + // A PositionDeletes entry must exist in the new snapshot's manifests. + let manifest_list = table_s1 + .manifest_list_reader(&std::sync::Arc::new(snap_s2.clone())) + .load() + .await + .unwrap(); + let mut found_position_delete = false; + for manifest_file in manifest_list.entries() { + let manifest = manifest_file + .load_manifest(table_s1.file_io()) + .await + .unwrap(); + for entry in manifest.entries() { + if entry.data_file().content_type() == DataContentType::PositionDeletes { + found_position_delete = true; + } + } + } + assert!( + found_position_delete, + "expected a PositionDeletes entry in the RowDelta snapshot's manifests" + ); + } + + /// End-to-end CoW test: append two files, then remove one via RowDelta. + /// + /// Verifies: + /// - The removed file appears as DELETED with correct sequence numbers. + /// - The surviving file appears as EXISTING with correct sequence numbers. + /// - The new file appears as ADDED. + /// - The snapshot summary counts `deleted-data-files = 1`. + #[tokio::test] + async fn test_row_delta_cow_manifest_rewrite() { + let base_table = make_v2_minimal_table(); + + // --- S1: append file-A and file-B --- + let file_a = make_data_file(&base_table, "test/a.parquet", 100); + let file_b = make_data_file(&base_table, "test/b.parquet", 200); + + let action1 = Transaction::new(&base_table) + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]); + let mut commit1 = Arc::new(action1).commit(&base_table).await.unwrap(); + let updates1 = commit1.take_updates(); + + let snapshot_s1 = + if let TableUpdate::AddSnapshot { snapshot } = updates1.into_iter().next().unwrap() { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + + let table_s1 = table_with_snapshot(&base_table, snapshot_s1).await; + + // --- S2: remove file-A (CoW), add file-C --- + let file_c = make_data_file(&table_s1, "test/c.parquet", 300); + let action2 = Transaction::new(&table_s1) + .row_delta() + .remove_data_files(vec![file_a.clone()]) + .add_data_files(vec![file_c.clone()]); + let mut commit2 = Arc::new(action2).commit(&table_s1).await.unwrap(); + let updates2 = commit2.take_updates(); + + let snapshot_s2 = if let TableUpdate::AddSnapshot { ref snapshot } = updates2[0] { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + + assert_eq!( + snapshot_s2.summary().operation, + crate::spec::Operation::Overwrite + ); + + // Verify snapshot summary metrics + let props = &snapshot_s2.summary().additional_properties; + assert_eq!( + props.get("deleted-data-files").map(String::as_str), + Some("1"), + "summary should count 1 deleted file" + ); + + // Scan all manifest entries in S2 + let manifest_list = table_s1 + .manifest_list_reader(&std::sync::Arc::new(snapshot_s2.clone())) + .load() + .await + .unwrap(); + + let mut found_deleted_a = false; + let mut found_existing_b = false; + let mut found_added_c = false; + + for manifest_file in manifest_list.entries() { + let manifest = manifest_file + .load_manifest(table_s1.file_io()) + .await + .unwrap(); + for entry in manifest.entries() { + match entry.data_file().file_path() { + "test/a.parquet" => { + assert_eq!( + entry.status(), + ManifestStatus::Deleted, + "file-A must be DELETED" + ); + assert!( + entry.sequence_number().is_some(), + "DELETED entry must have sequence number" + ); + assert!( + entry.file_sequence_number.is_some(), + "DELETED entry must have file sequence number" + ); + found_deleted_a = true; + } + "test/b.parquet" => { + assert_eq!( + entry.status(), + ManifestStatus::Existing, + "file-B must be EXISTING" + ); + assert!( + entry.sequence_number().is_some(), + "EXISTING entry must have sequence number" + ); + found_existing_b = true; + } + "test/c.parquet" => { + found_added_c = true; + } + other => panic!("unexpected file in S2 manifests: {other}"), + } + } + } + + assert!(found_deleted_a, "file-A should have a DELETED entry in S2"); + assert!( + found_existing_b, + "file-B should have an EXISTING entry in S2" + ); + assert!(found_added_c, "file-C should have an ADDED entry in S2"); + } + + /// Resurrection guard: an already-DELETED entry in a rewritten manifest is + /// dropped, never carried forward as EXISTING. Without the `is_alive()` + /// check, S3's rewrite of the manifest holding file-A's DELETED entry + /// would resurrect file-A — for deletion vectors this yields multiple + /// live DVs per data file and readers fail with "Can't index multiple DVs". + #[tokio::test] + async fn test_row_delta_rewrite_does_not_resurrect_deleted_entries() { + let base_table = make_v2_minimal_table(); + + // S1: append file-A and file-B (one manifest holds both). + let file_a = make_data_file(&base_table, "test/a.parquet", 100); + let file_b = make_data_file(&base_table, "test/b.parquet", 200); + let mut c1 = Arc::new( + Transaction::new(&base_table) + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]), + ) + .commit(&base_table) + .await + .unwrap(); + let snap1 = if let TableUpdate::AddSnapshot { snapshot } = + c1.take_updates().into_iter().next().unwrap() + { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + let table_s1 = table_with_snapshot(&base_table, snap1).await; + + // S2: CoW-rewrite file-A into file-A2 — the rewritten manifest now + // records file-A as DELETED and file-B as EXISTING. + let file_a2 = make_data_file(&table_s1, "test/a2.parquet", 100); + let mut c2 = Arc::new( + Transaction::new(&table_s1) + .row_delta() + .remove_data_files(vec![file_a.clone()]) + .add_data_files(vec![file_a2]), + ) + .commit(&table_s1) + .await + .unwrap(); + let snap2 = if let TableUpdate::AddSnapshot { snapshot } = + c2.take_updates().into_iter().next().unwrap() + { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + let table_s2 = table_with_snapshot(&table_s1, snap2).await; + + // S3: CoW-rewrite file-B — this rewrites the manifest that still + // carries file-A's DELETED entry. file-A must NOT come back as EXISTING. + let file_b2 = make_data_file(&table_s2, "test/b2.parquet", 200); + let mut c3 = Arc::new( + Transaction::new(&table_s2) + .row_delta() + .remove_data_files(vec![file_b.clone()]) + .add_data_files(vec![file_b2]), + ) + .commit(&table_s2) + .await + .unwrap(); + let updates3 = c3.take_updates(); + let snap3 = if let TableUpdate::AddSnapshot { ref snapshot } = updates3[0] { + snapshot + } else { + panic!("expected AddSnapshot"); + }; + + let manifest_list = table_s2 + .manifest_list_reader(&std::sync::Arc::new(snap3.clone())) + .load() + .await + .unwrap(); + for manifest_file in manifest_list.entries() { + let manifest = manifest_file + .load_manifest(table_s2.file_io()) + .await + .unwrap(); + for entry in manifest.entries() { + if entry.data_file().file_path() == "test/a.parquet" { + assert!( + !entry.is_alive(), + "file-A was superseded in S2; the S3 rewrite must not resurrect it" + ); + } + } + } + } +} diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index b4f1946a7c..f5824e5baf 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -60,10 +60,6 @@ use crate::{Error, ErrorKind, TableRequirement, TableUpdate}; /// 3. **Delete Entry Processing**: The `delete_entries()` method is intended for future delete /// operations to specify which manifest entries should be marked as deleted. pub(crate) trait SnapshotProduceOperation: Send + Sync { - /// Returns the operation type that will be recorded in the snapshot summary. - /// - /// This determines what kind of operation is being performed (e.g., `Append`, `Overwrite`), - /// which is stored in the snapshot metadata for tracking and auditing purposes. fn operation(&self) -> Operation; /// Returns manifest entries that should be marked as deleted in the new snapshot. @@ -73,18 +69,29 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { snapshot_produce: &SnapshotProducer, ) -> impl Future>> + Send; - /// Returns existing manifest files that should be included in the new snapshot. - /// - /// This method determines which manifest files from the current snapshot should be - /// carried forward to the new snapshot. The selection depends on the operation type: + /// Returns existing manifest files to carry forward (or rewrite) into the new snapshot. /// - /// - **Append operations**: Typically include all existing manifests - /// - **Overwrite operations**: May exclude manifests for partitions being overwritten - /// - **Delete operations**: May exclude manifests for partitions being deleted + /// Implementations that need to delete specific files within a manifest should rewrite that + /// manifest (DELETED + EXISTING entries) and return the rewritten `ManifestFile` here. + /// `&mut SnapshotProducer` is provided so that implementations can call + /// `snapshot_produce.new_manifest_writer()` to produce the rewritten manifest. fn existing_manifest( &self, - snapshot_produce: &SnapshotProducer<'_>, + snapshot_produce: &mut SnapshotProducer<'_>, ) -> impl Future>> + Send; + + /// Data files being removed in this operation (used for snapshot summary metrics). + fn removed_data_files(&self) -> &[DataFile] { + &[] + } + + /// Whether this Overwrite replaces the entire table content. When true, + /// `truncate_table_summary` sets `deleted-data-files` to the previous total. + /// Row-level operations (RowDelta) return false; full-table rewrites (future + /// OverwriteFiles / ReplacePartitions) return true. + fn is_truncate_full_table(&self) -> bool { + false + } } pub(crate) struct DefaultManifestProcess; @@ -113,6 +120,9 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + // Added MoR delete files (position/equality deletes, incl. V3 deletion vectors). + // Written into a separate content=Deletes manifest by `write_added_delete_manifest`. + added_delete_files: Vec, // A counter used to generate unique manifest file names. // It starts from 0 and increments for each new manifest file. // Note: This counter is limited to the range of (0..u64::MAX). @@ -132,6 +142,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, + added_delete_files: vec![], manifest_counter: (0..), } } @@ -237,7 +248,10 @@ impl<'a> SnapshotProducer<'a> { snapshot_id } - fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result { + pub(crate) fn new_manifest_writer( + &mut self, + content: ManifestContentType, + ) -> Result { let new_manifest_path = format!( "{}/{}-m{}.{}", self.table.metadata().metadata_location()?, @@ -340,6 +354,65 @@ impl<'a> SnapshotProducer<'a> { writer.write_manifest_file().await } + /// Set the added MoR delete files to be written into a content=Deletes manifest. + pub(crate) fn set_added_delete_files(&mut self, delete_files: Vec) { + self.added_delete_files = delete_files; + } + + // Write a content=Deletes manifest for added MoR delete files (position/equality + // deletes, incl. V3 deletion vectors) and return the ManifestFile for the ManifestList. + async fn write_added_delete_manifest(&mut self) -> Result { + let added_delete_files = std::mem::take(&mut self.added_delete_files); + if added_delete_files.is_empty() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added delete files found when writing a delete manifest file", + )); + } + + let snapshot_id = self.snapshot_id; + let format_version = self.table.metadata().format_version(); + let manifest_entries = added_delete_files.into_iter().map(|delete_file| { + let builder = ManifestEntry::builder() + .status(crate::spec::ManifestStatus::Added) + .data_file(delete_file); + if format_version == FormatVersion::V1 { + builder.snapshot_id(snapshot_id).build() + } else { + builder.build() + } + }); + let mut writer = self.new_manifest_writer(ManifestContentType::Deletes)?; + for entry in manifest_entries { + writer.add_entry(entry)?; + } + writer.write_manifest_file().await + } + + // Write a data manifest containing DELETED-status entries and return the ManifestFile. + // Note: this is NOT an Iceberg "delete manifest" (content=Deletes for MoR delete files). + // It is a data manifest (content=Data) whose entries carry ManifestStatus::Deleted to + // record which data files were removed in Copy-on-Write mode. + async fn write_manifest_with_deleted_entries( + &mut self, + delete_entries: Vec, + ) -> Result { + if delete_entries.is_empty() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No delete entries found when writing a delete manifest file", + )); + } + + let mut writer = self.new_manifest_writer(ManifestContentType::Data)?; + for entry in delete_entries { + // Use add_delete_entry() to preserve Deleted status instead of add_entry() + // which always overwrites status to Added + writer.add_delete_entry(entry)?; + } + writer.write_manifest_file().await + } + /// Creates new manifests for data files added or removed, /// and collects all of the manifests to be included in the new snapshot as [ManifestFile] entries. async fn produce_manifests( @@ -347,15 +420,23 @@ impl<'a> SnapshotProducer<'a> { snapshot_produce_operation: &OP, manifest_process: &MP, ) -> Result> { + // Check if there's any content to add to the new snapshot + let delete_entries = snapshot_produce_operation.delete_entries(self).await?; + let has_delete_entries = !delete_entries.is_empty(); + // Assert current snapshot producer contains new content to add to new snapshot. // // TODO: Allowing snapshot property setup with no added data files is a workaround. // We should clean it up after all necessary actions are supported. // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 - if self.added_data_files.is_empty() && self.snapshot_properties.is_empty() { + if self.added_data_files.is_empty() + && self.added_delete_files.is_empty() + && self.snapshot_properties.is_empty() + && !has_delete_entries + { return Err(Error::new( ErrorKind::PreconditionFailed, - "No added data files or added snapshot properties found when write a manifest file", + "No added data files, delete entries, or snapshot properties found when write a manifest file", )); } @@ -368,8 +449,19 @@ impl<'a> SnapshotProducer<'a> { manifest_files.push(added_manifest); } - // # TODO - // Support process delete entries. + // Process added MoR delete files (content=Deletes manifest, e.g. V3 deletion vectors). + if !self.added_delete_files.is_empty() { + let added_delete_manifest = self.write_added_delete_manifest().await?; + manifest_files.push(added_delete_manifest); + } + + // Process delete entries. + if has_delete_entries { + let delete_manifest = self + .write_manifest_with_deleted_entries(delete_entries) + .await?; + manifest_files.push(delete_manifest); + } let manifest_files = manifest_process.process_manifests(self, manifest_files); Ok(manifest_files) @@ -406,6 +498,14 @@ impl<'a> SnapshotProducer<'a> { ); } + for data_file in snapshot_produce_operation.removed_data_files() { + summary_collector.remove_file( + data_file, + table_metadata.current_schema().clone(), + table_metadata.default_partition_spec().clone(), + ); + } + let previous_snapshot = table_metadata.current_snapshot(); // User-supplied snapshot properties are applied first, then the computed @@ -424,7 +524,7 @@ impl<'a> SnapshotProducer<'a> { update_snapshot_summaries( summary, previous_snapshot.map(|s| s.summary()), - snapshot_produce_operation.operation() == Operation::Overwrite, + snapshot_produce_operation.is_truncate_full_table(), ) } diff --git a/crates/iceberg/testdata/puffin/deletion-vector-v1-payload.bin b/crates/iceberg/testdata/puffin/deletion-vector-v1-payload.bin new file mode 100644 index 0000000000..80829fae22 Binary files /dev/null and b/crates/iceberg/testdata/puffin/deletion-vector-v1-payload.bin differ diff --git a/crates/iceberg/testdata/puffin/empty-position-index.bin b/crates/iceberg/testdata/puffin/empty-position-index.bin new file mode 100644 index 0000000000..8bbc1265dc Binary files /dev/null and b/crates/iceberg/testdata/puffin/empty-position-index.bin differ diff --git a/crates/iceberg/testdata/puffin/small-and-large-values-position-index.bin b/crates/iceberg/testdata/puffin/small-and-large-values-position-index.bin new file mode 100644 index 0000000000..989dabf6ad Binary files /dev/null and b/crates/iceberg/testdata/puffin/small-and-large-values-position-index.bin differ