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 dd33e5c2af..da733b6104 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 f14a94af93..3ecbc831f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,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.0.0" datafusion-cli = "54.0.0" diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml index e043c195ef..8fed46e42c 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 9353a31842..a0af5dbd6a 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 c2013294e3..4643c11e0c 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -169,6 +169,31 @@ 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::from_serialized_bytes(data: &[u8]) -> 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 @@ -1229,10 +1254,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 @@ -1273,11 +1306,15 @@ pub fn iceberg::scan::FileScanTask::serialize<__S>(&self, __serializer: __S) -> impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask pub fn iceberg::scan::FileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::scan::FileScanTaskDeleteFile +pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option +pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option> +pub iceberg::scan::FileScanTaskDeleteFile::file_format: iceberg::spec::DataFileFormat pub iceberg::scan::FileScanTaskDeleteFile::file_path: alloc::string::String pub iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes: u64 pub iceberg::scan::FileScanTaskDeleteFile::file_type: iceberg::spec::DataContentType pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32 +pub iceberg::scan::FileScanTaskDeleteFile::referenced_data_file: core::option::Option impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile impl core::cmp::PartialEq for iceberg::scan::FileScanTaskDeleteFile @@ -1286,7 +1323,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskDeleteFile impl iceberg::scan::FileScanTaskDeleteFile -pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), ())> +pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), (), (), (), ())> impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskDeleteFile pub fn iceberg::scan::FileScanTaskDeleteFile::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskDeleteFile @@ -3101,6 +3138,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/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index de31d678a5..7fefb43b5b 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -30,11 +30,12 @@ use crate::delete_vector::DeleteVector; use crate::expr::Predicate::AlwaysTrue; use crate::expr::{Predicate, Reference}; use crate::io::FileIO; +use crate::puffin::PuffinReader; use crate::runtime::Runtime; use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile}; use crate::spec::{ - DataContentType, Datum, ListType, MapType, NestedField, NestedFieldRef, PartnerAccessor, - PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, + DataContentType, DataFileFormat, Datum, ListType, MapType, NestedField, NestedFieldRef, + PartnerAccessor, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor, StructType, Type, visit_schema_with_partner, }; use crate::{Error, ErrorKind, Result}; @@ -51,13 +52,21 @@ pub(crate) struct CachingDeleteFileLoader { // Intermediate context during processing of a delete file task. enum DeleteFileContext { - // TODO: Delete Vector loader from Puffin files ExistingEqDel, ExistingPosDel, PosDels { file_path: String, stream: ArrowRecordBatchStream, }, + /// A V3 deletion vector; the blob located by `content_offset`/`content_size` + /// is read and parsed into a `DeleteVector` in the parse phase. + DelVec { + file_path: String, + referenced_data_file: String, + content_offset: u64, + content_size: u64, + file_io: FileIO, + }, FreshEqDel { batch_stream: ArrowRecordBatchStream, equality_ids: HashSet, @@ -117,11 +126,11 @@ impl CachingDeleteFileLoader { /// tasks from starting to load the same equality delete file. We spawn a task to load /// the EQ delete's record batch stream, convert it to a predicate, update the delete filter, /// and notify any task that was waiting for it. - /// * When this gets updated to add support for delete vectors, the load phase will return - /// a PuffinReader for them. + /// * for delete vectors (V3, stored in Puffin) the load phase records the Puffin file + /// path + referenced data file; the parse phase reads the deletion-vector-v1 blob. /// * The parse phase parses each record batch stream according to its associated data type. /// The result of this is a map of data file paths to delete vectors for the positional - /// delete tasks (and in future for the delete vector tasks). For equality delete + /// delete tasks and the deletion-vector tasks. For equality delete /// file tasks, this results in an unbound Predicate. /// * The unbound Predicates resulting from equality deletes are sent to their associated oneshot /// channel to store them in the right place in the delete file managers state. @@ -143,7 +152,7 @@ impl CachingDeleteFileLoader { /// | /// | /// +-----------------------------+--------------------------+ - /// Pos Del Del Vec (Not yet Implemented) EQ Del + /// Pos Del Del Vec EQ Del /// | | | /// [parse pos del stream] [parse del vec puffin] [parse eq del] /// HashMap HashMap (Predicate, Sender) @@ -248,12 +257,48 @@ impl CachingDeleteFileLoader { notify.notified().await; Ok(DeleteFileContext::ExistingPosDel) } - PosDelLoadAction::Load => Ok(DeleteFileContext::PosDels { - file_path: task.file_path.clone(), - stream: basic_delete_file_loader - .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes) - .await?, - }), + PosDelLoadAction::Load => { + if task.file_format == DataFileFormat::Puffin { + // V3 deletion vector — the blob located by + // content_offset/content_size is read + parsed in the + // parse phase (not a Parquet positional-delete stream). + Ok(DeleteFileContext::DelVec { + file_path: task.file_path.clone(), + referenced_data_file: task + .referenced_data_file + .clone() + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "deletion vector is missing referenced_data_file", + ) + })?, + content_offset: task.content_offset.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "deletion vector is missing content_offset", + ) + })? as u64, + content_size: task.content_size_in_bytes.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "deletion vector is missing content_size_in_bytes", + ) + })? as u64, + file_io: basic_delete_file_loader.file_io().clone(), + }) + } else { + Ok(DeleteFileContext::PosDels { + file_path: task.file_path.clone(), + stream: basic_delete_file_loader + .parquet_to_batch_stream( + &task.file_path, + task.file_size_in_bytes, + ) + .await?, + }) + } + } } } @@ -304,6 +349,20 @@ impl CachingDeleteFileLoader { results: del_vecs, }) } + DeleteFileContext::DelVec { + file_path, + referenced_data_file, + content_offset, + content_size, + file_io, + } => { + let delete_vector = + Self::read_deletion_vector(&file_io, &file_path, content_offset, content_size) + .await?; + let mut results = HashMap::default(); + results.insert(referenced_data_file, delete_vector); + Ok(ParsedDeleteFileContext::DelVecs { file_path, results }) + } DeleteFileContext::FreshEqDel { sender, batch_stream, @@ -326,6 +385,74 @@ impl CachingDeleteFileLoader { } } + /// Reads a V3 deletion-vector-v1 blob located at `content_offset` (length + /// `content_size`) within `file_path` and parses it into a `DeleteVector`. + /// + /// The blob is addressed by the manifest entry's content offset/size rather + /// than by parsing a Puffin footer, so this works whether the deletion vector + /// is a standalone blob file (e.g. written by DuckDB) or stored inside a + /// Puffin container. + async fn read_deletion_vector( + file_io: &FileIO, + file_path: &str, + content_offset: u64, + content_size: u64, + ) -> Result { + let bytes = file_io.new_input(file_path)?.read().await?; + let start = content_offset as usize; + let end = start + .checked_add(content_size as usize) + .filter(|end| *end <= bytes.len()) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector blob range {content_offset}..+{content_size} is out of bounds for {file_path} ({} bytes)", + bytes.len() + ), + ) + })?; + + // Fast path: per the Iceberg spec a `deletion-vector-v1` blob is stored + // uncompressed ("Omit compression-codec; deletion-vector-v1 is not + // compressed"), so the bytes located by content_offset/content_size are the + // serialized blob directly. This also covers container-less blob files + // (e.g. written by DuckDB) that have no Puffin footer. + if let Ok(delete_vector) = DeleteVector::from_serialized_bytes(&bytes[start..end]) { + return Ok(delete_vector); + } + + // Fallback: a deletion vector blob that is compressed (which the spec + // forbids, but a non-conforming writer could produce) can only be + // decompressed using the codec recorded in the Puffin footer. If this file + // is a valid Puffin file, read the blob through its footer, which honors + // the per-blob compression codec. + let reader = PuffinReader::new(file_io.new_input(file_path)?); + let metadata = reader.file_metadata().await.map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!( + "deletion vector at offset {content_offset} in {file_path} is neither a valid uncompressed deletion-vector-v1 blob nor a Puffin container" + ), + ) + .with_source(err) + })?; + let blob_metadata = metadata + .blobs() + .iter() + .find(|blob| blob.offset() == content_offset) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Puffin file {file_path} has no blob at content_offset {content_offset}" + ), + ) + })?; + let blob = reader.blob(blob_metadata).await?; + DeleteVector::from_puffin_blob(blob) + } + /// Parses a record batch stream coming from positional delete files /// /// Returns a map of data file path to a delete vector @@ -1125,6 +1252,100 @@ mod tests { ); } + /// Ported from apache/iceberg-go `TestFilterByDeletionVector` / + /// `TestReadAllDeletionVectors`: a V3 deletion vector (Puffin) that references a + /// data file is loaded during scan and its marked positions become the data + /// file's delete vector. Before DV-applied scan support, the loader tried to read + /// the Puffin file as a Parquet positional-delete and the marked rows were not + /// excluded. + #[tokio::test] + async fn test_load_deletes_applies_v3_deletion_vector() { + use crate::scan::FileScanTask; + use crate::spec::{DataFileFormat, Schema, Struct}; + + let tmp_dir = TempDir::new().unwrap(); + let table_location = tmp_dir.path(); + let file_io = FileIO::new_with_fs(); + + let data_file_schema = Arc::new( + Schema::builder() + .with_fields(vec![ + crate::spec::NestedField::optional( + 2, + "y", + crate::spec::Type::Primitive(crate::spec::PrimitiveType::Long), + ) + .into(), + ]) + .build() + .unwrap(), + ); + + let data_file_path = format!("{}/data-dv.parquet", table_location.to_str().unwrap()); + + // Build a deletion vector marking positions 1 and 3 as deleted and write it to + // a Puffin file via the DV-write path. + let mut dv = DeleteVector::default(); + dv.insert(1); + dv.insert(3); + let dv_puffin_path = format!("{}/dv-1.puffin", table_location.to_str().unwrap()); + let dv_data_file = dv + .write_to_puffin_file( + &file_io, + dv_puffin_path.clone(), + data_file_path.clone(), + Struct::empty(), + 0, + ) + .await + .unwrap(); + + let dv_del = FileScanTaskDeleteFile { + file_path: dv_data_file.file_path().to_string(), + file_size_in_bytes: dv_data_file.file_size_in_bytes(), + file_type: DataContentType::PositionDeletes, + partition_spec_id: 0, + equality_ids: None, + content_offset: dv_data_file.content_offset(), + content_size_in_bytes: dv_data_file.content_size_in_bytes(), + file_format: DataFileFormat::Puffin, + referenced_data_file: dv_data_file.referenced_data_file(), + }; + + let file_scan_task = FileScanTask { + file_size_in_bytes: 0, + start: 0, + length: 0, + record_count: None, + data_file_path: data_file_path.clone(), + data_file_format: DataFileFormat::Parquet, + schema: data_file_schema.clone(), + project_field_ids: vec![2], + predicate: None, + deletes: vec![dv_del], + partition: None, + partition_spec: None, + name_mapping: None, + case_sensitive: false, + }; + + let delete_file_loader = + CachingDeleteFileLoader::new(file_io.clone(), 10, Runtime::current()); + let delete_filter = delete_file_loader + .load_deletes(&file_scan_task.deletes, file_scan_task.schema_ref()) + .await + .unwrap() + .unwrap(); + + // The DV's marked positions are surfaced as the data file's delete vector. + let dv_handle = delete_filter + .get_delete_vector(&file_scan_task) + .expect("expected a deletion vector for the data file"); + let mut positions: Vec = dv_handle.lock().unwrap().iter().collect(); + positions.sort(); + assert_eq!(positions, vec![1, 3]); + } + #[tokio::test] async fn test_large_equality_delete_batch_stack_overflow() { let tmp_dir = TempDir::new().unwrap(); diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs index a0ad5b627c..5d58425e09 100644 --- a/crates/iceberg/src/arrow/reader/row_filter.rs +++ b/crates/iceberg/src/arrow/reader/row_filter.rs @@ -1240,6 +1240,10 @@ mod tests { partition_spec_id: 0, equality_ids: None, file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(), + content_offset: None, + content_size_in_bytes: None, + file_format: DataFileFormat::Parquet, + referenced_data_file: None, }], partition: None, partition_spec: None, diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index e58bb671b6..4cc79b634c 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -25,7 +25,7 @@ use tokio::sync::Notify; use crate::runtime::Runtime; use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile}; -use crate::spec::{DataContentType, DataFile, Struct}; +use crate::spec::{DataContentType, DataFile, DataFileFormat, Struct}; /// Index of delete files #[derive(Debug, Clone)] @@ -46,8 +46,10 @@ struct PopulatedDeleteFileIndex { pos_deletes_by_partition: HashMap>>, // TODO: do we need this? // pos_deletes_by_path: HashMap>>, - - // TODO: Deletion Vector support + /// V3 deletion vectors indexed by the data file path they apply to (the + /// manifest entry's `referenced_data_file`). The spec allows at most one + /// deletion vector per data file. + dvs_by_data_file_path: HashMap>, } impl DeleteFileIndex { @@ -128,10 +130,29 @@ impl PopulatedDeleteFileIndex { HashMap::default(); let mut global_equality_deletes: Vec> = vec![]; + let mut dvs_by_data_file_path: HashMap> = HashMap::default(); files.into_iter().for_each(|ctx| { let arc_ctx = Arc::new(ctx); + // V3 deletion vectors are PositionDeletes-content entries stored as Puffin + // files carrying a `referenced_data_file`. Index them by that path (the spec + // allows at most one DV per data file) instead of by partition. + let dv_referenced = { + let df = arc_ctx.manifest_entry.data_file(); + if df.content_type() == DataContentType::PositionDeletes + && df.file_format() == DataFileFormat::Puffin + { + df.referenced_data_file() + } else { + None + } + }; + if let Some(referenced) = dv_referenced { + dvs_by_data_file_path.insert(referenced, arc_ctx); + return; + } + let partition = arc_ctx.manifest_entry.data_file().partition(); // The spec states that "Equality delete files stored with an unpartitioned spec are applied as global deletes". @@ -161,6 +182,7 @@ impl PopulatedDeleteFileIndex { global_equality_deletes, eq_deletes_by_partition, pos_deletes_by_partition, + dvs_by_data_file_path, } } @@ -195,11 +217,21 @@ impl PopulatedDeleteFileIndex { .for_each(|delete| results.push(delete.as_ref().into())); } - // TODO: the spec states that: - // "The data file's file_path is equal to the delete file's referenced_data_file if it is non-null". - // we're not yet doing that here. The referenced data file's name will also be present in the positional - // delete file's file path column. - if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { + // V3 deletion vectors are matched to a specific data file via + // `referenced_data_file` and, per the spec, supersede positional delete files + // for that data file ("when a DV applies, positional-delete files must NOT be + // applied"). A DV applies when its sequence number is >= the data file's. + if let Some(dv) = self + .dvs_by_data_file_path + .get(data_file.file_path()) + .filter(|dv| { + seq_num + .map(|seq_num| dv.manifest_entry.sequence_number() >= Some(seq_num)) + .unwrap_or(true) + }) + { + results.push(dv.as_ref().into()); + } else if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { deletes .iter() // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` diff --git a/crates/iceberg/src/delete_vector.rs b/crates/iceberg/src/delete_vector.rs index df8a10193c..6c18b7abe9 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,219 @@ 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()), + )); + } + Self::from_serialized_bytes(blob.data()) + } + + /// Parse a serialized `deletion-vector-v1` blob payload: + /// `[u32 BE length][magic 0xD1D33964][serialized RoaringTreemap][u32 BE CRC32]`. + /// + /// These are exactly the bytes located by a manifest entry's `content_offset` + /// and `content_size_in_bytes`, so this works whether the deletion vector is a + /// standalone blob file (e.g. written by DuckDB) or stored inside a Puffin + /// container (the manifest offset points past the container framing). + pub fn from_serialized_bytes(data: &[u8]) -> Result { + 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 +309,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 +347,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 +433,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/scan/task.rs b/crates/iceberg/src/scan/task.rs index f3b556bcbf..843486df83 100644 --- a/crates/iceberg/src/scan/task.rs +++ b/crates/iceberg/src/scan/task.rs @@ -161,6 +161,10 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile { .with_file_type(ctx.manifest_entry.content_type()) .with_partition_spec_id(ctx.partition_spec_id) .with_equality_ids(ctx.manifest_entry.data_file.equality_ids.clone()) + .with_file_format(ctx.manifest_entry.data_file().file_format()) + .with_referenced_data_file(ctx.manifest_entry.data_file().referenced_data_file()) + .with_content_offset(ctx.manifest_entry.data_file().content_offset()) + .with_content_size_in_bytes(ctx.manifest_entry.data_file().content_size_in_bytes()) .build() } } @@ -184,4 +188,24 @@ pub struct FileScanTaskDeleteFile { /// equality ids for equality deletes (null for anything other than equality-deletes) #[builder(default)] pub equality_ids: Option>, + + /// Format of the delete file. `Puffin` marks a V3 deletion vector + /// (as opposed to a Parquet positional/equality delete file). + #[builder(default = DataFileFormat::Parquet)] + pub file_format: DataFileFormat, + + /// The data file a V3 deletion vector applies to (the manifest entry's + /// `referenced_data_file`). `None` for positional/equality delete files. + #[builder(default)] + pub referenced_data_file: Option, + + /// Offset of the deletion-vector blob within the file (the manifest entry's + /// `content_offset`). `None` for positional/equality delete files. + #[builder(default)] + pub content_offset: Option, + + /// Size in bytes of the deletion-vector blob (the manifest entry's + /// `content_size_in_bytes`). `None` for positional/equality delete files. + #[builder(default)] + pub content_size_in_bytes: Option, } diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 9d94e18ede..e4f56a8cde 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -113,7 +113,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 d78f41cd42..e0637778df 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..386db64917 --- /dev/null +++ b/crates/iceberg/src/transaction/row_delta.rs @@ -0,0 +1,987 @@ +// 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" + ); + } + } + } + } +} + +/// End-to-end coverage for the COMBINED RowDelta commit: a deletion vector +/// against an existing data file plus a new data-file append, in ONE +/// transaction. This is the SCD2/upsert merge shape — the demote of a +/// superseded row and the insert of its replacement must be crash-atomic, +/// which a single snapshot gives by construction. Uses a real V3 table +/// (memory catalog), real Parquet data files, and a real Puffin DV, and reads +/// the result back through the DV-applying scan. +#[cfg(test)] +mod e2e_tests { + use std::collections::HashMap; + use std::sync::Arc; + + use arrow_array::{Int32Array, RecordBatch}; + use arrow_schema::{DataType, Field, Schema as ArrowSchema}; + use futures::TryStreamExt; + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use parquet::file::properties::WriterProperties; + use roaring::RoaringTreemap; + use tempfile::TempDir; + + use crate::catalog::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; + use crate::delete_vector::DeleteVector; + use crate::spec::{ + DataContentType, DataFile, DataFileFormat, FormatVersion, Literal, ManifestList, + NestedField, Operation, PrimitiveType, Schema, Struct, Type, + }; + use crate::table::Table; + use crate::transaction::{ApplyTransactionAction, Transaction}; + use crate::writer::base_writer::data_file_writer::DataFileWriterBuilder; + use crate::writer::file_writer::ParquetWriterBuilder; + use crate::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, + }; + use crate::writer::file_writer::rolling_writer::RollingFileWriterBuilder; + use crate::writer::{IcebergWriter, IcebergWriterBuilder}; + use crate::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation, TableIdent}; + + fn arrow_schema() -> Arc { + Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + Field::new("ver", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])), + ])) + } + + fn batch(ids: Vec, vers: Vec) -> RecordBatch { + RecordBatch::try_new(arrow_schema(), vec![ + Arc::new(Int32Array::from(ids)), + Arc::new(Int32Array::from(vers)), + ]) + .unwrap() + } + + /// Write `batch` to a single new data file. `prefix` must be unique per + /// write: `DefaultFileNameGenerator`'s counter resets per instance, so a + /// shared prefix would collide two writes onto one path. + async fn write_data_file(table: &Table, prefix: &str, batch: RecordBatch) -> Vec { + let schema = table.metadata().current_schema().clone(); + let rolling = RollingFileWriterBuilder::new_with_default_file_size( + ParquetWriterBuilder::new(WriterProperties::builder().build(), schema), + table.file_io().clone(), + DefaultLocationGenerator::new(table.metadata()).unwrap(), + DefaultFileNameGenerator::new(prefix.to_string(), None, DataFileFormat::Parquet), + ); + let mut writer = DataFileWriterBuilder::new(rolling) + .build(None) + .await + .unwrap(); + writer.write(batch).await.unwrap(); + writer.close().await.unwrap() + } + + /// All live (id, ver) rows via a scan — deletion vectors applied. + async fn live_rows(table: &Table) -> Vec<(i32, i32)> { + let mut stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let mut rows = Vec::new(); + while let Some(b) = stream.try_next().await.unwrap() { + let ids = b + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let vers = b + .column_by_name("ver") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for i in 0..b.num_rows() { + rows.push((ids.value(i), vers.value(i))); + } + } + rows.sort_unstable(); + rows + } + + /// (file_path, content_type, data_sequence_number) for every ALIVE + /// manifest entry in the current snapshot. + async fn live_entries(table: &Table) -> Vec<(String, DataContentType, i64)> { + let snap = table.metadata().current_snapshot().unwrap(); + let bytes = table + .file_io() + .new_input(snap.manifest_list()) + .unwrap() + .read() + .await + .unwrap(); + let ml = + ManifestList::parse_with_version(&bytes, table.metadata().format_version()).unwrap(); + let mut out = Vec::new(); + for mf in ml.entries() { + let m = mf.load_manifest(table.file_io()).await.unwrap(); + for e in m.entries() { + if e.is_alive() { + out.push(( + e.data_file().file_path().to_string(), + e.data_file().content_type(), + e.sequence_number().expect("sequence number inherited"), + )); + } + } + } + out + } + + /// Seed a V3 (id, ver) table with one data file (1..4, v1). Returns the + /// catalog, ident, and the seed file's path. + async fn seed_table(warehouse: &TempDir) -> (impl Catalog, TableIdent, String) { + let catalog = MemoryCatalogBuilder::default() + .load( + "memory", + HashMap::from([( + MEMORY_CATALOG_WAREHOUSE.to_string(), + warehouse.path().to_str().unwrap().to_string(), + )]), + ) + .await + .unwrap(); + let ns = NamespaceIdent::new("db".to_string()); + catalog.create_namespace(&ns, HashMap::new()).await.unwrap(); + + let schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "ver", Type::Primitive(PrimitiveType::Int)).into(), + ]) + .build() + .unwrap(); + let ident = TableIdent::new(ns.clone(), "t".to_string()); + let table = catalog + .create_table( + &ns, + TableCreation::builder() + .name("t".to_string()) + .schema(schema) + .format_version(FormatVersion::V3) + .build(), + ) + .await + .unwrap(); + + let data_files = + write_data_file(&table, "seed", batch(vec![1, 2, 3, 4], vec![1, 1, 1, 1])).await; + let file_a = data_files[0].file_path().to_string(); + let tx = Transaction::new(&table); + let table = tx + .fast_append() + .add_data_files(data_files) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + assert_eq!(live_rows(&table).await.len(), 4); + + (catalog, ident, file_a) + } + + /// One RowDelta commit carries a DV demoting a row in the seed file AND a + /// new data file with its replacement: exactly one snapshot, correct end + /// state, and the same-sequence DV binds only to its referenced file (the + /// appended file's position 0 holds the replacement row, which survives). + #[tokio::test] + async fn row_delta_demote_and_append_commit_as_single_snapshot() { + let warehouse = TempDir::new().unwrap(); + let (catalog, ident, file_a) = seed_table(&warehouse).await; + let table = catalog.load_table(&ident).await.unwrap(); + let snapshots_before = table.metadata().snapshots().count(); + + // DV on the seed file dropping position 1 — the old (2, v1). + let mut positions = RoaringTreemap::new(); + positions.insert(1); + let dv_path = format!("{}/dv-merge-0.puffin", warehouse.path().to_str().unwrap()); + let dv_file = DeleteVector::new(positions) + .write_to_puffin_file( + table.file_io(), + dv_path, + file_a.clone(), + Struct::from_iter(Vec::>::new()), + 0, + ) + .await + .unwrap(); + + // New file: replacement (2, v2) at POSITION 0 + a fresh insert (5, v1). + let new_files = write_data_file(&table, "merged", batch(vec![2, 5], vec![2, 1])).await; + + let tx = Transaction::new(&table); + let table = tx + .row_delta() + .add_data_files(new_files) + .add_delete_files(vec![dv_file]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + + assert_eq!( + table.metadata().snapshots().count(), + snapshots_before + 1, + "demote + append must land in a single snapshot" + ); + assert_eq!( + table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .operation, + Operation::Overwrite, + "RowDelta with data + deletes commits an Overwrite" + ); + assert_eq!(live_rows(&table).await, vec![ + (1, 1), + (2, 2), + (3, 1), + (4, 1), + (5, 1) + ]); + } + + /// Sequence-number semantics of the combined commit: the seed file keeps + /// its original sequence; the appended file and the DV share the new + /// snapshot's sequence. + #[tokio::test] + async fn row_delta_combined_commit_sequence_numbers() { + let warehouse = TempDir::new().unwrap(); + let (catalog, ident, file_a) = seed_table(&warehouse).await; + let table = catalog.load_table(&ident).await.unwrap(); + + let mut positions = RoaringTreemap::new(); + positions.insert(1); + let dv_path = format!("{}/dv-merge-1.puffin", warehouse.path().to_str().unwrap()); + let dv_file = DeleteVector::new(positions) + .write_to_puffin_file( + table.file_io(), + dv_path, + file_a.clone(), + Struct::from_iter(Vec::>::new()), + 0, + ) + .await + .unwrap(); + let new_files = write_data_file(&table, "merged", batch(vec![2, 5], vec![2, 1])).await; + let file_b = new_files[0].file_path().to_string(); + + let tx = Transaction::new(&table); + let table = tx + .row_delta() + .add_data_files(new_files) + .add_delete_files(vec![dv_file]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + + let entries = live_entries(&table).await; + let seq_of = |path: &str, content: DataContentType| -> i64 { + entries + .iter() + .find(|(p, c, _)| p == path && *c == content) + .unwrap_or_else(|| panic!("no alive entry for {path} ({content:?})")) + .2 + }; + + assert_eq!(seq_of(&file_a, DataContentType::Data), 1); + assert_eq!(seq_of(&file_b, DataContentType::Data), 2); + let dv_seq = entries + .iter() + .find(|(_, c, _)| *c == DataContentType::PositionDeletes) + .expect("DV entry alive") + .2; + assert_eq!(dv_seq, 2); + assert_eq!(entries.len(), 3, "seed + appended + DV, nothing else"); + } +} diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index d200c5ba9c..0fea899d9f 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -62,10 +62,6 @@ const META_ROOT_PATH: &str = "metadata"; /// 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. @@ -75,18 +71,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; @@ -115,6 +122,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). @@ -134,6 +144,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, + added_delete_files: vec![], manifest_counter: (0..), } } @@ -239,7 +250,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().location(), @@ -334,6 +348,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( @@ -341,15 +414,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", )); } @@ -362,8 +443,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) @@ -400,6 +492,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 @@ -418,7 +518,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