feat(scan): apply V3 deletion vectors on read (stacked on #2678)

.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"]

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"

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 }

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<dyn std::error::Error>> {
+    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(())
+}

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 }

crates/iceberg/public-api.txt

@@ -169,6 +169,31 @@ impl serde_core::ser::Serialize for iceberg::compression::CompressionCodec
 pub fn iceberg::compression::CompressionCodec::serialize<S: serde_core::ser::Serializer>(&self, serializer: S) -> core::result::Result<<S as serde_core::ser::Serializer>::Ok, <S as serde_core::ser::Serializer>::Error>
 impl<'de> serde_core::de::Deserialize<'de> for iceberg::compression::CompressionCodec
 pub fn iceberg::compression::CompressionCodec::deserialize<D: serde_core::de::Deserializer<'de>>(deserializer: D) -> core::result::Result<Self, <D as serde_core::de::Deserializer>::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<Self>
+pub fn iceberg::delete_vector::DeleteVector::from_serialized_bytes(data: &[u8]) -> iceberg::Result<Self>
+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<usize>
+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<alloc::string::String, alloc::string::String>) -> iceberg::Result<iceberg::puffin::Blob>
+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<iceberg::spec::DataFile>
+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<Self::Item>
 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<iceberg::puffin::Blob>
 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<iceberg::puffin::BlobMetadata>
+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<iceberg::puffin::PuffinWriteResult>
 pub async fn iceberg::puffin::PuffinWriter::new(output_file: &iceberg::io::OutputFile, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, compress_footer: bool) -> iceberg::Result<Self>
 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<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub struct iceberg::scan::FileScanTaskDeleteFile
+pub iceberg::scan::FileScanTaskDeleteFile::content_offset: core::option::Option<i64>
+pub iceberg::scan::FileScanTaskDeleteFile::content_size_in_bytes: core::option::Option<i64>
 pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<alloc::vec::Vec<i32>>
+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<alloc::string::String>
 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