You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/10/15 19:45:04 UTC
(iceberg-rust) branch gh-readonly-queue/main/pr-3253-b3470c8463c281c6e00ea9eee402881bb6c6d1fd created (now e4bec84bc)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-3253-b3470c8463c281c6e00ea9eee402881bb6c6d1fd
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at e4bec84bc feat(transaction): let a snapshot operation report removed files and rewrite manifests (#3253)
This branch includes the following new commits:
new e4bec84bc feat(transaction): let a snapshot operation report removed files and rewrite manifests (#3253)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: feat(transaction): let a snapshot operation report removed files and rewrite manifests (#3253)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3253-b3470c8463c281c6e00ea9eee402881bb6c6d1fd
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit e4bec84bcca5ba4f3e19583e95b1de7c1fab2559
Author: Jegor van Opdorp <je...@gmail.com>
AuthorDate: Thu Oct 15 18:28:41 2026 +0000
feat(transaction): let a snapshot operation report removed files and rewrite manifests (#3253)
* Let a snapshot operation report the files it removed
The snapshot summary could only count what the caller asked to remove, so an
operation that skips a path it never had, or is handed the same path twice,
inflated deleted-data-files and deleted-records. SnapshotProduceOperation now
reports the entries it actually marked Deleted, each with the schema and
partition spec of the manifest that recorded it, and the summary counts those.
Two changes follow from it. The manifests are produced before the summary,
since the counts are only known once the operation has run, which means the
added data files are cloned rather than taken. And the full-table truncate that
drops the previous totals is gated on a new operation hook instead of on the
operation being Overwrite: a partial overwrite keeps the totals, and with
truncate off update_totals needs a saturating subtraction so removing more than
the previous total does not underflow.
* Keep manifest names unique across a commit
The manifest name is built inline in new_manifest_writer from the metadata
location, the commit uuid and a counter, so anything else that needs to write a
manifest for the same commit has to reinvent it, and a name invented elsewhere
is neither unique within the commit nor findable by orphan cleanup after a
failed one. Pull the name out into new_manifest_path, and make the counter an
AtomicU64 so it can be bumped behind a shared reference.
* Write a manifest under the schema and spec it was written with
new_manifest_writer hardcoded the table's current schema and default partition
spec, so an operation that rewrites an existing manifest could not keep the
schema and spec that manifest's entries were recorded under, and building a
writer by hand instead means losing the encryption manager the method already
branches on: on an encrypted table the manifest would go out in plaintext while
its key metadata said otherwise. The schema and partition spec become arguments,
and the added-files path passes the table's current ones.
* test(transaction): cover partial overwrite and per-file spec in snapshot summary
Co-authored-by: Marco Kroll <21...@users.noreply.github.com>
* Drop the full-table truncation from the snapshot summary
Java and PyIceberg derive totals from the files an operation marks deleted, with no truncation path.
* Omit a summary total that would go negative, as Java does
* Name the removed-file fields and take the manifest spec by ref
* Take the added files again instead of cloning them for the summary
* Assert an operation reports exactly the files it marked deleted
* Say in the removed_data_files doc what it must match
* Summarize deleted files from the producer's list, as pyiceberg does
Drops the removed_data_files hook and the manifest-first reorder, and truncation defaults back to every overwrite.
* Check that deletion summaries match manifest entries
* Format dropped-source integer partition values
---------
Co-authored-by: Marco Kroll <21...@users.noreply.github.com>
---
crates/iceberg/src/spec/partition.rs | 47 ++++++
crates/iceberg/src/spec/snapshot_summary.rs | 24 ++-
crates/iceberg/src/spec/values/datum.rs | 8 +-
crates/iceberg/src/transaction/append.rs | 1 +
crates/iceberg/src/transaction/snapshot.rs | 223 +++++++++++++++++++++++++---
5 files changed, 278 insertions(+), 25 deletions(-)
diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs
index f3c9f9e54..fdb5a4a7b 100644
--- a/crates/iceberg/src/spec/partition.rs
+++ b/crates/iceberg/src/spec/partition.rs
@@ -1913,6 +1913,53 @@ mod tests {
assert!(!spec.has_sequential_ids());
}
+ #[test]
+ fn test_partition_to_path_with_dropped_source_column() {
+ for (source_type, transform, value, expected) in [
+ (
+ PrimitiveType::Int,
+ Transform::Identity,
+ Literal::int(1),
+ "p=1",
+ ),
+ (
+ PrimitiveType::Timestamp,
+ Transform::Identity,
+ Literal::long(1),
+ "p=1",
+ ),
+ (
+ PrimitiveType::Int,
+ Transform::Truncate(10),
+ Literal::int(10),
+ "p=10",
+ ),
+ ] {
+ let schema = Schema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "kept", PrimitiveType::Int.into()).into(),
+ NestedField::required(2, "dropped", source_type.into()).into(),
+ ])
+ .build()
+ .unwrap();
+ let spec = PartitionSpec::builder(schema)
+ .add_partition_field("dropped", "p", transform)
+ .unwrap()
+ .build()
+ .unwrap();
+ let current_schema = Schema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "kept", PrimitiveType::Int.into()).into(),
+ ])
+ .build()
+ .unwrap();
+ assert_eq!(
+ spec.partition_to_path(&Struct::from_iter([Some(value)]), current_schema.into()),
+ expected
+ );
+ }
+ }
+
#[test]
fn test_partition_to_path() {
let schema = Schema::builder()
diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs
index f8e69c4c8..d68a49d14 100644
--- a/crates/iceberg/src/spec/snapshot_summary.rs
+++ b/crates/iceberg/src/spec/snapshot_summary.rs
@@ -530,7 +530,10 @@ fn update_totals(
return;
};
- let new_total = previous_total + added - removed;
+ // Like Java, a total that would go negative is omitted rather than written as a false basis.
+ let Some(new_total) = previous_total.saturating_add(added).checked_sub(removed) else {
+ return;
+ };
summary
.additional_properties
.insert(total_property.to_string(), new_total.to_string());
@@ -1274,4 +1277,23 @@ mod tests {
assert_eq!(props.get(TOTAL_POSITION_DELETES).unwrap(), "2");
assert_eq!(props.get(TOTAL_EQUALITY_DELETES).unwrap(), "1");
}
+
+ #[test]
+ fn test_update_totals_omits_total_when_removed_exceeds_previous_total() {
+ let previous_summary = Summary {
+ operation: Operation::Append,
+ additional_properties: HashMap::from([(TOTAL_DATA_FILES.to_string(), "2".to_string())]),
+ };
+ let summary = Summary {
+ operation: Operation::Delete,
+ additional_properties: HashMap::from([(
+ DELETED_DATA_FILES.to_string(),
+ "3".to_string(),
+ )]),
+ };
+
+ let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap();
+
+ assert!(!updated.additional_properties.contains_key(TOTAL_DATA_FILES));
+ }
}
diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs
index a501466b3..2a7bd3f44 100644
--- a/crates/iceberg/src/spec/values/datum.rs
+++ b/crates/iceberg/src/spec/values/datum.rs
@@ -294,8 +294,12 @@ 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, PrimitiveLiteral::Int(val)) => write!(f, "{val}"),
- (PrimitiveType::Long, PrimitiveLiteral::Long(val)) => write!(f, "{val}"),
+ (PrimitiveType::Int | PrimitiveType::Unknown, PrimitiveLiteral::Int(val)) => {
+ write!(f, "{val}")
+ }
+ (PrimitiveType::Long | PrimitiveType::Unknown, PrimitiveLiteral::Long(val)) => {
+ write!(f, "{val}")
+ }
(_, PrimitiveLiteral::Float(val)) => write!(f, "{val}"),
(_, PrimitiveLiteral::Double(val)) => write!(f, "{val}"),
(PrimitiveType::Date, PrimitiveLiteral::Int(val)) => {
diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs
index 2b6f01044..e3e74338c 100644
--- a/crates/iceberg/src/transaction/append.rs
+++ b/crates/iceberg/src/transaction/append.rs
@@ -100,6 +100,7 @@ impl TransactionAction for FastAppendAction {
self.commit_uuid.unwrap_or_else(Uuid::now_v7),
self.snapshot_properties.clone(),
self.dedupe_added_files(),
+ vec![],
);
// validate added files
diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index a0ce1fba1..c30175001 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -17,7 +17,7 @@
use std::collections::{HashMap, HashSet};
use std::future::Future;
-use std::ops::RangeFrom;
+use std::sync::atomic::{AtomicU64, Ordering};
use futures::TryStreamExt;
use futures::stream::FuturesUnordered;
@@ -26,9 +26,10 @@ 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,
+ ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation,
+ PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention,
+ SnapshotSummaryCollector, Struct, StructType, Summary, TableProperties,
+ update_snapshot_summaries,
};
use crate::table::Table;
use crate::transaction::ActionCommit;
@@ -85,6 +86,12 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync {
&self,
snapshot_produce: &SnapshotProducer<'_>,
) -> impl Future<Output = Result<Vec<ManifestFile>>> + Send;
+
+ /// Returns whether the snapshot summary resets the table totals. Defaults to every
+ /// overwrite; an operation that overwrites only some files returns `false`.
+ fn truncate_full_table(&self) -> bool {
+ self.operation() == Operation::Overwrite
+ }
}
pub(crate) struct DefaultManifestProcess;
@@ -113,10 +120,10 @@ pub(crate) struct SnapshotProducer<'a> {
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>,
+ /// Must match the data files marked Deleted by this commit, including their metrics.
+ deleted_data_files: Vec<DataFile>,
+ // Numbers this commit's manifest files. Atomic so `new_manifest_writer` can take `&self`.
+ manifest_counter: AtomicU64,
}
impl<'a> SnapshotProducer<'a> {
@@ -125,6 +132,7 @@ impl<'a> SnapshotProducer<'a> {
commit_uuid: Uuid,
snapshot_properties: HashMap<String, String>,
added_data_files: Vec<DataFile>,
+ deleted_data_files: Vec<DataFile>,
) -> Self {
Self {
table,
@@ -132,7 +140,8 @@ impl<'a> SnapshotProducer<'a> {
commit_uuid,
snapshot_properties,
added_data_files,
- manifest_counter: (0..),
+ deleted_data_files,
+ manifest_counter: AtomicU64::new(0),
}
}
@@ -229,22 +238,23 @@ impl<'a> SnapshotProducer<'a> {
snapshot_id
}
- fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result<ManifestWriter> {
+ /// Returns a writer for the next manifest file of this commit, written under the given
+ /// schema and partition spec.
+ pub(crate) fn new_manifest_writer(
+ &self,
+ content: ManifestContentType,
+ schema: SchemaRef,
+ partition_spec: PartitionSpecRef,
+ ) -> Result<ManifestWriter> {
let new_manifest_path = format!(
"{}/{}-m{}.{}",
self.table.metadata().metadata_location()?,
self.commit_uuid,
- self.manifest_counter.next().unwrap(),
+ self.manifest_counter.fetch_add(1, Ordering::Relaxed),
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 partition_spec = partition_spec.as_ref().clone();
let builder = if let Some(em) = self.table.encryption_manager() {
ManifestWriterBuilder::new_from_encrypted(
@@ -323,7 +333,11 @@ impl<'a> SnapshotProducer<'a> {
builder.build()
}
});
- let mut writer = self.new_manifest_writer(ManifestContentType::Data)?;
+ let mut writer = self.new_manifest_writer(
+ ManifestContentType::Data,
+ self.table.metadata().current_schema().clone(),
+ self.table.metadata().default_partition_spec().clone(),
+ )?;
for entry in manifest_entries {
writer.add_entry(entry)?;
}
@@ -342,10 +356,13 @@ impl<'a> SnapshotProducer<'a> {
// TODO: Allowing snapshot property setup with no added data files is a workaround.
// We should clean it up after all necessary actions are supported.
// For details, please refer to https://github.com/apache/iceberg-rust/issues/1548
- if self.added_data_files.is_empty() && self.snapshot_properties.is_empty() {
+ if self.added_data_files.is_empty()
+ && self.deleted_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",
+ "No added or deleted data files or added snapshot properties found when write a manifest file",
));
}
@@ -361,6 +378,18 @@ impl<'a> SnapshotProducer<'a> {
// # TODO
// Support process delete entries.
+ debug_assert_eq!(
+ manifest_files
+ .iter()
+ .filter(|m| m.added_snapshot_id == self.snapshot_id
+ && m.content == ManifestContentType::Data)
+ .filter_map(|m| m.deleted_files_count)
+ .map(u64::from)
+ .sum::<u64>(),
+ self.deleted_data_files.len() as u64,
+ "deleted files must match this commit's manifest entries"
+ );
+
let manifest_files = manifest_process.process_manifests(self, manifest_files);
Ok(manifest_files)
}
@@ -396,6 +425,19 @@ impl<'a> SnapshotProducer<'a> {
);
}
+ for data_file in &self.deleted_data_files {
+ let partition_spec = table_metadata
+ .partition_spec_by_id(data_file.partition_spec_id)
+ .ok_or_else(|| {
+ invalid_data!("Unknown partition spec id {}", data_file.partition_spec_id)
+ })?;
+ summary_collector.remove_file(
+ data_file,
+ table_metadata.current_schema().clone(),
+ partition_spec.clone(),
+ );
+ }
+
let previous_snapshot = table_metadata.current_snapshot();
// User-supplied snapshot properties are applied first, then the computed
@@ -414,7 +456,7 @@ impl<'a> SnapshotProducer<'a> {
update_snapshot_summaries(
summary,
previous_snapshot.map(|s| s.summary()),
- snapshot_produce_operation.operation() == Operation::Overwrite,
+ snapshot_produce_operation.truncate_full_table(),
)
}
@@ -558,3 +600,140 @@ impl<'a> SnapshotProducer<'a> {
Ok(ActionCommit::new(updates, requirements))
}
}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use uuid::Uuid;
+
+ use super::*;
+ use crate::spec::{
+ DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec,
+ PrimitiveType, Schema, Transform, Type,
+ };
+ use crate::transaction::tests::make_v2_minimal_table;
+
+ struct DeleteOperation;
+
+ impl SnapshotProduceOperation for DeleteOperation {
+ fn operation(&self) -> Operation {
+ Operation::Delete
+ }
+
+ async fn delete_entries(&self, _: &SnapshotProducer<'_>) -> Result<Vec<ManifestEntry>> {
+ Ok(vec![])
+ }
+
+ async fn existing_manifest(&self, _: &SnapshotProducer<'_>) -> Result<Vec<ManifestFile>> {
+ Ok(vec![])
+ }
+ }
+
+ #[test]
+ fn test_summary_counts_deleted_data_files() {
+ let table = make_v2_minimal_table();
+ let deleted = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("test/deleted.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(2)
+ .partition_spec_id(table.metadata().default_partition_spec_id())
+ .partition(Struct::from_iter([Some(Literal::long(300))]))
+ .build()
+ .unwrap();
+
+ let producer = SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![
+ deleted,
+ ]);
+ let summary = producer.summary(&DeleteOperation).unwrap();
+ let properties = &summary.additional_properties;
+
+ assert_eq!(properties.get("deleted-data-files").unwrap(), "1");
+ assert_eq!(properties.get("deleted-records").unwrap(), "2");
+ assert_eq!(properties.get("removed-files-size").unwrap(), "100");
+ }
+
+ #[cfg(debug_assertions)]
+ #[tokio::test]
+ #[should_panic(expected = "deleted files must match this commit's manifest entries")]
+ async fn test_commit_rejects_unrecorded_deletions() {
+ let table = make_v2_minimal_table();
+ let deleted = DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("test/deleted.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(2)
+ .partition_spec_id(table.metadata().default_partition_spec_id())
+ .partition(Struct::from_iter([Some(Literal::long(300))]))
+ .build()
+ .unwrap();
+ SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![
+ deleted,
+ ])
+ .commit(DeleteOperation, DefaultManifestProcess)
+ .await
+ .unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_new_manifest_writer_uses_the_given_schema_and_spec() {
+ let table = make_v2_minimal_table();
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(7)
+ .with_fields(vec![
+ NestedField::required(1, "a", Type::Primitive(PrimitiveType::Long)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+ let partition_spec = Arc::new(
+ PartitionSpec::builder(schema.clone())
+ .with_spec_id(3)
+ .add_partition_field("a", "a", Transform::Identity)
+ .unwrap()
+ .build()
+ .unwrap(),
+ );
+
+ let producer =
+ SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![]);
+ let first = producer
+ .new_manifest_writer(
+ ManifestContentType::Data,
+ schema.clone(),
+ partition_spec.clone(),
+ )
+ .unwrap()
+ .write_manifest_file()
+ .await
+ .unwrap();
+ let second = producer
+ .new_manifest_writer(ManifestContentType::Data, schema.clone(), partition_spec)
+ .unwrap()
+ .write_manifest_file()
+ .await
+ .unwrap();
+
+ assert_ne!(first.manifest_path, second.manifest_path);
+
+ let bytes = table
+ .file_io()
+ .new_input(&first.manifest_path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let metadata = Manifest::parse_avro(&bytes).unwrap().into_parts().1;
+
+ // Not the table's defaults, which are schema 0 and spec 0.
+ assert_eq!(metadata.schema_id(), 7);
+ assert_eq!(metadata.schema().as_ref(), schema.as_ref());
+ assert_eq!(metadata.partition_spec().spec_id(), 3);
+ assert_eq!(first.partition_spec_id, 3);
+ }
+}