SreeramGarlapati commented on code in PR #2509:
URL: https://github.com/apache/iceberg-rust/pull/2509#discussion_r3609012034
##########
crates/iceberg/src/transaction/append.rs:
##########
@@ -84,6 +85,10 @@ impl FastAppendAction {
#[async_trait]
impl TransactionAction for FastAppendAction {
async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
+ if self.check_duplicate {
Review Comment:
@blackmwk agreed, fixed. `dedupe_added_files()` now runs unconditionally in
`commit()`; `check_duplicate` only gates `validate_duplicate_files()` (the
cross-snapshot check that touches existing files + costs io). the intra-batch
dedup of newly-added files always runs now. covered by
`test_fast_append_dedupes_regardless_of_check_duplicate_flag`.
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -508,3 +508,19 @@ impl<'a> SnapshotProducer<'a> {
Ok(ActionCommit::new(updates, requirements))
}
}
+
+pub(super) fn check_no_duplicate_paths_in_batch(files: &[DataFile]) ->
Result<()> {
Review Comment:
@blackmwk done - the dedupe now lives in
`FastAppendAction::dedupe_added_files()`, and `SnapshotProducer` just consumes
the already-deduped `added_data_files`. no dedup logic left in the producer.
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -164,11 +164,28 @@ impl<'a> SnapshotProducer<'a> {
}
pub(crate) async fn validate_duplicate_files(&self) -> Result<()> {
- let new_files: HashSet<&str> = self
- .added_data_files
- .iter()
- .map(|df| df.file_path.as_str())
- .collect();
+ // First, detect duplicate file paths within the batch itself.
+ // Collecting straight into a `HashSet` would silently deduplicate
+ // and pass corruption through to the manifest, so we walk the list
+ // explicitly and surface every distinct duplicated path.
+ let mut new_files: HashSet<&str> =
HashSet::with_capacity(self.added_data_files.len());
+ let mut duplicates_in_batch: HashSet<&str> = HashSet::new();
+ for data_file in &self.added_data_files {
+ if !new_files.insert(data_file.file_path.as_str()) {
+ duplicates_in_batch.insert(data_file.file_path.as_str());
+ }
+ }
+ if !duplicates_in_batch.is_empty() {
+ let mut paths: Vec<&str> =
duplicates_in_batch.into_iter().collect();
+ paths.sort_unstable();
Review Comment:
@xanderbailey no sort anymore - we dedupe the added paths in place and keep
first occurrence, so there's nothing to order.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
