This is an automated email from the ASF dual-hosted git repository.
liurenjie1024 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 bddffa17 fix(iceberg): add_files correctly check duplicates (#1395)
bddffa17 is described below
commit bddffa174dcafb7b3adc9d868edfb645a30f41aa
Author: Roman Shanin <[email protected]>
AuthorDate: Sat Jun 7 05:15:44 2025 +0300
fix(iceberg): add_files correctly check duplicates (#1395)
## Which issue does this PR close?
- Closes #1394.
## What changes are included in this PR?
- compare duplicates by loading manifest files and taking `file_path`
from it
- use direct calls instead of `scan`
## Are these changes tested?
- work for me local experiments
- fixed existing tests
- added a new test to showcase behavior
Co-authored-by: Roman Shanin <[email protected]>
---
crates/iceberg/src/transaction/append.rs | 119 ++++++---------------
.../tests/shared_tests/append_data_file_test.rs | 21 ----
2 files changed, 31 insertions(+), 109 deletions(-)
diff --git a/crates/iceberg/src/transaction/append.rs
b/crates/iceberg/src/transaction/append.rs
index d3b3cb2e..017107f9 100644
--- a/crates/iceberg/src/transaction/append.rs
+++ b/crates/iceberg/src/transaction/append.rs
@@ -17,8 +17,6 @@
use std::collections::{HashMap, HashSet};
-use arrow_array::StringArray;
-use futures::TryStreamExt;
use uuid::Uuid;
use crate::error::Result;
@@ -129,32 +127,19 @@ impl<'a> FastAppendAction<'a> {
.map(|df| df.file_path.as_str())
.collect();
- let mut manifest_stream = self
- .snapshot_produce_action
- .tx
- .current_table
- .inspect()
- .manifests()
- .scan()
- .await?;
let mut referenced_files = Vec::new();
-
- while let Some(batch) = manifest_stream.try_next().await? {
- let file_path_array = batch
- .column(1)
- .as_any()
- .downcast_ref::<StringArray>()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::DataInvalid,
- "Failed to downcast file_path column to
StringArray",
- )
- })?;
-
- for i in 0..batch.num_rows() {
- let file_path = file_path_array.value(i);
- if new_files.contains(file_path) {
- referenced_files.push(file_path.to_string());
+ let table = &self.snapshot_produce_action.tx.current_table;
+ if let Some(current_snapshot) =
table.metadata().current_snapshot() {
+ let manifest_list = current_snapshot
+ .load_manifest_list(table.file_io(), &table.metadata_ref())
+ .await?;
+ for manifest_list_entry in manifest_list.entries() {
+ let manifest =
manifest_list_entry.load_manifest(table.file_io()).await?;
+ for entry in manifest.entries() {
+ let file_path = entry.file_path();
+ if new_files.contains(file_path) && entry.is_alive() {
+ referenced_files.push(file_path.to_string());
+ }
}
}
}
@@ -364,81 +349,39 @@ mod tests {
}
#[tokio::test]
- async fn test_add_existing_parquet_files_to_unpartitioned_table() {
+ async fn test_add_duplicated_parquet_files_to_unpartitioned_table() {
let mut fixture = TableTestFixture::new_unpartitioned();
fixture.setup_unpartitioned_manifest_files().await;
let tx = crate::transaction::Transaction::new(&fixture.table);
let file_paths = vec![
format!("{}/1.parquet", &fixture.table_location),
- format!("{}/2.parquet", &fixture.table_location),
format!("{}/3.parquet", &fixture.table_location),
];
let fast_append_action = tx.fast_append(None, vec![]).unwrap();
- // Attempt to add the existing Parquet files with fast append.
- let new_tx = fast_append_action
- .add_parquet_files(file_paths.clone())
- .await
- .expect("Adding existing Parquet files should succeed");
-
- let mut found_add_snapshot = false;
- let mut found_set_snapshot_ref = false;
- for update in new_tx.updates.iter() {
- match update {
- TableUpdate::AddSnapshot { .. } => {
- found_add_snapshot = true;
- }
- TableUpdate::SetSnapshotRef {
- ref_name,
- reference,
- } => {
- found_set_snapshot_ref = true;
- assert_eq!(ref_name, MAIN_BRANCH);
- assert!(reference.snapshot_id > 0);
- }
- _ => {}
- }
- }
- assert!(found_add_snapshot);
- assert!(found_set_snapshot_ref);
-
- let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } =
&new_tx.updates[0] {
- snapshot
- } else {
- panic!("Expected the first update to be an AddSnapshot update");
- };
-
- let manifest_list = new_snapshot
- .load_manifest_list(fixture.table.file_io(),
fixture.table.metadata())
- .await
- .expect("Failed to load manifest list");
-
- assert_eq!(
- manifest_list.entries().len(),
- 2,
- "Expected 2 manifest list entries, got {}",
- manifest_list.entries().len()
+ // Attempt to add duplicated Parquet files with fast append.
+ assert!(
+ fast_append_action
+ .add_parquet_files(file_paths.clone())
+ .await
+ .is_err(),
+ "file already in table"
);
- // Load the manifest from the manifest list
- let manifest = manifest_list.entries()[0]
- .load_manifest(fixture.table.file_io())
- .await
- .expect("Failed to load manifest");
+ let file_paths = vec![format!("{}/2.parquet",
&fixture.table_location)];
- // Check that the manifest contains three entries.
- assert_eq!(manifest.entries().len(), 3);
+ let tx = crate::transaction::Transaction::new(&fixture.table);
+ let fast_append_action = tx.fast_append(None, vec![]).unwrap();
- // Verify each file path appears in manifest.
- let manifest_paths: Vec<String> = manifest
- .entries()
- .iter()
- .map(|entry| entry.data_file().file_path.clone())
- .collect();
- for path in file_paths {
- assert!(manifest_paths.contains(&path));
- }
+ // Attempt to add Parquet file which was deleted from table.
+ assert!(
+ fast_append_action
+ .add_parquet_files(file_paths.clone())
+ .await
+ .is_ok(),
+ "file not in table"
+ );
}
}
diff --git
a/crates/integration_tests/tests/shared_tests/append_data_file_test.rs
b/crates/integration_tests/tests/shared_tests/append_data_file_test.rs
index f3ee17c7..38a02951 100644
--- a/crates/integration_tests/tests/shared_tests/append_data_file_test.rs
+++ b/crates/integration_tests/tests/shared_tests/append_data_file_test.rs
@@ -129,25 +129,4 @@ async fn test_append_data_file() {
let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
assert_eq!(batches.len(), 1);
assert_eq!(batches[0], batch);
-
- // commit result again
- let tx = Transaction::new(&table);
- let mut append_action = tx.fast_append(None, vec![]).unwrap();
- append_action.add_data_files(data_file.clone()).unwrap();
- let tx = append_action.apply().await.unwrap();
- let table = tx.commit(&rest_catalog).await.unwrap();
-
- // check result again
- let batch_stream = table
- .scan()
- .select_all()
- .build()
- .unwrap()
- .to_arrow()
- .await
- .unwrap();
- let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
- assert_eq!(batches.len(), 2);
- assert_eq!(batches[0], batch);
- assert_eq!(batches[1], batch);
}
