vendor/iceberg-arrow59/src/transaction/snapshot.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.
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::ops::RangeFrom;
use futures::TryStreamExt;
use futures::stream::FuturesUnordered;
use uuid::Uuid;
use crate::error::{Result, invalid_data};
use crate::spec::{
DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry,
ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, Snapshot,
SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType, Summary,
TableProperties, update_snapshot_summaries,
};
use crate::table::Table;
use crate::transaction::ActionCommit;
use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
/// Process-global count of completed [`SnapshotProducer::validate_duplicate_files`]
/// scans — the O(live data files) manifest walk that runs on every fast append
/// whose `check_duplicate` is on.
///
/// It exists so a caller can *assert* whether the scan ran, rather than inferring
/// it from a wall-clock difference (which on a shallow table is pure noise — the
/// trap this whole knob was measured around). `skade`'s `duplicate_check_law`
/// test uses it to pin the `WriteProps::check_duplicate_files` → `FastAppendAction`
/// wiring: hardcode the flag anywhere along that path and the observed delta stops
/// matching. Mirrors the existing `skade_katalog::redb_commit_count` pattern.
static DUPLICATE_SCAN_COUNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
/// Read [`DUPLICATE_SCAN_COUNT`]. Monotonic for the life of the process; only
/// differences between two reads are meaningful.
pub fn duplicate_scan_count() -> u64 {
DUPLICATE_SCAN_COUNT.load(std::sync::atomic::Ordering::Relaxed)
}
/// A trait that defines how different table operations produce new snapshots.
///
/// `SnapshotProduceOperation` is used by [`SnapshotProducer`] to customize snapshot creation
/// based on the type of operation being performed (e.g., `Append`, `Overwrite`, `Delete`, etc.).
/// Each operation type implements this trait to specify:
/// - Which operation type to record in the snapshot summary
/// - Which existing manifest files should be included in the new snapshot
/// - Which manifest entries should be marked as deleted
///
/// # When it accomplishes
///
/// This trait is used during the snapshot creation process in [`SnapshotProducer::commit()`]:
///
/// 1. **Operation Type Recording**: The `operation()` method determines which operation type
/// (e.g., `Operation::Append`, `Operation::Overwrite`) is recorded in the snapshot summary.
/// This metadata helps track what kind of change was made to the table.
///
/// 2. **Manifest File Selection**: The `existing_manifest()` method determines which existing
/// manifest files from the current snapshot should be carried forward to the new snapshot.
/// For example:
/// - An `Append` operation typically includes all existing manifests plus new ones
/// - An `Overwrite` operation might exclude manifests for partitions being overwritten
///
/// 3. **Delete Entry Processing**: The `delete_entries()` method is intended for future delete
/// operations to specify which manifest entries should be marked as deleted.
pub(crate) trait SnapshotProduceOperation: Send + Sync {
/// The live data files this operation removes (an overwrite / replace):
/// what the snapshot summary subtracts, so `total-records` and
/// `total-data-files` stay the table's and `deleted-records` /
/// `removed-files-size` say what left. Empty for an append. skade patch
/// (2026-10-17): the producer builds its summary before the manifests
/// are rewritten, so the operation reports them up front.
fn removed_data_files(&self) -> Vec<DataFile> {
Vec::new()
}
/// Returns the operation type that will be recorded in the snapshot summary.
///
/// This determines what kind of operation is being performed (e.g., `Append`, `Overwrite`),
/// which is stored in the snapshot metadata for tracking and auditing purposes.
fn operation(&self) -> Operation;
/// Returns manifest entries that should be marked as deleted in the new snapshot.
#[allow(unused)]
fn delete_entries(
&self,
snapshot_produce: &SnapshotProducer,
) -> impl Future<Output = Result<Vec<ManifestEntry>>> + Send;
/// Returns existing manifest files that should be included in the new snapshot.
///
/// This method determines which manifest files from the current snapshot should be
/// carried forward to the new snapshot. The selection depends on the operation type:
///
/// - **Append operations**: Typically include all existing manifests
/// - **Overwrite operations**: May exclude manifests for partitions being overwritten
/// - **Delete operations**: May exclude manifests for partitions being deleted
fn existing_manifest(
&self,
snapshot_produce: &SnapshotProducer<'_>,
) -> impl Future<Output = Result<Vec<ManifestFile>>> + Send;
}
pub(crate) struct DefaultManifestProcess;
impl ManifestProcess for DefaultManifestProcess {
fn process_manifests(
&self,
_snapshot_produce: &SnapshotProducer<'_>,
manifests: Vec<ManifestFile>,
) -> Vec<ManifestFile> {
manifests
}
}
pub(crate) trait ManifestProcess: Send + Sync {
fn process_manifests(
&self,
snapshot_produce: &SnapshotProducer<'_>,
manifests: Vec<ManifestFile>,
) -> Vec<ManifestFile>;
}
pub(crate) struct SnapshotProducer<'a> {
pub(crate) table: &'a Table,
snapshot_id: i64,
commit_uuid: Uuid,
snapshot_properties: HashMap<String, String>,
added_data_files: Vec<DataFile>,
// A counter used to generate unique manifest file names.
// It starts from 0 and increments for each new manifest file.
// Note: This counter is limited to the range of (0..u64::MAX).
manifest_counter: RangeFrom<u64>,
}
impl<'a> SnapshotProducer<'a> {
pub(crate) fn new(
table: &'a Table,
commit_uuid: Uuid,
snapshot_properties: HashMap<String, String>,
added_data_files: Vec<DataFile>,
) -> Self {
Self {
table,
snapshot_id: Self::generate_unique_snapshot_id(table),
commit_uuid,
snapshot_properties,
added_data_files,
manifest_counter: (0..),
}
}
/// The snapshot id this producer will commit under.
///
/// Exposed for a [`SnapshotProduceOperation`] that has to write its OWN
/// manifests — `OverwriteFilesOperation` rewrites the manifests that hold
/// replaced data files, and a rewritten manifest's `Deleted` entries must
/// carry the *new* snapshot id. `existing_manifest` is handed `&self`, so it
/// cannot reach [`Self::new_manifest_writer`] (which needs `&mut self` for
/// the manifest counter); it builds its own writer from these.
pub(crate) fn snapshot_id(&self) -> i64 {
self.snapshot_id
}
/// The commit uuid, used to name manifests uniquely within one attempt.
pub(crate) fn commit_uuid(&self) -> Uuid {
self.commit_uuid
}
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"
));
}
// 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 {
return Err(invalid_data!(
"Data file partition spec id does not match table default partition spec id"
));
}
Self::validate_partition_value(
data_file.partition(),
self.table.metadata().default_partition_type(),
)?;
}
Ok(())
}
pub(crate) async fn validate_duplicate_files(&self) -> Result<()> {
DUPLICATE_SCAN_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(current_snapshot) = self.table.metadata().current_snapshot() else {
return Ok(());
};
let new_files: HashSet<&str> = self
.added_data_files
.iter()
.map(|df| df.file_path.as_str())
.collect();
let runtime = self.table.runtime();
let manifest_list = self
.table
.manifest_list_reader(current_snapshot)
.load()
.await?;
let new_files_ref = &new_files;
let referenced_files: Vec<String> = manifest_list
.consume_entries()
.into_iter()
.map(|entry| {
let reader = self.table.manifest_reader();
runtime.io().spawn(async move { reader.read(&entry).await })
})
.collect::<FuturesUnordered<_>>()
.try_fold(Vec::new(), |mut acc, manifest| async move {
acc.extend(
manifest?
.entries()
.iter()
.filter(|e| new_files_ref.contains(e.file_path()) && e.is_alive())
.map(|e| e.file_path().to_string()),
);
Ok(acc)
})
.await?;
if !referenced_files.is_empty() {
return Err(invalid_data!(
"Cannot add files that are already referenced by table, files: {}",
referenced_files.join(", ")
));
}
Ok(())
}
fn generate_unique_snapshot_id(table: &Table) -> i64 {
let generate_random_id = || -> i64 {
let (lhs, rhs) = Uuid::new_v4().as_u64_pair();
let snapshot_id = (lhs ^ rhs) as i64;
if snapshot_id < 0 {
-snapshot_id
} else {
snapshot_id
}
};
let mut snapshot_id = generate_random_id();
while table
.metadata()
.snapshots()
.any(|s| s.snapshot_id() == snapshot_id)
{
snapshot_id = generate_random_id();
}
snapshot_id
}
fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result<ManifestWriter> {
let new_manifest_path = format!(
"{}/{}-m{}.{}",
self.table.metadata().metadata_location()?,
self.commit_uuid,
self.manifest_counter.next().unwrap(),
DataFileFormat::Avro
);
let output_file = self.table.file_io().new_output(new_manifest_path)?;
let partition_spec = self
.table
.metadata()
.default_partition_spec()
.as_ref()
.clone();
let schema = self.table.metadata().current_schema().clone();
let builder = if let Some(em) = self.table.encryption_manager() {
ManifestWriterBuilder::new_from_encrypted(
em.encrypt(output_file),
Some(self.snapshot_id),
schema,
partition_spec,
)?
} else {
ManifestWriterBuilder::new(output_file, Some(self.snapshot_id), schema, partition_spec)
};
match self.table.metadata().format_version() {
FormatVersion::V1 => Ok(builder.build_v1()),
FormatVersion::V2 => match content {
ManifestContentType::Data => Ok(builder.build_v2_data()),
ManifestContentType::Deletes => Ok(builder.build_v2_deletes()),
},
FormatVersion::V3 => match content {
ManifestContentType::Data => Ok(builder.build_v3_data()),
ManifestContentType::Deletes => Ok(builder.build_v3_deletes()),
},
}
}
// Check if the partition value is compatible with the partition type.
fn validate_partition_value(
partition_value: &Struct,
partition_type: &StructType,
) -> Result<()> {
if partition_value.fields().len() != partition_type.fields().len() {
return Err(invalid_data!(
"Partition value is not compatible with partition type"
));
}
for (value, field) in partition_value.fields().iter().zip(partition_type.fields()) {
let field = field.field_type.as_primitive_type().ok_or_else(|| {
Error::new(
ErrorKind::Unexpected,
"Partition field should only be primitive type.",
)
})?;
if let Some(value) = value
&& !field.compatible(&value.as_primitive_literal().unwrap())
{
return Err(invalid_data!(
"Partition value is not compatible partition type"
));
}
}
Ok(())
}
// Write manifest file for added data files and return the ManifestFile for ManifestList.
async fn write_added_manifest(&mut self) -> Result<ManifestFile> {
let added_data_files = std::mem::take(&mut self.added_data_files);
if added_data_files.is_empty() {
return Err(Error::new(
ErrorKind::PreconditionFailed,
"No added data files found when write an added manifest file",
));
}
let snapshot_id = self.snapshot_id;
let format_version = self.table.metadata().format_version();
let manifest_entries = added_data_files.into_iter().map(|data_file| {
let builder = ManifestEntry::builder()
.status(crate::spec::ManifestStatus::Added)
.data_file(data_file);
if format_version == FormatVersion::V1 {
builder.snapshot_id(snapshot_id).build()
} else {
// For format version > 1, we set the snapshot id at the inherited time to avoid rewrite the manifest file when
// commit failed.
builder.build()
}
});
let mut writer = self.new_manifest_writer(ManifestContentType::Data)?;
for entry in manifest_entries {
writer.add_entry(entry)?;
}
writer.write_manifest_file().await
}
/// Creates new manifests for data files added or removed,
/// and collects all of the manifests to be included in the new snapshot as [ManifestFile] entries.
async fn produce_manifests<OP: SnapshotProduceOperation, MP: ManifestProcess>(
&mut self,
snapshot_produce_operation: &OP,
manifest_process: &MP,
) -> Result<Vec<ManifestFile>> {
// 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.
// 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() {
return Err(Error::new(
ErrorKind::PreconditionFailed,
"No added data files or added snapshot properties found when write a manifest file",
));
}
let existing_manifests = snapshot_produce_operation.existing_manifest(self).await?;
let mut manifest_files = existing_manifests;
// Process added entries.
if !self.added_data_files.is_empty() {
let added_manifest = self.write_added_manifest().await?;
manifest_files.push(added_manifest);
}
// # TODO
// Support process delete entries.
let manifest_files = manifest_process.process_manifests(self, manifest_files);
Ok(manifest_files)
}
// Returns a `Summary` of the current snapshot
fn summary<OP: SnapshotProduceOperation>(
&self,
snapshot_produce_operation: &OP,
) -> Result<Summary> {
let mut summary_collector = SnapshotSummaryCollector::default();
let table_metadata = self.table.metadata_ref();
let partition_summary_limit = if let Some(limit) = table_metadata
.properties()
.get(TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT)
{
if let Ok(limit) = limit.parse::<u64>() {
limit
} else {
TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT
}
} else {
TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT
};
summary_collector.set_partition_summary_limit(partition_summary_limit);
for data_file in &self.added_data_files {
summary_collector.add_file(
data_file,
table_metadata.current_schema().clone(),
table_metadata.default_partition_spec().clone(),
);
}
for data_file in snapshot_produce_operation.removed_data_files() {
summary_collector.remove_file(
&data_file,
table_metadata.current_schema().clone(),
table_metadata.default_partition_spec().clone(),
);
}
let previous_snapshot = table_metadata.current_snapshot();
// User-supplied snapshot properties are applied first, then the computed
// metrics overwrite any colliding keys. This matches iceberg-java
// (`SnapshotProducer.summary`), where computed `added-*`/`total-*` values
// are written after user properties so a user cannot shadow them with a
// bad (or merely wrong) value that would corrupt the snapshot summary.
let mut additional_properties = self.snapshot_properties.clone();
additional_properties.extend(summary_collector.build());
let summary = Summary {
operation: snapshot_produce_operation.operation(),
additional_properties,
};
update_snapshot_summaries(
summary,
previous_snapshot.map(|s| s.summary()),
snapshot_produce_operation.operation() == Operation::Overwrite,
)
}
fn generate_manifest_list_file_path(&self, attempt: i64) -> Result<String> {
Ok(format!(
"{}/snap-{}-{}-{}.{}",
self.table.metadata().metadata_location()?,
self.snapshot_id,
attempt,
self.commit_uuid,
DataFileFormat::Avro
))
}
/// Finished building the action and return the [`ActionCommit`] to the transaction.
pub(crate) async fn commit<OP: SnapshotProduceOperation, MP: ManifestProcess>(
mut self,
snapshot_produce_operation: OP,
process: MP,
) -> Result<ActionCommit> {
let manifest_list_path = self.generate_manifest_list_file_path(0)?;
let next_seq_num = self.table.metadata().next_sequence_number();
let first_row_id = self.table.metadata().next_row_id();
let raw_output = self
.table
.file_io()
.new_output(manifest_list_path.clone())?;
let (writer, encrypted_output) = match self.table.encryption_manager() {
Some(em) => {
let encrypted_output = em.encrypt(raw_output);
(
encrypted_output.writer().await?,
Some((em.clone(), encrypted_output)),
)
}
None => (raw_output.writer().await?, None),
};
let parent_snapshot_id = self.table.metadata().current_snapshot_id();
let mut manifest_list_writer = match self.table.metadata().format_version() {
FormatVersion::V1 => {
ManifestListWriter::v1(writer, self.snapshot_id, parent_snapshot_id)
}
FormatVersion::V2 => {
ManifestListWriter::v2(writer, self.snapshot_id, parent_snapshot_id, next_seq_num)
}
FormatVersion::V3 => ManifestListWriter::v3(
writer,
self.snapshot_id,
parent_snapshot_id,
next_seq_num,
Some(first_row_id),
),
};
// 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
.produce_manifests(&snapshot_produce_operation, &process)
.await?;
manifest_list_writer.add_manifests(new_manifests.into_iter())?;
let writer_next_row_id = manifest_list_writer.next_row_id();
let file_metadata = manifest_list_writer.close().await?;
let encryption_key_id = match encrypted_output {
Some((em, encrypted_output)) => Some(
em.encrypt_manifest_list_key_metadata(
&encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata),
)
.await?,
),
None => None,
};
let commit_ts = chrono::Utc::now().timestamp_millis();
let new_snapshot = Snapshot::builder()
.with_manifest_list(manifest_list_path)
.with_snapshot_id(self.snapshot_id)
.with_parent_snapshot_id(self.table.metadata().current_snapshot_id())
.with_sequence_number(next_seq_num)
.with_summary(summary)
.with_schema_id(self.table.metadata().current_schema_id())
.with_encryption_key_id(encryption_key_id)
.with_timestamp_ms(commit_ts);
let new_snapshot = if let Some(writer_next_row_id) = writer_next_row_id {
let assigned_rows = writer_next_row_id - self.table.metadata().next_row_id();
new_snapshot
.with_row_range(first_row_id, assigned_rows)
.build()
} else {
new_snapshot.build()
};
let encryption_key_updates: Vec<TableUpdate> = self
.table
.encryption_manager()
.map(|em| {
em.with_encryption_keys(|keys| {
keys.values()
.filter(|k| self.table.metadata().encryption_key(k.key_id()).is_none())
.map(|k| TableUpdate::AddEncryptionKey {
encryption_key: k.clone(),
})
.collect()
})
})
.unwrap_or_default();
let updates = [encryption_key_updates, vec![
TableUpdate::AddSnapshot {
snapshot: new_snapshot,
},
TableUpdate::SetSnapshotRef {
ref_name: MAIN_BRANCH.to_string(),
reference: SnapshotReference::new(
self.snapshot_id,
SnapshotRetention::branch(None, None, None),
),
},
]]
.concat();
let requirements = vec![
TableRequirement::UuidMatch {
uuid: self.table.metadata().uuid(),
},
TableRequirement::RefSnapshotIdMatch {
r#ref: MAIN_BRANCH.to_string(),
snapshot_id: self.table.metadata().current_snapshot_id(),
},
];
Ok(ActionCommit::new(updates, requirements))
}
}