You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by li...@apache.org on 2026/07/22 08:04:19 UTC
(iceberg-rust) branch main updated: fix(transaction): de-duplicate file paths within a FastAppend batch (#2509)
This is an automated email from the ASF dual-hosted git repository.
blackmwk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 252bb601f fix(transaction): de-duplicate file paths within a FastAppend batch (#2509)
252bb601f is described below
commit 252bb601f71491aad6a716c57348935da44214c3
Author: Sreeram Garlapati <Sr...@users.noreply.github.com>
AuthorDate: Wed Jul 22 01:04:13 2026 -0700
fix(transaction): de-duplicate file paths within a FastAppend batch (#2509)
Closes #2507.
A FastAppend batch carrying the same file path twice writes both entries
into the manifest today, so scans double-count rows and the snapshot
summary is inflated.
This drops the duplicates and keeps the first occurrence, matching
Iceberg Java, which de-duplicates by file location via
[`DataFileSet`](https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/DataFileSet.java)
(used in
[`FastAppend`](https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/FastAppend.java)).
The dedupe is in-memory and always runs, independent of
`check_duplicate` (which only gates the I/O-bound cross-snapshot check).
Tested: a batch with repeated paths now commits a single manifest entry,
with and without `check_duplicate`.
Co-authored-by: Aman Rawat <ra...@gmail.com>
---
crates/iceberg/src/transaction/append.rs | 101 +++++++++++++++++++++++++++++--
1 file changed, 96 insertions(+), 5 deletions(-)
diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs
index dadd850c2..dabb46461 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;
+use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use async_trait::async_trait;
@@ -71,6 +71,18 @@ 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]
@@ -80,7 +92,7 @@ impl TransactionAction for FastAppendAction {
table,
self.commit_uuid.unwrap_or_else(Uuid::now_v7),
self.snapshot_properties.clone(),
- self.added_data_files.clone(),
+ self.dedupe_added_files(),
);
// validate added files
@@ -153,9 +165,9 @@ mod tests {
use crate::encryption::kms::MemoryKeyManagementClient;
use crate::io::FileIO;
use crate::spec::{
- DataContentType, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, ManifestEntry,
- ManifestListWriter, ManifestStatus, ManifestWriterBuilder, SnapshotRef, Struct,
- TableMetadata,
+ DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH,
+ ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, SnapshotRef,
+ Struct, TableMetadata,
};
use crate::table::Table;
use crate::test_utils::test_runtime;
@@ -354,6 +366,27 @@ mod tests {
);
}
+ /// Load the data files written by a single-manifest fast-append commit.
+ async fn committed_data_files(table: &Table, updates: &[TableUpdate]) -> Vec<DataFile> {
+ let TableUpdate::AddSnapshot { snapshot } = &updates[0] else {
+ unreachable!("first update is always AddSnapshot")
+ };
+ let manifest_list = table
+ .manifest_list_reader(&SnapshotRef::new(snapshot.clone()))
+ .load()
+ .await
+ .unwrap();
+ assert_eq!(1, manifest_list.entries().len());
+ manifest_list.entries()[0]
+ .load_manifest(table.file_io())
+ .await
+ .unwrap()
+ .entries()
+ .iter()
+ .map(|entry| entry.data_file().clone())
+ .collect()
+ }
+
#[tokio::test]
async fn test_empty_data_append_action() {
let table = make_v2_minimal_table();
@@ -691,6 +724,64 @@ mod tests {
assert!(Arc::new(action).commit(&table).await.is_err());
}
+ #[tokio::test]
+ async fn test_fast_append_dedupes_intra_batch_duplicate_paths() {
+ let table = make_v2_minimal_table();
+ let tx = Transaction::new(&table);
+
+ let make_file = |size: u64, records: u64| {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("test/dup.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(size)
+ .record_count(records)
+ .partition_spec_id(table.metadata().default_partition_spec_id())
+ .partition(Struct::from_iter([Some(Literal::long(1))]))
+ .build()
+ .unwrap()
+ };
+
+ // Same path three times: the manifest keeps a single entry, the first one.
+ let action = tx.fast_append().add_data_files(vec![
+ make_file(100, 10),
+ make_file(200, 20),
+ make_file(300, 30),
+ ]);
+ let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
+ let files = committed_data_files(&table, &action_commit.take_updates()).await;
+ assert_eq!(1, files.len());
+ assert_eq!(100, files[0].file_size_in_bytes());
+ }
+
+ #[tokio::test]
+ async fn test_fast_append_dedupes_regardless_of_check_duplicate_flag() {
+ let table = make_v2_minimal_table();
+ let tx = Transaction::new(&table);
+
+ let make_file = || {
+ DataFileBuilder::default()
+ .content(DataContentType::Data)
+ .file_path("test/dup.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(100)
+ .record_count(10)
+ .partition_spec_id(table.metadata().default_partition_spec_id())
+ .partition(Struct::from_iter([Some(Literal::long(1))]))
+ .build()
+ .unwrap()
+ };
+
+ // `check_duplicate` only gates the cross-snapshot check; intra-batch dedupe runs regardless.
+ let action = tx
+ .fast_append()
+ .with_check_duplicate(false)
+ .add_data_files(vec![make_file(), make_file()]);
+ let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
+ let files = committed_data_files(&table, &action_commit.take_updates()).await;
+ assert_eq!(1, files.len());
+ }
+
#[tokio::test]
async fn test_fast_append() {
let table = make_v2_minimal_table();