feat(transaction): add overwrite action (continues #2185)

crates/iceberg/public-api.txt

@@ -3341,6 +3341,15 @@ pub fn iceberg::transaction::FastAppendAction::set_snapshot_properties(self, sna
 pub fn iceberg::transaction::FastAppendAction::with_check_duplicate(self, v: bool) -> Self
 impl core::clone::Clone for iceberg::transaction::FastAppendAction
 pub fn iceberg::transaction::FastAppendAction::clone(&self) -> iceberg::transaction::FastAppendAction
+pub struct iceberg::transaction::OverwriteAction
+impl iceberg::transaction::OverwriteAction
+pub fn iceberg::transaction::OverwriteAction::add_data_files(self, data_files: impl core::iter::traits::collect::IntoIterator<Item = iceberg::spec::DataFile>) -> Self
+pub fn iceberg::transaction::OverwriteAction::delete_data_files(self, data_files: impl core::iter::traits::collect::IntoIterator<Item = iceberg::spec::DataFile>) -> Self
+pub fn iceberg::transaction::OverwriteAction::set_commit_uuid(self, commit_uuid: uuid::Uuid) -> Self
+pub fn iceberg::transaction::OverwriteAction::set_snapshot_properties(self, snapshot_properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self
+pub fn iceberg::transaction::OverwriteAction::with_check_duplicate(self, v: bool) -> Self
+impl core::clone::Clone for iceberg::transaction::OverwriteAction
+pub fn iceberg::transaction::OverwriteAction::clone(&self) -> iceberg::transaction::OverwriteAction
 pub struct iceberg::transaction::ReplaceSortOrderAction
 impl iceberg::transaction::ReplaceSortOrderAction
 pub fn iceberg::transaction::ReplaceSortOrderAction::asc(self, name: &str, null_order: iceberg::spec::NullOrder) -> Self
@@ -3355,6 +3364,7 @@ pub async fn iceberg::transaction::Transaction::commit(self, catalog: &dyn icebe
 pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::transaction::ExpireSnapshotsAction
 pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::FastAppendAction
 pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self
+pub fn iceberg::transaction::Transaction::overwrite(&self) -> iceberg::transaction::OverwriteAction
 pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::ReplaceSortOrderAction
 pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::UpdateLocationAction
 pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::UpdateSchemaAction

crates/iceberg/src/spec/manifest/writer.rs

@@ -315,10 +315,6 @@ impl ManifestWriter {
     /// Add a delete manifest entry. This method will update following status of the entry:
     /// - Update the entry status to `Deleted`
     /// - Set the snapshot id to the current snapshot id
-    ///
-    /// # TODO
-    /// Remove this allow later
-    #[allow(dead_code)]
     pub(crate) fn add_delete_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
         self.check_data_file(&entry.data_file)?;
         entry.status = ManifestStatus::Deleted;
@@ -350,10 +346,6 @@ impl ManifestWriter {
 
     /// Add an existing manifest entry. This method will update following status of the entry:
     /// - Update the entry status to `Existing`
-    ///
-    /// # TODO
-    /// Remove this allow later
-    #[allow(dead_code)]
     pub(crate) fn add_existing_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
         self.check_data_file(&entry.data_file)?;
         entry.status = ManifestStatus::Existing;

crates/iceberg/src/spec/partition.rs

@@ -1934,6 +1934,21 @@ mod tests {
                 Literal::int(10),
                 "p=10",
             ),
+            (
+                PrimitiveType::Decimal {
+                    precision: 9,
+                    scale: 2,
+                },
+                Transform::Identity,
+                Literal::decimal(123),
+                "p=123",
+            ),
+            (
+                PrimitiveType::Uuid,
+                Transform::Identity,
+                Literal::uuid_from_str("f79c3e09-677c-4bbd-a479-3f349cb785e7").unwrap(),
+                "p=f79c3e09-677c-4bbd-a479-3f349cb785e7",
+            ),
         ] {
             let schema = Schema::builder()
                 .with_fields(vec![

crates/iceberg/src/spec/values/datum.rs

@@ -294,7 +294,10 @@ impl Display for Datum {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match (&self.r#type, &self.literal) {
             (_, PrimitiveLiteral::Boolean(val)) => write!(f, "{val}"),
-            (PrimitiveType::Int | PrimitiveType::Unknown, PrimitiveLiteral::Int(val)) => {
+            (
+                PrimitiveType::Int | PrimitiveType::Long | PrimitiveType::Unknown,
+                PrimitiveLiteral::Int(val),
+            ) => {
                 write!(f, "{val}")
             }
             (PrimitiveType::Long | PrimitiveType::Unknown, PrimitiveLiteral::Long(val)) => {
@@ -321,7 +324,7 @@ impl Display for Datum {
                 write!(f, "{}", timestamptz::nanoseconds_to_datetimetz(*val))
             }
             (_, PrimitiveLiteral::String(val)) => write!(f, r#""{val}""#),
-            (PrimitiveType::Uuid, PrimitiveLiteral::UInt128(val)) => {
+            (PrimitiveType::Uuid | PrimitiveType::Unknown, PrimitiveLiteral::UInt128(val)) => {
                 write!(f, "{}", uuid::Uuid::from_u128(*val))
             }
             (_, PrimitiveLiteral::Binary(val)) => display_bytes(val, f),
@@ -334,6 +337,8 @@ impl Display for Datum {
             ) => {
                 write!(f, "{}", decimal_from_i128_with_scale(*val, *scale))
             }
+            // The scale lives on the dropped source type, so only the unscaled value is left.
+            (PrimitiveType::Unknown, PrimitiveLiteral::Int128(val)) => write!(f, "{val}"),
             (_, _) => {
                 unreachable!()
             }

crates/iceberg/src/transaction/append.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 
 use async_trait::async_trait;
 use uuid::Uuid;
@@ -72,18 +72,6 @@ impl FastAppendAction {
         self.snapshot_properties = snapshot_properties;
         self
     }
-
-    /// Collapse files sharing a path to their first occurrence, so a single
-    /// manifest never references the same file twice. Always runs (unlike the
-    /// `check_duplicate`-gated cross-snapshot check) since it is in-memory only.
-    fn dedupe_added_files(&self) -> Vec<DataFile> {
-        let mut seen = HashSet::with_capacity(self.added_data_files.len());
-        self.added_data_files
-            .iter()
-            .filter(|data_file| seen.insert(data_file.file_path.as_str()))
-            .cloned()
-            .collect()
-    }
 }
 
 #[async_trait]
@@ -99,7 +87,7 @@ impl TransactionAction for FastAppendAction {
             table,
             self.commit_uuid.unwrap_or_else(Uuid::now_v7),
             self.snapshot_properties.clone(),
-            self.dedupe_added_files(),
+            self.added_data_files.clone(),
             vec![],
         );
 

crates/iceberg/src/transaction/mod.rs

@@ -55,6 +55,7 @@ mod action;
 pub use action::*;
 mod append;
 mod expire_snapshots;
+mod overwrite;
 mod snapshot;
 mod sort_order;
 mod update_location;
@@ -75,6 +76,7 @@ use crate::table::Table;
 use crate::transaction::action::TransactionActionEntry;
 pub use crate::transaction::append::FastAppendAction;
 pub use crate::transaction::expire_snapshots::ExpireSnapshotsAction;
+pub use crate::transaction::overwrite::OverwriteAction;
 pub use crate::transaction::sort_order::ReplaceSortOrderAction;
 pub use crate::transaction::update_location::UpdateLocationAction;
 pub use crate::transaction::update_properties::UpdatePropertiesAction;
@@ -150,6 +152,11 @@ impl Transaction {
         FastAppendAction::new()
     }
 
+    /// Creates an overwrite action.
+    pub fn overwrite(&self) -> OverwriteAction {
+        OverwriteAction::new()
+    }
+
     /// Creates replace sort order action.
     pub fn replace_sort_order(&self) -> ReplaceSortOrderAction {
         ReplaceSortOrderAction::new()