vendor/iceberg-arrow59/src/transaction/overwrite.rs
// 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.
//! `OverwriteFilesAction` — the **copy-on-write** commit.
//!
//! Before this module the vendored crate had exactly one data-writing verb,
//! `fast_append`. There was no overwrite, no rewrite-files, no `RowDelta`, no
//! delete-file action — and upstream iceberg-rust main (e9c22bf, 2026-10-02) still
//! has no overwrite COMMIT action: #2752 added a copy-on-write *rewrite* primitive
//! that returns removed/added file lists "for future overwrite-style commit
//! actions", but nothing that commits them (#1104 RowDelta and #735 delete
//! support are still open).
//! But the generic machinery was already here and already named the operation:
//! [`SnapshotProducer::summary`] branches on
//! `snapshot_produce_operation.operation() == Operation::Overwrite`, and the
//! [`SnapshotProduceOperation`] doc says in as many words that an overwrite
//! "may exclude manifests for partitions being overwritten". `existing_manifest`
//! IS the seam where replaced files are dropped. This action uses it.
//!
//! # What it commits
//!
//! One snapshot with `operation = overwrite` that, atomically:
//!
//! - **removes** every data file whose path is in `removed_data_file_paths`, and
//! - **adds** every data file in `added_data_files`.
//!
//! That is copy-on-write: the caller has already rewritten the surviving rows of
//! the removed files into the added files. The table's *live* file set moves in
//! one catalog pointer swap; a reader on the old snapshot still sees the old
//! files, because nothing is physically deleted.
//!
//! # How a file is removed
//!
//! A manifest is immutable, so "removing" a data file means writing a NEW
//! manifest in its place. For each data manifest in the current snapshot's
//! manifest list:
//!
//! | live entries hit by `removed_data_file_paths` | what happens |
//! |---|---|
//! | none | the manifest file is carried forward **byte-identical** — no rewrite, no read amplification |
//! | some, but not all live entries | rewritten: survivors as `Existing` (original snapshot id + sequence numbers preserved), removed ones as `Deleted` |
//! | all live entries | dropped from the manifest list entirely |
//!
//! Dropping a wholly-dead manifest is what keeps a table merged N times from
//! accumulating N empty manifests. It is safe for files THIS action removes,
//! because the removal is expressed by rewriting (or dropping) the very manifest
//! that held the file — no other manifest still lists it as live.
//!
//! **Divergence from upstream, recorded 2026-10-02.** Upstream's fast append now
//! carries delete-only manifests forward (`has_deleted_files()`, apache/iceberg-rust
//! #2148), because upstream's rewrite path marks a removal with a delete-only
//! manifest that masks `Added` entries elsewhere — drop it and the files come back.
//! This action still skips data manifests with no live entries (unchanged
//! behaviour from the 0.9 fork). That is correct for tables whose removals were
//! all made by this action or by Java/Spark (which rewrite the holding manifest
//! too), but NOT if an upstream-style mask manifest is ever in the table. Revisit
//! before anything in skade commits through upstream's rewrite primitive.
//!
//! Delete manifests (`ManifestContentType::Deletes`, i.e. the equality-delete
//! files `skade::Table::delete_equality` writes) are **carried forward
//! untouched**. This action replaces DATA files; it is the copy-on-write half,
//! and it does not know how to reason about somebody else's merge-on-read
//! tombstones.
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, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile,
ManifestWriter, ManifestWriterBuilder, Operation,
};
use crate::table::Table;
use crate::transaction::snapshot::{
DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer,
};
use crate::transaction::{ActionCommit, TransactionAction};
use crate::{Error, ErrorKind};
/// Snapshot summary property carrying the number of data files this overwrite
/// removed. Also the reason an overwrite that adds nothing still has a
/// non-empty summary — [`SnapshotProducer::manifest_file`] refuses a snapshot
/// with neither added data files nor added snapshot properties.
pub const REMOVED_DATA_FILES_PROP: &str = "removed-data-files";
/// A transaction action that commits an Iceberg **overwrite**: a set of data
/// files replaced by another set, in one snapshot.
///
/// See the [module docs](self) for the manifest rewriting rules.
pub struct OverwriteFilesAction {
check_duplicate: bool,
commit_uuid: Option<Uuid>,
snapshot_properties: HashMap<String, String>,
added_data_files: Vec<DataFile>,
removed_data_file_paths: HashSet<String>,
replace: bool,
}
impl OverwriteFilesAction {
pub(crate) fn new() -> Self {
Self {
check_duplicate: true,
commit_uuid: None,
snapshot_properties: HashMap::default(),
added_data_files: vec![],
removed_data_file_paths: HashSet::default(),
replace: false,
}
}
/// Commit as a **`replace`** snapshot instead of `overwrite`: the same file
/// swap, declared row-preserving — what a compaction / Spark's
/// `rewrite_data_files` commits. Readers (a changelog) treat it as no change,
/// so only use it when the added files hold exactly the removed files' live
/// rows. skade patch, 2026-10-02.
pub fn as_replace(mut self) -> Self {
self.replace = true;
self
}
/// Set whether to check that no added file path is already live in the
/// table. Costs one manifest walk; on by default, as for a fast append.
pub fn with_check_duplicate(mut self, v: bool) -> Self {
self.check_duplicate = v;
self
}
/// Data files this overwrite adds.
pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self {
self.added_data_files.extend(data_files);
self
}
/// Data files this overwrite removes, by exact `file_path`.
pub fn delete_files(mut self, paths: impl IntoIterator<Item = String>) -> Self {
self.removed_data_file_paths.extend(paths);
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 OverwriteFilesAction {
async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
// An overwrite that neither adds nor removes anything is a no-op, and a
// no-op that produces a snapshot is worse than an error: it grows the
// metadata tree and lets a caller believe a merge did something. Refuse
// it here so `merge_into` has a reason to skip the commit entirely
// rather than "successfully" writing nothing.
if self.added_data_files.is_empty() && self.removed_data_file_paths.is_empty() {
return Err(Error::new(
ErrorKind::PreconditionFailed,
"Overwrite adds no data files and removes no data files — refusing to commit an \
empty snapshot",
));
}
// A path cannot be both replaced and re-added: the manifest rewrite
// would emit the same path as `Deleted` and `Added` in one snapshot and
// a reader could legitimately resolve it either way.
let re_added: Vec<&str> = self
.added_data_files
.iter()
.map(|f| f.file_path.as_str())
.filter(|p| self.removed_data_file_paths.contains(*p))
.collect();
if !re_added.is_empty() {
return Err(Error::new(
ErrorKind::DataInvalid,
format!(
"Overwrite both removes and adds the same data file(s): {}",
re_added.join(", ")
),
));
}
let mut snapshot_properties = self.snapshot_properties.clone();
// Always non-empty: `SnapshotProducer::manifest_file` refuses a snapshot
// with no added data files AND no added snapshot properties, and a pure
// delete (`WHEN MATCHED THEN DELETE` that empties a file) adds none.
snapshot_properties.insert(
REMOVED_DATA_FILES_PROP.to_string(),
self.removed_data_file_paths.len().to_string(),
);
let removed_files = live_files_at(table, &self.removed_data_file_paths).await?;
let snapshot_producer = SnapshotProducer::new(
table,
self.commit_uuid.unwrap_or_else(Uuid::now_v7),
snapshot_properties,
self.added_data_files.clone(),
);
snapshot_producer.validate_added_data_files()?;
if self.check_duplicate {
snapshot_producer.validate_duplicate_files().await?;
}
snapshot_producer
.commit(
OverwriteFilesOperation {
removed_data_file_paths: self.removed_data_file_paths.clone(),
replace: self.replace,
removed_files,
},
DefaultManifestProcess,
)
.await
}
}
struct OverwriteFilesOperation {
removed_data_file_paths: HashSet<String>,
replace: bool,
/// The live `DataFile`s behind `removed_data_file_paths`, read before the
/// producer runs: the snapshot summary subtracts them
/// ([`SnapshotProduceOperation::removed_data_files`]).
removed_files: Vec<DataFile>,
}
/// The live data files of the current snapshot whose paths are in `paths`.
async fn live_files_at(table: &Table, paths: &HashSet<String>) -> Result<Vec<DataFile>> {
let mut out = Vec::new();
if paths.is_empty() {
return Ok(out);
}
let Some(snapshot) = table.metadata().current_snapshot() else { return Ok(out) };
let manifest_list = table.manifest_list_reader(snapshot).load().await?;
for list_entry in manifest_list.entries() {
if list_entry.content != ManifestContentType::Data
|| !(list_entry.has_added_files() || list_entry.has_existing_files())
{
continue;
}
let manifest = table.manifest_reader().read(list_entry).await?;
for e in manifest.entries() {
if e.is_alive() && paths.contains(e.file_path()) {
out.push(e.data_file().clone());
}
}
}
Ok(out)
}
impl OverwriteFilesOperation {
/// A manifest writer for the rewritten copy of `ordinal`-th data manifest,
/// stamped with the producer's new snapshot id.
///
/// `SnapshotProducer::new_manifest_writer` is `&mut self` (it owns the
/// manifest-name counter) and `existing_manifest` only gets `&self`, so the
/// name is derived here from the commit uuid plus the ordinal of the
/// manifest being rewritten — unique within the attempt for the same reason
/// the producer's counter is.
fn rewrite_writer(
&self,
snapshot_produce: &SnapshotProducer<'_>,
ordinal: usize,
) -> Result<ManifestWriter> {
let table = snapshot_produce.table;
// Same location and encryption rules as `SnapshotProducer::new_manifest_writer`
// (`write.metadata.path`, the table's encryption manager), so a rewritten
// manifest lands and is protected exactly like a freshly written one.
let path = format!(
"{}/{}-ow{}.{}",
table.metadata().metadata_location()?,
snapshot_produce.commit_uuid(),
ordinal,
DataFileFormat::Avro
);
let output_file = table.file_io().new_output(path)?;
let schema = table.metadata().current_schema().clone();
let partition_spec = table.metadata().default_partition_spec().as_ref().clone();
let snapshot_id = Some(snapshot_produce.snapshot_id());
let builder = if let Some(em) = table.encryption_manager() {
ManifestWriterBuilder::new_from_encrypted(
em.encrypt(output_file),
snapshot_id,
schema,
partition_spec,
)?
} else {
ManifestWriterBuilder::new(output_file, snapshot_id, schema, partition_spec)
};
Ok(match table.metadata().format_version() {
FormatVersion::V1 => builder.build_v1(),
FormatVersion::V2 => builder.build_v2_data(),
FormatVersion::V3 => builder.build_v3_data(),
})
}
}
impl SnapshotProduceOperation for OverwriteFilesOperation {
fn operation(&self) -> Operation {
if self.replace { Operation::Replace } else { Operation::Overwrite }
}
fn removed_data_files(&self) -> Vec<DataFile> {
self.removed_files.clone()
}
/// Empty on purpose.
///
/// `SnapshotProducer::manifest_file` never calls this — it carries a
/// `# TODO Support process delete entries` where the call would go. The
/// removals this action performs are expressed where the producer actually
/// reads them: in [`Self::existing_manifest`], which is exactly what the
/// trait doc says an overwrite does.
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 Some(snapshot) = table.metadata().current_snapshot() else {
// Nothing live to replace. If the caller named removals anyway they
// are not in this table, and silently dropping them would be the
// "silent no-op" this whole action exists to avoid.
if !self.removed_data_file_paths.is_empty() {
return Err(Error::new(
ErrorKind::DataInvalid,
"Overwrite names data files to remove, but the table has no current snapshot",
));
}
return Ok(vec![]);
};
let manifest_list = table.manifest_list_reader(snapshot).load().await?;
let mut carried: Vec<ManifestFile> = Vec::new();
let mut seen_removals: HashSet<String> = HashSet::new();
for (ordinal, list_entry) in manifest_list.entries().iter().enumerate() {
let alive = list_entry.has_added_files() || list_entry.has_existing_files();
// Delete manifests belong to the merge-on-read path; carry them.
if list_entry.content != ManifestContentType::Data {
if alive {
carried.push(list_entry.clone());
}
continue;
}
if !alive {
continue;
}
let manifest = table.manifest_reader().read(list_entry).await?;
let live: Vec<&ManifestEntry> = manifest
.entries()
.iter()
.filter(|e| e.is_alive())
.map(|e| e.as_ref())
.collect();
let hits: Vec<&&ManifestEntry> = live
.iter()
.filter(|e| self.removed_data_file_paths.contains(e.file_path()))
.collect();
if hits.is_empty() {
carried.push(list_entry.clone());
continue;
}
for e in &hits {
seen_removals.insert(e.file_path().to_string());
}
// Every live entry is going away: drop the manifest rather than
// write one that reads back as empty forever.
if hits.len() == live.len() {
continue;
}
let mut writer = self.rewrite_writer(snapshot_produce, ordinal)?;
for entry in live {
let seq = entry.sequence_number().ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"Manifest entry for {} has no data sequence number; cannot rewrite \
the manifest that holds it",
entry.file_path()
),
)
})?;
let file_seq = entry.file_sequence_number;
if self.removed_data_file_paths.contains(entry.file_path()) {
writer.add_delete_file(entry.data_file().clone(), seq, file_seq)?;
} else {
let snap = entry.snapshot_id().ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"Manifest entry for {} has no snapshot id; cannot carry it \
forward as an existing file",
entry.file_path()
),
)
})?;
writer.add_existing_file(entry.data_file().clone(), snap, seq, file_seq)?;
}
}
carried.push(writer.write_manifest_file().await?);
}
// A removal that matched nothing means the caller's view of the table is
// stale — the file was already replaced by a concurrent commit. Failing
// here is what makes "a merge whose commit fails leaves the table
// exactly as it was" true instead of "the merge dropped half its work".
let missed: Vec<&str> = self
.removed_data_file_paths
.iter()
.map(|s| s.as_str())
.filter(|p| !seen_removals.contains(*p))
.collect();
if !missed.is_empty() {
return Err(Error::new(
ErrorKind::DataInvalid,
format!(
"Overwrite names {} data file(s) to remove that are not live in the current \
snapshot: {}",
missed.len(),
missed.join(", ")
),
));
}
Ok(carried)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::spec::{
DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, Operation, Struct,
};
use crate::table::Table;
use crate::transaction::tests::make_v2_minimal_table;
use crate::transaction::{Transaction, TransactionAction};
use crate::{TableRequirement, TableUpdate};
fn data_file(path: &str, table: &Table) -> 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(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(300))]))
.build()
.unwrap()
}
/// Commit `action` and fold its updates into a new [`Table`], the way a
/// catalog would.
async fn advance(table: Table, action: impl TransactionAction + 'static) -> Table {
let mut commit = Arc::new(action).commit(&table).await.unwrap();
let updates = commit.take_updates();
Transaction::update_table_metadata(table, &updates).unwrap()
}
/// An overwrite that adds nothing and removes nothing is refused, not
/// committed as an empty snapshot.
#[tokio::test]
async fn test_empty_overwrite_is_refused() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let err = Arc::new(tx.overwrite_files())
.commit(&table)
.await
.err()
.expect("an overwrite with no adds and no removes must be refused");
assert!(
err.to_string().contains("removes no data files"),
"the refusal must say what was empty, got: {err}"
);
}
/// The same path may not be both removed and re-added: the rewritten
/// manifest would carry it as `Deleted` and `Added` in one snapshot.
#[tokio::test]
async fn test_overwrite_refuses_a_path_it_both_removes_and_adds() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let action = tx
.overwrite_files()
.add_data_files(vec![data_file("test/1.parquet", &table)])
.delete_files(["test/1.parquet".to_string()]);
let err = Arc::new(action)
.commit(&table)
.await
.err()
.expect("removing and adding the same path must be refused");
assert!(
err.to_string().contains("removes and adds the same data file"),
"got: {err}"
);
}
/// A removal naming a file that is not live fails the whole commit, rather
/// than being dropped on the floor. This is what makes an overwrite built
/// on a stale view of the table safe.
#[tokio::test]
async fn test_overwrite_refuses_a_removal_that_is_not_live() {
let table = make_v2_minimal_table();
let table = advance(
table.clone(),
Transaction::new(&table)
.fast_append()
.add_data_files(vec![data_file("test/1.parquet", &table)]),
)
.await;
let tx = Transaction::new(&table);
let action = tx
.overwrite_files()
.add_data_files(vec![data_file("test/9.parquet", &table)])
.delete_files(["test/nope.parquet".to_string()]);
let err = Arc::new(action)
.commit(&table)
.await
.err()
.expect("a removal that matches no live file must fail the commit");
assert!(
err.to_string().contains("not live in the current snapshot"),
"got: {err}"
);
}
/// A manifest ALL of whose live entries are removed is dropped from the new
/// manifest list, not carried as a manifest that reads back empty forever.
#[tokio::test]
async fn test_overwrite_drops_a_wholly_replaced_manifest() {
let table = make_v2_minimal_table();
let table = advance(
table.clone(),
Transaction::new(&table)
.fast_append()
.add_data_files(vec![data_file("test/1.parquet", &table)]),
)
.await;
let tx = Transaction::new(&table);
let action = tx
.overwrite_files()
.add_data_files(vec![data_file("test/2.parquet", &table)])
.delete_files(["test/1.parquet".to_string()]);
let mut commit = Arc::new(action).commit(&table).await.unwrap();
let updates = commit.take_updates();
let requirements = commit.take_requirements();
let TableUpdate::AddSnapshot { snapshot } = &updates[0] else {
unreachable!()
};
assert_eq!(snapshot.summary().operation, Operation::Overwrite);
assert_eq!(
snapshot
.summary()
.additional_properties
.get(super::REMOVED_DATA_FILES_PROP)
.map(String::as_str),
Some("1")
);
assert_eq!(requirements, vec![
TableRequirement::UuidMatch {
uuid: table.metadata().uuid()
},
TableRequirement::RefSnapshotIdMatch {
r#ref: crate::spec::MAIN_BRANCH.to_string(),
snapshot_id: table.metadata().current_snapshot_id
}
]);
let list = table
.manifest_list_reader(&Arc::new(snapshot.clone()))
.load()
.await
.unwrap();
assert_eq!(
list.entries().len(),
1,
"the manifest holding only the replaced file is gone; only the added manifest is left"
);
let manifest = table
.manifest_reader()
.read(&list.entries()[0])
.await
.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(manifest.entries()[0].file_path(), "test/2.parquet");
}
/// A manifest with a survivor is REWRITTEN: the survivor as `Existing` with
/// its original snapshot id and sequence numbers, the replaced file as
/// `Deleted`.
#[tokio::test]
async fn test_overwrite_rewrites_a_partially_replaced_manifest() {
let table = make_v2_minimal_table();
let table = advance(
table.clone(),
Transaction::new(&table).fast_append().add_data_files(vec![
data_file("test/1.parquet", &table),
data_file("test/2.parquet", &table),
]),
)
.await;
let appended_snapshot = table.metadata().current_snapshot_id().unwrap();
let appended_seq = table.metadata().last_sequence_number();
let tx = Transaction::new(&table);
let action = tx
.overwrite_files()
.add_data_files(vec![data_file("test/3.parquet", &table)])
.delete_files(["test/1.parquet".to_string()]);
let mut commit = Arc::new(action).commit(&table).await.unwrap();
let updates = commit.take_updates();
let TableUpdate::AddSnapshot { snapshot } = &updates[0] else {
unreachable!()
};
let list = table
.manifest_list_reader(&Arc::new(snapshot.clone()))
.load()
.await
.unwrap();
assert_eq!(list.entries().len(), 2, "the rewritten manifest + the added one");
let mut seen: Vec<(String, String)> = Vec::new();
for entry in list.entries() {
for e in table.manifest_reader().read(entry).await.unwrap().entries() {
seen.push((e.file_path().to_string(), format!("{:?}", e.status())));
if e.file_path() == "test/2.parquet" {
assert_eq!(
e.snapshot_id(),
Some(appended_snapshot),
"a carried-forward entry keeps the snapshot that added it"
);
assert_eq!(
e.sequence_number(),
Some(appended_seq),
"and its original data sequence number"
);
}
}
}
seen.sort();
assert_eq!(seen, vec![
("test/1.parquet".to_string(), "Deleted".to_string()),
("test/2.parquet".to_string(), "Existing".to_string()),
("test/3.parquet".to_string(), "Added".to_string()),
]);
}
}