diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index ad1af4389c..9e15bdefc8 100644
--- a/crates/iceberg/public-api.txt
+++ b/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
diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs
index a79312dfab..41a3d88e44 100644
--- a/crates/iceberg/src/spec/manifest/writer.rs
+++ b/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;
diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs
index fdb5a4a7b8..2e440086dc 100644
--- a/crates/iceberg/src/spec/partition.rs
+++ b/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![
diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs
index 2a7bd3f44a..a2128c0210 100644
--- a/crates/iceberg/src/spec/values/datum.rs
+++ b/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!()
             }
diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs
index e3e74338c9..b5f9089512 100644
--- a/crates/iceberg/src/transaction/append.rs
+++ b/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![],
         );
 
diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs
index efae9d01ba..22b840560f 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 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()
diff --git a/crates/iceberg/src/transaction/overwrite.rs b/crates/iceberg/src/transaction/overwrite.rs
new file mode 100644
index 0000000000..e032249546
--- /dev/null
+++ b/crates/iceberg/src/transaction/overwrite.rs
@@ -0,0 +1,957 @@
+// 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, invalid_data};
+use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation};
+use crate::table::Table;
+use crate::transaction::action::CommitStatus;
+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`.
+#[derive(Clone)]
+pub struct OverwriteAction {
+    check_duplicate: bool,
+    commit_uuid: Option<Uuid>,
+    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,
+            snapshot_properties: HashMap::default(),
+            added_data_files: vec![],
+            deleted_data_files: vec![],
+        }
+    }
+
+    /// Set whether to reject added paths already live in the table (enabled by default).
+    /// Paths deleted in the same overwrite may be re-added.
+    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.
+    ///
+    /// Files are matched by path against the current snapshot's live data files.
+    /// Their `record_count`, `file_size_in_bytes`, `partition_spec_id` and `partition`
+    /// must match the committed entry. Duplicate paths keep the first occurrence.
+    ///
+    /// # Errors
+    ///
+    /// The commit fails with [`crate::ErrorKind::DataInvalid`] for non-data files,
+    /// paths that are not live in the current snapshot, or mismatched metadata.
+    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 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 {
+    type State = ();
+
+    fn new_state(&self) -> Self::State {}
+
+    async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> {
+        let mut deleted_files = HashMap::new();
+        for file in &self.deleted_data_files {
+            if file.content_type() != crate::spec::DataContentType::Data {
+                return Err(invalid_data!("Only data content type is allowed"));
+            }
+            deleted_files.entry(file.file_path.as_str()).or_insert(file);
+        }
+        let snapshot_producer = SnapshotProducer::new(
+            table,
+            self.commit_uuid.unwrap_or_else(Uuid::now_v7),
+            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 operation = OverwriteOperation {
+            has_added_data_files: !self.added_data_files.is_empty(),
+            deleted_files,
+        };
+        snapshot_producer
+            .commit(operation, DefaultManifestProcess)
+            .await
+    }
+
+    async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {}
+}
+
+struct OverwriteOperation<'a> {
+    has_added_data_files: bool,
+    deleted_files: HashMap<&'a str, &'a DataFile>,
+}
+
+impl SnapshotProduceOperation for OverwriteOperation<'_> {
+    fn operation(&self) -> Operation {
+        match (self.has_added_data_files, !self.deleted_files.is_empty()) {
+            (true, true) => Operation::Overwrite,
+            (false, true) => Operation::Delete,
+            // Also a properties-only commit, the workaround from #1548.
+            _ => Operation::Append,
+        }
+    }
+
+    // Only the listed files are replaced, so the table totals carry over.
+    fn truncate_full_table(&self) -> bool {
+        false
+    }
+
+    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 table = snapshot_produce.table;
+        let mut manifests = vec![];
+        let mut matched: HashSet<&str> = HashSet::new();
+
+        if let Some(snapshot) = table.metadata().current_snapshot() {
+            let manifest_list = table.manifest_list_reader(snapshot).load().await?;
+            for manifest_file in manifest_list.entries() {
+                // Delete-only manifests record which files were removed and must survive
+                // until `expire_snapshots` cleans them up (see #2148).
+                if !manifest_file.has_added_files()
+                    && !manifest_file.has_existing_files()
+                    && !manifest_file.has_deleted_files()
+                {
+                    continue;
+                }
+                if manifest_file.content == crate::spec::ManifestContentType::Deletes
+                    || self.deleted_files.is_empty()
+                {
+                    manifests.push(manifest_file.clone());
+                    continue;
+                }
+
+                let manifest = table.manifest_reader().read(manifest_file).await?;
+                let mut deletes = vec![];
+                for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) {
+                    if let Some(requested) = self.deleted_files.get(entry.file_path()) {
+                        let actual = entry.data_file();
+                        let mismatch = if requested.record_count != actual.record_count {
+                            format!(
+                                "record_count: requested {}, table has {}",
+                                requested.record_count, actual.record_count
+                            )
+                        } else if requested.file_size_in_bytes != actual.file_size_in_bytes {
+                            format!(
+                                "file_size_in_bytes: requested {}, table has {}",
+                                requested.file_size_in_bytes, actual.file_size_in_bytes
+                            )
+                        } else if requested.partition_spec_id != actual.partition_spec_id {
+                            format!(
+                                "partition_spec_id: requested {}, table has {}",
+                                requested.partition_spec_id, actual.partition_spec_id
+                            )
+                        } else if requested.partition != actual.partition {
+                            format!(
+                                "partition: requested {:?}, table has {:?}",
+                                requested.partition, actual.partition
+                            )
+                        } else {
+                            deletes.push(requested.file_path.as_str());
+                            continue;
+                        };
+                        return Err(invalid_data!(
+                            "Delete file metadata does not match the table: {} ({mismatch})",
+                            entry.file_path()
+                        ));
+                    }
+                }
+                if deletes.is_empty() {
+                    manifests.push(manifest_file.clone());
+                    continue;
+                }
+                matched.extend(deletes);
+
+                let mut writer = snapshot_produce.new_manifest_writer(
+                    manifest_file.content,
+                    manifest.metadata().schema.clone(),
+                    Arc::new(manifest.metadata().partition_spec.clone()),
+                )?;
+                // Like Java, only live entries are carried into the rewrite.
+                for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) {
+                    if self.deleted_files.contains_key(entry.file_path()) {
+                        writer.add_delete_entry(entry.as_ref().clone())?;
+                    } else {
+                        writer.add_existing_entry(entry.as_ref().clone())?;
+                    }
+                }
+                manifests.push(writer.write_manifest_file().await?);
+            }
+        }
+
+        let mut missing: Vec<&str> = self
+            .deleted_files
+            .keys()
+            .copied()
+            .filter(|path| !matched.contains(path))
+            .collect();
+        if !missing.is_empty() {
+            missing.sort_unstable();
+            return Err(invalid_data!(
+                "Missing required files to delete: {}",
+                missing.join(", ")
+            ));
+        }
+
+        Ok(manifests)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use crate::memory::tests::new_memory_catalog;
+    use crate::spec::{
+        DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, ManifestListWriter,
+        ManifestStatus, ManifestWriterBuilder, NestedField, Operation, PrimitiveType, Schema,
+        SnapshotRef, Struct, Transform, Type, UnboundPartitionField, UnboundPartitionSpec,
+    };
+    use crate::table::Table;
+    use crate::test_utils::commit_with_fresh_state;
+    use crate::transaction::tests::make_v3_minimal_table_in_catalog;
+    use crate::transaction::{AddColumn, ApplyTransactionAction, Transaction};
+    use crate::{Catalog, ErrorKind, TableCommit, TableCreation, 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()
+    }
+
+    async fn current_manifest_entries(table: &Table) -> Vec<(ManifestStatus, String)> {
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap();
+
+        let mut entries = vec![];
+        for manifest_file in manifest_list.entries() {
+            let manifest = table.manifest_reader().read(manifest_file).await.unwrap();
+            for entry in manifest.entries() {
+                entries.push((entry.status(), entry.file_path().to_string()));
+            }
+        }
+        entries
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_with_deleted_files() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+
+        let original_file1 = test_data_file("test/original1.parquet", spec_id);
+        let original_file2 = test_data_file("test/original2.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx
+            .fast_append()
+            .add_data_files(vec![original_file1.clone(), original_file2.clone()]);
+        let tx = action.apply(tx).unwrap();
+        let table = tx.commit(&catalog).await.unwrap();
+
+        let replacement_file = test_data_file("test/replacement.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![replacement_file.clone()])
+            .delete_data_files(vec![original_file1.clone(), original_file2.clone()]);
+        let tx = action.apply(tx).unwrap();
+        let table = tx.commit(&catalog).await.unwrap();
+
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        assert_eq!(snapshot.summary().operation, Operation::Overwrite);
+        let summary = &snapshot.summary().additional_properties;
+        assert_eq!(summary.get("deleted-data-files").unwrap(), "2");
+        assert_eq!(summary.get("deleted-records").unwrap(), "2");
+
+        let entries = current_manifest_entries(&table).await;
+        assert_eq!(3, entries.len(), "{entries:?}");
+        for entry in [
+            (ManifestStatus::Deleted, "test/original1.parquet"),
+            (ManifestStatus::Deleted, "test/original2.parquet"),
+            (ManifestStatus::Added, "test/replacement.parquet"),
+        ] {
+            assert!(
+                entries.contains(&(entry.0, entry.1.to_string())),
+                "{entries:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_second_overwrite_does_not_resurrect_a_deleted_file() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+
+        let file_a = test_data_file("test/a.parquet", spec_id);
+        let file_b = test_data_file("test/b.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx
+            .fast_append()
+            .add_data_files(vec![file_a.clone(), file_b.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![test_data_file("test/c.parquet", spec_id)])
+            .delete_data_files(vec![file_a.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![test_data_file("test/d.parquet", spec_id)])
+            .delete_data_files(vec![file_b.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let entries = current_manifest_entries(&table).await;
+        assert!(
+            !entries.iter().any(
+                |(status, path)| path == "test/a.parquet" && *status != ManifestStatus::Deleted
+            ),
+            "the first overwrite's delete must not come back, entries: {entries:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_rewritten_manifest_keeps_the_schema_and_spec_it_was_written_with() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let original_spec_id = table.metadata().default_partition_spec_id();
+        let original_schema_id = table.metadata().current_schema_id();
+
+        let file_a = test_data_file("test/a.parquet", original_spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx.fast_append().add_data_files(vec![file_a.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx.update_schema().add_column(AddColumn::optional(
+            "extra",
+            Type::Primitive(PrimitiveType::Int),
+        ));
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        // There is no transaction action for partition evolution yet, so the spec is replaced
+        // on the metadata directly.
+        let evolved = table
+            .metadata()
+            .clone()
+            .into_builder(None)
+            .add_default_partition_spec(
+                UnboundPartitionSpec::builder()
+                    .add_partition_field(
+                        UnboundPartitionField::builder()
+                            .source_ids(vec![2])
+                            .name("y")
+                            .transform(Transform::Identity)
+                            .build()
+                            .unwrap(),
+                    )
+                    .unwrap()
+                    .build(),
+            )
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+        let table = table.with_metadata(Arc::new(evolved));
+        assert_ne!(original_schema_id, table.metadata().current_schema_id());
+        assert_ne!(
+            original_spec_id,
+            table.metadata().default_partition_spec_id()
+        );
+
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![test_data_file(
+                "test/b.parquet",
+                table.metadata().default_partition_spec_id(),
+            )])
+            .delete_data_files(vec![file_a.clone()]);
+        let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap();
+        let updates = action_commit.take_updates();
+        let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
+            SnapshotRef::new(snapshot.clone())
+        } else {
+            unreachable!()
+        };
+
+        let manifest_list = table
+            .manifest_list_reader(&new_snapshot)
+            .load()
+            .await
+            .unwrap();
+        let mut rewritten = None;
+        for manifest_file in manifest_list.entries() {
+            let manifest = table.manifest_reader().read(manifest_file).await.unwrap();
+            if manifest
+                .entries()
+                .iter()
+                .any(|entry| entry.status() == ManifestStatus::Deleted)
+            {
+                rewritten = Some((manifest_file.clone(), manifest));
+            }
+        }
+        let (rewritten_file, rewritten_manifest) =
+            rewritten.expect("the manifest holding test/a.parquet must have been rewritten");
+
+        assert_eq!(original_schema_id, rewritten_manifest.metadata().schema_id);
+        assert_eq!(original_spec_id, rewritten_file.partition_spec_id);
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_counts_a_duplicated_delete_path_once() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+
+        let file_a = test_data_file("test/a.parquet", spec_id);
+        let file_b = test_data_file("test/b.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx
+            .fast_append()
+            .add_data_files(vec![file_a.clone(), file_b.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![test_data_file("test/c.parquet", spec_id)])
+            .delete_data_files(vec![file_a.clone(), file_a.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let summary = table.metadata().current_snapshot().unwrap().summary();
+        assert_eq!(
+            summary.additional_properties.get("deleted-data-files"),
+            Some(&"1".to_string())
+        );
+        assert_eq!(
+            summary.additional_properties.get("deleted-records"),
+            Some(&"1".to_string())
+        );
+        assert_eq!(
+            summary.additional_properties.get("total-data-files"),
+            Some(&"2".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn test_delete_only_overwrite_reports_a_delete_operation() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+
+        let file_a = test_data_file("test/a.parquet", spec_id);
+        let file_b = test_data_file("test/b.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx
+            .fast_append()
+            .add_data_files(vec![file_a.clone(), file_b.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx.overwrite().delete_data_files(vec![file_a.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        assert_eq!(snapshot.summary().operation, Operation::Delete);
+        assert_eq!(
+            snapshot
+                .summary()
+                .additional_properties
+                .get("deleted-data-files"),
+            Some(&"1".to_string())
+        );
+
+        let entries = current_manifest_entries(&table).await;
+        assert!(
+            entries.contains(&(ManifestStatus::Deleted, "test/a.parquet".to_string())),
+            "the deleted file must be a tombstone, entries: {entries:?}"
+        );
+        assert!(
+            entries.contains(&(ManifestStatus::Existing, "test/b.parquet".to_string())),
+            "the untouched file must stay live, entries: {entries:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_rejects_deleting_a_file_the_table_does_not_have() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+
+        let file_a = test_data_file("test/a.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let action = tx.fast_append().add_data_files(vec![file_a.clone()]);
+        let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap();
+
+        let tx = Transaction::new(&table);
+        let action = tx
+            .overwrite()
+            .add_data_files(vec![test_data_file("test/b.parquet", spec_id)])
+            .delete_data_files(vec![
+                file_a.clone(),
+                test_data_file("test/never-committed.parquet", spec_id),
+            ]);
+        let err = commit_with_fresh_state(action, &table).await.err().unwrap();
+        assert_eq!(err.kind(), ErrorKind::DataInvalid);
+        assert!(
+            err.to_string()
+                .contains("Missing required files to delete: test/never-committed.parquet"),
+            "{err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_deletes_files_after_partition_source_promotion() {
+        let catalog = new_memory_catalog().await;
+        let base = make_v3_minimal_table_in_catalog(&catalog).await;
+        for transform in [Transform::Identity, Transform::Truncate(10)] {
+            let schema = Schema::builder()
+                .with_fields([NestedField::required(1, "x", PrimitiveType::Int.into()).into()])
+                .build()
+                .unwrap();
+            let spec = UnboundPartitionSpec::builder()
+                .add_partition_field(
+                    UnboundPartitionField::builder()
+                        .source_ids(vec![1])
+                        .name("p")
+                        .transform(transform)
+                        .build()
+                        .unwrap(),
+                )
+                .unwrap()
+                .build();
+            let table = catalog
+                .create_table(
+                    base.identifier().namespace(),
+                    TableCreation::builder()
+                        .name(format!("promoted-{transform}"))
+                        .schema(schema)
+                        .partition_spec(spec)
+                        .format_version(crate::spec::FormatVersion::V3)
+                        .build(),
+                )
+                .await
+                .unwrap();
+            let file = DataFile {
+                partition: Struct::from_iter([Some(Literal::int(300))]),
+                ..test_data_file(
+                    "test/a.parquet",
+                    table.metadata().default_partition_spec_id(),
+                )
+            };
+            let tx = Transaction::new(&table);
+            let table = tx
+                .fast_append()
+                .add_data_files([file.clone()])
+                .apply(tx)
+                .unwrap()
+                .commit(&catalog)
+                .await
+                .unwrap();
+            let schema = Schema::builder()
+                .with_fields([NestedField::required(1, "x", PrimitiveType::Long.into()).into()])
+                .build()
+                .unwrap();
+            let table = catalog
+                .update_table(
+                    TableCommit::builder()
+                        .ident(table.identifier().clone())
+                        .requirements(vec![])
+                        .updates(vec![
+                            TableUpdate::AddSchema { schema },
+                            TableUpdate::SetCurrentSchema { schema_id: -1 },
+                        ])
+                        .build(),
+                )
+                .await
+                .unwrap();
+            let tx = Transaction::new(&table);
+            let table = tx
+                .overwrite()
+                .delete_data_files([file])
+                .apply(tx)
+                .unwrap()
+                .commit(&catalog)
+                .await
+                .unwrap();
+            let snapshot = table.metadata().current_snapshot().unwrap();
+            assert_eq!(snapshot.summary().operation, Operation::Delete);
+            assert_eq!(
+                snapshot.summary().additional_properties["deleted-records"],
+                "1"
+            );
+            assert_eq!(current_manifest_entries(&table).await, [(
+                ManifestStatus::Deleted,
+                "test/a.parquet".to_string()
+            )]);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_readds_a_deleted_path_by_default() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let file = test_data_file(
+            "test/a.parquet",
+            table.metadata().default_partition_spec_id(),
+        );
+        let tx = Transaction::new(&table);
+        let table = tx
+            .fast_append()
+            .add_data_files([file.clone()])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        let tx = Transaction::new(&table);
+        let table = tx
+            .overwrite()
+            .delete_data_files([file.clone()])
+            .add_data_files([file])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        assert_eq!(current_manifest_entries(&table).await, [
+            (ManifestStatus::Deleted, "test/a.parquet".to_string()),
+            (ManifestStatus::Added, "test/a.parquet".to_string()),
+        ]);
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_rejects_incorrect_delete_metrics() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+        let file = test_data_file("test/a.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let table = tx
+            .fast_append()
+            .add_data_files(vec![file.clone()])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        for (incorrect, mismatch) in [
+            (
+                DataFile {
+                    record_count: 99,
+                    ..file.clone()
+                },
+                "record_count: requested 99, table has 1",
+            ),
+            (
+                DataFile {
+                    file_size_in_bytes: 999,
+                    ..file.clone()
+                },
+                "file_size_in_bytes: requested 999, table has 100",
+            ),
+            (
+                DataFile {
+                    partition: Struct::from_iter([Some(Literal::long(999))]),
+                    ..file.clone()
+                },
+                "partition: requested Struct { fields: [Some(Primitive(Long(999)))] }, table has Struct { fields: [Some(Primitive(Long(300)))] }",
+            ),
+            (
+                DataFile {
+                    partition_spec_id: 99,
+                    ..file.clone()
+                },
+                "partition_spec_id: requested 99, table has 0",
+            ),
+            (
+                DataFile {
+                    partition: Struct::from_iter([Some(Literal::string("invalid"))]),
+                    ..file.clone()
+                },
+                "partition: requested",
+            ),
+        ] {
+            let action = Transaction::new(&table)
+                .overwrite()
+                .delete_data_files(vec![incorrect]);
+            let err = commit_with_fresh_state(action, &table).await.err().unwrap();
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(err.message().contains(mismatch));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_rejects_non_data_deletions() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+        for content in [
+            DataContentType::PositionDeletes,
+            DataContentType::EqualityDeletes,
+        ] {
+            let file = DataFile {
+                content,
+                ..test_data_file("test/deletes.parquet", spec_id)
+            };
+            let action = Transaction::new(&table)
+                .overwrite()
+                .delete_data_files(vec![file]);
+            let err = commit_with_fresh_state(action, &table).await.err().unwrap();
+            assert_eq!(err.kind(), ErrorKind::DataInvalid);
+            assert!(err.message().contains("Only data content type is allowed"));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_preserves_v3_row_lineage() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+        let a = test_data_file("test/a.parquet", spec_id);
+        let b = test_data_file("test/b.parquet", spec_id);
+        let c = test_data_file("test/c.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let table = tx
+            .fast_append()
+            .add_data_files(vec![a, b.clone(), c])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        let manifests = table.manifest_list_reader(snapshot).load().await.unwrap();
+        let manifest = table
+            .manifest_reader()
+            .read(&manifests.entries()[0])
+            .await
+            .unwrap();
+        let original_ids: Vec<_> = manifest
+            .entries()
+            .iter()
+            .map(|e| (e.file_path().to_string(), e.data_file().first_row_id))
+            .collect();
+        assert!(original_ids.iter().all(|(_, id)| id.is_some()));
+        let next_row_id = table.metadata().next_row_id();
+        let replacement = test_data_file("test/d.parquet", spec_id);
+        let tx = Transaction::new(&table);
+        let table = tx
+            .overwrite()
+            .delete_data_files(vec![b])
+            .add_data_files(vec![replacement])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        let manifests = table
+            .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+            .load()
+            .await
+            .unwrap();
+        let mut actual_ids = vec![];
+        for file in manifests.entries() {
+            let manifest = table.manifest_reader().read(file).await.unwrap();
+            actual_ids.extend(
+                manifest
+                    .entries()
+                    .iter()
+                    .filter(|e| e.is_alive())
+                    .map(|e| (e.file_path().to_string(), e.data_file().first_row_id)),
+            );
+        }
+        for (path, id) in original_ids.iter().filter(|(p, _)| p != "test/b.parquet") {
+            assert!(actual_ids.contains(&(path.clone(), *id)));
+        }
+        // The rewritten manifest reserves IDs for its existing rows; their explicit IDs stay unchanged.
+        assert!(actual_ids.contains(&(
+            "test/d.parquet".to_string(),
+            Some((next_row_id + 2).try_into().unwrap())
+        )));
+        assert_eq!(table.metadata().next_row_id(), next_row_id + 3);
+    }
+
+    #[tokio::test]
+    async fn test_overwrite_preserves_delete_manifests_and_rejects_their_paths() {
+        let catalog = new_memory_catalog().await;
+        let table = make_v3_minimal_table_in_catalog(&catalog).await;
+        let spec_id = table.metadata().default_partition_spec_id();
+        let tx = Transaction::new(&table);
+        let table = tx
+            .fast_append()
+            .add_data_files(vec![test_data_file("test/a.parquet", spec_id)])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        let file = test_data_file("test/deletes.parquet", spec_id);
+        let output = table
+            .file_io()
+            .new_output("test/delete-manifest.avro")
+            .unwrap();
+        let mut writer = ManifestWriterBuilder::new(
+            output,
+            Some(snapshot.snapshot_id()),
+            table.metadata().current_schema().clone(),
+            table.metadata().default_partition_spec().as_ref().clone(),
+        )
+        .build_v3_deletes();
+        writer
+            .add_file(
+                DataFile {
+                    content: DataContentType::PositionDeletes,
+                    ..file.clone()
+                },
+                snapshot.sequence_number(),
+            )
+            .unwrap();
+        let delete_manifest = writer.write_manifest_file().await.unwrap();
+        let mut manifests: Vec<_> = table
+            .manifest_list_reader(snapshot)
+            .load()
+            .await
+            .unwrap()
+            .consume_entries()
+            .into_iter()
+            .collect();
+        manifests.push(delete_manifest);
+        let mut writer = ManifestListWriter::v3(
+            table
+                .file_io()
+                .new_output(snapshot.manifest_list())
+                .unwrap()
+                .writer()
+                .await
+                .unwrap(),
+            snapshot.snapshot_id(),
+            snapshot.parent_snapshot_id(),
+            snapshot.sequence_number(),
+            Some(table.metadata().next_row_id()),
+        );
+        writer.add_manifests(manifests.into_iter()).unwrap();
+        writer.close().await.unwrap();
+        let action = Transaction::new(&table)
+            .overwrite()
+            .delete_data_files(vec![file]);
+        let err = commit_with_fresh_state(action, &table).await.err().unwrap();
+        assert_eq!(err.kind(), ErrorKind::DataInvalid);
+        assert!(
+            err.message()
+                .contains("Missing required files to delete: test/deletes.parquet")
+        );
+
+        let original = table
+            .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+            .load()
+            .await
+            .unwrap()
+            .entries()
+            .iter()
+            .find(|m| m.manifest_path == "test/delete-manifest.avro")
+            .unwrap()
+            .clone();
+        let tx = Transaction::new(&table);
+        let table = tx
+            .overwrite()
+            .delete_data_files([test_data_file("test/a.parquet", spec_id)])
+            .apply(tx)
+            .unwrap()
+            .commit(&catalog)
+            .await
+            .unwrap();
+        let manifests = table
+            .manifest_list_reader(table.metadata().current_snapshot().unwrap())
+            .load()
+            .await
+            .unwrap();
+        let preserved = manifests
+            .entries()
+            .iter()
+            .find(|m| m.manifest_path == original.manifest_path)
+            .unwrap();
+        assert_eq!(preserved, &original);
+    }
+}
diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index c301750012..73277bda66 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -139,18 +139,29 @@ impl<'a> SnapshotProducer<'a> {
             snapshot_id: Self::generate_unique_snapshot_id(table),
             commit_uuid,
             snapshot_properties,
-            added_data_files,
-            deleted_data_files,
+            added_data_files: Self::dedupe_data_files(added_data_files),
+            deleted_data_files: Self::dedupe_data_files(deleted_data_files),
             manifest_counter: AtomicU64::new(0),
         }
     }
 
+    fn dedupe_data_files(files: Vec<DataFile>) -> Vec<DataFile> {
+        let mut seen = HashSet::with_capacity(files.len());
+        let keep: Vec<bool> = files
+            .iter()
+            .map(|f| seen.insert(f.file_path.as_str()))
+            .collect();
+        files
+            .into_iter()
+            .zip(keep)
+            .filter_map(|(file, keep)| keep.then_some(file))
+            .collect()
+    }
+
     pub(crate) fn validate_added_data_files(&self) -> Result<()> {
         for data_file in &self.added_data_files {
             if data_file.content_type() != crate::spec::DataContentType::Data {
-                return Err(invalid_data!(
-                    "Only data content type is allowed for fast append"
-                ));
+                return Err(invalid_data!("Only data content type is allowed"));
             }
             // Check if the data file partition spec id matches the table default partition spec id.
             if self.table.metadata().default_partition_spec_id() != data_file.partition_spec_id {
@@ -172,11 +183,14 @@ impl<'a> SnapshotProducer<'a> {
             return Ok(());
         };
 
-        let new_files: HashSet<&str> = self
+        let mut new_files: HashSet<&str> = self
             .added_data_files
             .iter()
             .map(|df| df.file_path.as_str())
             .collect();
+        for df in &self.deleted_data_files {
+            new_files.remove(df.file_path.as_str());
+        }
 
         let runtime = self.table.runtime();
         let manifest_list = self
@@ -350,7 +364,7 @@ impl<'a> SnapshotProducer<'a> {
         &mut self,
         snapshot_produce_operation: &OP,
         manifest_process: &MP,
-    ) -> Result<Vec<ManifestFile>> {
+    ) -> Result<(Vec<ManifestFile>, Summary)> {
         // 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.
@@ -367,6 +381,9 @@ impl<'a> SnapshotProducer<'a> {
         }
 
         let existing_manifests = snapshot_produce_operation.existing_manifest(self).await?;
+        let summary = self.summary(snapshot_produce_operation).map_err(|err| {
+            Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err)
+        })?;
         let mut manifest_files = existing_manifests;
 
         // Process added entries.
@@ -391,7 +408,7 @@ impl<'a> SnapshotProducer<'a> {
         );
 
         let manifest_files = manifest_process.process_manifests(self, manifest_files);
-        Ok(manifest_files)
+        Ok((manifest_files, summary))
     }
 
     // Returns a `Summary` of the current snapshot
@@ -514,14 +531,7 @@ impl<'a> SnapshotProducer<'a> {
             ),
         };
 
-        // Calling self.summary() before self.produce_manifests() is important because self.added_data_files
-        // will be set to an empty vec after self.produce_manifests() returns, resulting in an empty summary
-        // being generated.
-        let summary = self.summary(&snapshot_produce_operation).map_err(|err| {
-            Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err)
-        })?;
-
-        let new_manifests = self
+        let (new_manifests, summary) = self
             .produce_manifests(&snapshot_produce_operation, &process)
             .await?;