laskoviymishka commented on code in PR #2185: URL: https://github.com/apache/iceberg-rust/pull/2185#discussion_r4014361884
########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", + table.metadata().location(), + Uuid::now_v7(), + ); + let output_file = table.file_io().new_output(&new_manifest_path)?; + let builder = ManifestWriterBuilder::new( + output_file, + Some(self.snapshot_id), + manifest_file.key_metadata.clone(), + table.metadata().current_schema().clone(), Review Comment: We're stamping the rewritten manifest with the table's current schema and default partition spec, but the entries were written under the original manifest's schema/spec. After any schema or partition evolution these differ, and the Avro header ends up with the wrong schema-id (505) and partition-spec-id (507). Readers trust those ids to interpret partition values, so pruning and pushdown silently run against the wrong mapping — this is also how Java/PyIceberg will misread the table. I'd pull both from the manifest itself: `manifest.metadata().schema.clone()` and `manifest.metadata().partition_spec.clone()` (or `partition_spec_by_id(manifest_file.partition_spec_id)`). Java does exactly this via `newManifestWriter(reader.spec())`. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) Review Comment: This filter drops any manifest whose only entries are `Deleted`-status. `FastAppendOperation.existing_manifest()` hit exactly this and added `|| entry.has_deleted_files()` (append.rs:142, referencing #2148) — dropping delete-only manifests lets the removed files reappear as live. So `overwrite().delete(A).add(B)` followed by `overwrite().add(C)` (no deletes) would drop the manifest that recorded A as Deleted, and A comes back. I'd mirror the append fix and add `|| manifest_file.has_deleted_files()` here — and note the same guard is repeated in the deletes-path loop just below (line 178). ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", + table.metadata().location(), + Uuid::now_v7(), + ); + let output_file = table.file_io().new_output(&new_manifest_path)?; + let builder = ManifestWriterBuilder::new( + output_file, + Some(self.snapshot_id), + manifest_file.key_metadata.clone(), + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().as_ref().clone(), + ); + + let mut writer = match table.metadata().format_version() { + FormatVersion::V1 => builder.build_v1(), + FormatVersion::V2 => match manifest_file.content { + ManifestContentType::Data => builder.build_v2_data(), + ManifestContentType::Deletes => builder.build_v2_deletes(), + }, + FormatVersion::V3 => match manifest_file.content { + ManifestContentType::Data => builder.build_v3_data(), + ManifestContentType::Deletes => builder.build_v3_deletes(), + }, + }; + + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + let mut deleted: ManifestEntry = (**entry).clone(); + deleted.snapshot_id = Some(self.snapshot_id); + writer.add_deleted_entry(deleted)?; + } else { + let cloned: ManifestEntry = (**entry).clone(); + writer.add_existing_entry(cloned)?; Review Comment: This else branch resurrects previously-deleted files. `add_existing_entry` forces `status = Existing` (writer.rs:380), and an entry that was already `Deleted` in a prior overwrite isn't a live match here, so it falls through and comes back as live data. Concretely: an overwrite deletes A / adds C, so M1 is rewritten with A=Deleted, B=Existing. A later overwrite that touches B rewrites that manifest again — A isn't in the new delete set, hits this branch, and reappears as Existing. Java's `filterManifestWithDeletedFiles` only ever re-adds `liveEntries()` (Added/Existing); already-Deleted entries are dropped. I'd guard on `entry.is_alive()` first — skip or re-add tombstones as Deleted, `add_deleted_entry` for paths in the delete set, `add_existing_entry` only for the rest. Re-adding tombstones needs the original `snapshot_id` preserved, which ties into the `add_deleted_entry` note below. wdyt? ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", + table.metadata().location(), + Uuid::now_v7(), + ); + let output_file = table.file_io().new_output(&new_manifest_path)?; Review Comment: `rewrite_manifest` always writes through a plain `output_file`, but `new_manifest_writer` in snapshot.rs branches on `table.encryption_manager()` and wraps it with `em.encrypt(...)` when present. For an encrypted table the rewritten manifest goes out in plaintext, and readers using the copied `key_metadata` to decrypt it will fail or read garbage. I'd mirror `new_manifest_writer` and route through the encryption manager when one is configured. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite Review Comment: A delete-only overwrite (deletes, no adds) still reports `Operation::Overwrite`. Java's `BaseOverwriteFiles` returns `DELETE` when there are only deletes, `APPEND` when only adds, and `OVERWRITE` when both — tools keying off the summary operation will read a delete-only snapshot as an overwrite. Minor, but I'd match on `(has_adds, has_deletes)`. ########## crates/iceberg/src/transaction/snapshot.rs: ########## @@ -348,7 +355,13 @@ impl<'a> SnapshotProducer<'a> { // 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() { + // + // A delete-only overwrite (no added files, no properties, but with deleted files) is + // valid per the Iceberg spec — the existing manifests are rewritten with deleted entries. + if self.added_data_files.is_empty() + && self.snapshot_properties.is_empty() + && self.deleted_data_files.is_empty() Review Comment: This branch is what enables a delete-only overwrite, but nothing tests that path — no overwrite with deletes and no adds. I'd add one, plus a case that deletes a path absent from every manifest (which is where the phantom-count issue above bites). Small thing while we're here: the error message just below still only mentions added files/properties — worth adding deleted files so it matches the guard. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", + table.metadata().location(), + Uuid::now_v7(), + ); + let output_file = table.file_io().new_output(&new_manifest_path)?; + let builder = ManifestWriterBuilder::new( + output_file, + Some(self.snapshot_id), + manifest_file.key_metadata.clone(), + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().as_ref().clone(), + ); + + let mut writer = match table.metadata().format_version() { + FormatVersion::V1 => builder.build_v1(), + FormatVersion::V2 => match manifest_file.content { + ManifestContentType::Data => builder.build_v2_data(), + ManifestContentType::Deletes => builder.build_v2_deletes(), + }, + FormatVersion::V3 => match manifest_file.content { + ManifestContentType::Data => builder.build_v3_data(), + ManifestContentType::Deletes => builder.build_v3_deletes(), + }, + }; + + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + let mut deleted: ManifestEntry = (**entry).clone(); + deleted.snapshot_id = Some(self.snapshot_id); + writer.add_deleted_entry(deleted)?; + } else { + let cloned: ManifestEntry = (**entry).clone(); + writer.add_existing_entry(cloned)?; + } + } + + writer.write_manifest_file().await + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use crate::spec::{ + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, + ManifestStatus, Operation, SnapshotRef, Struct, + }; + use crate::transaction::tests::make_v2_minimal_table; + use crate::transaction::{Transaction, TransactionAction}; + use crate::{TableRequirement, TableUpdate}; + + fn test_data_file(path: &str, partition_spec_id: i32) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition_spec_id(partition_spec_id) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_empty_data_overwrite_action() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx.overwrite().add_data_files(vec![]); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + #[tokio::test] + async fn test_overwrite_snapshot_properties() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let mut snapshot_properties = HashMap::new(); + snapshot_properties.insert("key".to_string(), "val".to_string()); + + let data_file = test_data_file( + "test/1.parquet", + table.metadata().default_partition_spec_id(), + ); + + let action = tx + .overwrite() + .set_snapshot_properties(snapshot_properties) + .add_data_files(vec![data_file]); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + snapshot + } else { + unreachable!() + }; + assert_eq!( + new_snapshot + .summary() + .additional_properties + .get("key") + .unwrap(), + "val" + ); + } + + #[tokio::test] + async fn test_overwrite_incompatible_partition_value() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("test/3.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::string("test"))])) + .build() + .unwrap(); + + let action = tx.overwrite().add_data_files(vec![data_file]); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + #[tokio::test] + async fn test_overwrite_basic() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let data_file = test_data_file( + "test/3.parquet", + table.metadata().default_partition_spec_id(), + ); + + let action = tx.overwrite().add_data_files(vec![data_file.clone()]); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + let requirements = action_commit.take_requirements(); + + assert!( + matches!((&updates[0],&updates[1]), (TableUpdate::AddSnapshot { snapshot },TableUpdate::SetSnapshotRef { reference,ref_name }) if snapshot.snapshot_id() == reference.snapshot_id && ref_name == MAIN_BRANCH) + ); + + assert_eq!( + vec![ + TableRequirement::UuidMatch { + uuid: table.metadata().uuid() + }, + TableRequirement::RefSnapshotIdMatch { + r#ref: MAIN_BRANCH.to_string(), + snapshot_id: table.metadata().current_snapshot_id + } + ], + requirements + ); + + let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + SnapshotRef::new(snapshot.clone()) + } else { + unreachable!() + }; + assert_eq!(new_snapshot.summary().operation, Operation::Overwrite); + + let manifest_list = table + .manifest_list_reader(&new_snapshot) + .load() + .await + .unwrap(); + assert_eq!(1, manifest_list.entries().len()); + assert_eq!( + manifest_list.entries()[0].sequence_number, + new_snapshot.sequence_number() + ); + + let manifest = manifest_list.entries()[0] + .load_manifest(table.file_io()) + .await + .unwrap(); + assert_eq!(1, manifest.entries().len()); + assert_eq!( + new_snapshot.sequence_number(), + manifest.entries()[0] + .sequence_number() + .expect("Inherit sequence number by load manifest") + ); + assert_eq!( + new_snapshot.snapshot_id(), + manifest.entries()[0].snapshot_id().unwrap() + ); + assert_eq!(data_file, *manifest.entries()[0].data_file()); + } + + #[tokio::test] + async fn test_overwrite_with_deleted_files() { Review Comment: This test overwrites a manifest whose entries are all freshly-appended (Added), so it never exercises rewriting a manifest that already holds a Deleted entry — which is exactly the resurrection path flagged above. As written it can't catch that bug. I'd add a second overwrite: after this one leaves original1/original2 as Deleted, run another overwrite that rewrites the same manifest and assert those two stay Deleted rather than flipping back to Existing. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", + table.metadata().location(), + Uuid::now_v7(), + ); + let output_file = table.file_io().new_output(&new_manifest_path)?; + let builder = ManifestWriterBuilder::new( + output_file, + Some(self.snapshot_id), + manifest_file.key_metadata.clone(), + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().as_ref().clone(), + ); + + let mut writer = match table.metadata().format_version() { + FormatVersion::V1 => builder.build_v1(), + FormatVersion::V2 => match manifest_file.content { + ManifestContentType::Data => builder.build_v2_data(), + ManifestContentType::Deletes => builder.build_v2_deletes(), + }, + FormatVersion::V3 => match manifest_file.content { + ManifestContentType::Data => builder.build_v3_data(), + ManifestContentType::Deletes => builder.build_v3_deletes(), + }, + }; + + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + let mut deleted: ManifestEntry = (**entry).clone(); + deleted.snapshot_id = Some(self.snapshot_id); + writer.add_deleted_entry(deleted)?; + } else { + let cloned: ManifestEntry = (**entry).clone(); + writer.add_existing_entry(cloned)?; + } + } + + writer.write_manifest_file().await + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use crate::spec::{ + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, + ManifestStatus, Operation, SnapshotRef, Struct, + }; + use crate::transaction::tests::make_v2_minimal_table; + use crate::transaction::{Transaction, TransactionAction}; + use crate::{TableRequirement, TableUpdate}; + + fn test_data_file(path: &str, partition_spec_id: i32) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition_spec_id(partition_spec_id) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_empty_data_overwrite_action() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx.overwrite().add_data_files(vec![]); + assert!(Arc::new(action).commit(&table).await.is_err()); Review Comment: `is_err()` alone passes on any error, so if this ever starts failing earlier for an unrelated reason the test still goes green. I'd match on `err.kind() == ErrorKind::PreconditionFailed` so it actually pins the precondition. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,528 @@ +// 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, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestWriterBuilder, Operation, +}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + deleted_data_files: vec![], + } + } + + /// Set whether to check duplicate files. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Specify data files to be removed from the table in this overwrite. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let deleted_file_paths: HashSet<String> = self + .deleted_data_files + .iter() + .map(|f| f.file_path.clone()) + .collect(); + + let snapshot_id = snapshot_producer.snapshot_id(); + snapshot_producer + .commit( + OverwriteOperation { + deleted_file_paths, + snapshot_id, + }, + DefaultManifestProcess, + ) + .await + } +} + +struct OverwriteOperation { + deleted_file_paths: HashSet<String>, + snapshot_id: i64, +} + +impl SnapshotProduceOperation for OverwriteOperation { + fn operation(&self) -> Operation { + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + 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?; + + if self.deleted_file_paths.is_empty() { + return Ok(manifest_list + .entries() + .iter() + .filter(|entry| entry.has_added_files() || entry.has_existing_files()) + .cloned() + .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 has_deletes = manifest.entries().iter().any(|entry| { + entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) + }); + + if has_deletes { + let rewritten = self + .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .await?; + result.push(rewritten); + } else { + result.push(manifest_file.clone()); + } + } + + Ok(result) + } +} + +impl OverwriteOperation { + /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` + /// as `ManifestStatus::Deleted`. + async fn rewrite_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + manifest_file: &ManifestFile, + manifest: &Manifest, + ) -> Result<ManifestFile> { + let table = snapshot_produce.table; + + let new_manifest_path = format!( + "{}/metadata/{}-m-overwrite.avro", Review Comment: This hardcodes `{location}/metadata/...` and a fresh `Uuid::now_v7()` per rewrite. Every other writer goes through `table.metadata().metadata_location()?`, which respects `write.metadata.path` — a custom metadata path lands these rewritten manifests in the wrong prefix. The naming is also off from the `{commit_uuid}-m{counter}` convention. Since the random UUID isn't tied to the commit, orphan cleanup can't find these on a failed commit and they leak. I'd use `metadata_location()?` and thread the commit uuid + counter through (a `commit_uuid()` getter alongside `snapshot_id()`, or move this into `SnapshotProducer`). ########## crates/iceberg/src/spec/manifest/writer.rs: ########## @@ -382,6 +382,14 @@ impl ManifestWriter { Ok(()) } + /// Add a deleted manifest entry, preserving the original sequence numbers. + pub(crate) fn add_deleted_entry(&mut self, mut entry: ManifestEntry) -> Result<()> { Review Comment: This sets `status` but leaves `snapshot_id` to the caller, and it sits right next to `add_delete_entry` (line 343) which stamps both `status` and `snapshot_id = self.snapshot_id`. The two differ only by tense, so it's easy to reach for the wrong one. A tombstone written with the wrong `snapshot_id` attributes the delete to the wrong snapshot, and `expireSnapshots` then either drops the data too early or never. Once the resurrection fix starts re-adding prior tombstones through here, this contract matters. I'd either stamp `snapshot_id` here too, or rename (`add_tombstone_entry`) and document that the caller owns `snapshot_id`. ########## crates/iceberg/src/transaction/snapshot.rs: ########## @@ -402,6 +415,14 @@ impl<'a> SnapshotProducer<'a> { ); } + for data_file in &self.deleted_data_files { Review Comment: Two things tangle up around the deletion counts here. `remove_file` runs for every user-supplied deleted file regardless of whether it was actually found live in a manifest, so double-deletes or never-committed paths inflate `deleted-data-files` / `deleted-records`. I'd count from what the rewrite pass actually marked Deleted, and error if a delete path matched nothing (the existence validation Java does). Separately, `produce_manifests` still calls `summary()` with `truncate_full_table = (operation == Overwrite)`, so for a partial overwrite the truncate path zeroes the totals and reports `deleted-data-files` as the previous total rather than the k you deleted. `test_overwrite_with_deleted_files` only passes because it deletes all files (k == N). I'd gate truncate behind an operation hook that `OverwriteOperation` returns false for — with one caveat: once truncate is off, `update_totals` needs saturating subtraction, otherwise the delete-heavy case @timsaucer hit underflows again (the June rebase only avoided it because truncate capped `removed` at the previous total). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
