vendor/iceberg-arrow59/src/transaction/append.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::sync::Arc;
use async_trait::async_trait;
use uuid::Uuid;
use crate::error::Result;
use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation};
use crate::table::Table;
use crate::transaction::snapshot::{
DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer,
};
use crate::transaction::{ActionCommit, TransactionAction};
/// FastAppendAction is a transaction action for fast append data files to the table.
pub struct FastAppendAction {
check_duplicate: bool,
// below are properties used to create SnapshotProducer when commit
commit_uuid: Option<Uuid>,
snapshot_properties: HashMap<String, String>,
added_data_files: Vec<DataFile>,
}
impl FastAppendAction {
pub(crate) fn new() -> Self {
Self {
check_duplicate: true,
commit_uuid: None,
snapshot_properties: HashMap::default(),
added_data_files: vec![],
}
}
/// Set whether to check duplicate files
pub fn with_check_duplicate(mut self, v: bool) -> Self {
self.check_duplicate = v;
self
}
/// Add data files to the snapshot.
pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self {
self.added_data_files.extend(data_files);
self
}
/// Set commit UUID for the snapshot.
pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self {
self.commit_uuid = Some(commit_uuid);
self
}
/// Set snapshot summary properties.
pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self {
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]
impl TransactionAction for FastAppendAction {
async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
let snapshot_producer = SnapshotProducer::new(
table,
self.commit_uuid.unwrap_or_else(Uuid::now_v7),
self.snapshot_properties.clone(),
self.dedupe_added_files(),
);
// validate added files
snapshot_producer.validate_added_data_files()?;
// Checks duplicate files
if self.check_duplicate {
snapshot_producer.validate_duplicate_files().await?;
}
snapshot_producer
.commit(FastAppendOperation, DefaultManifestProcess)
.await
}
}
struct FastAppendOperation;
impl SnapshotProduceOperation for FastAppendOperation {
fn operation(&self) -> Operation {
Operation::Append
}
async fn delete_entries(
&self,
_snapshot_produce: &SnapshotProducer<'_>,
) -> Result<Vec<ManifestEntry>> {
Ok(vec![])
}
async fn existing_manifest(
&self,
snapshot_produce: &SnapshotProducer<'_>,
) -> Result<Vec<ManifestFile>> {
let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else {
return Ok(vec![]);
};
let manifest_list = snapshot_produce
.table
.manifest_list_reader(snapshot)
.load()
.await?;
Ok(manifest_list
.entries()
.iter()
.filter(|entry| {
// Keep delete-only manifests too: they record which files were removed and
// must persist across snapshots until `expire_snapshots` cleans them up.
// Dropping them lets the removed files reappear as live data (see #2148).
entry.has_added_files() || entry.has_existing_files() || entry.has_deleted_files()
})
.cloned()
.collect())
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::fs;
use std::sync::Arc;
use minijinja::{AutoEscape, Environment, Value, context};
use tempfile::TempDir;
use uuid::Uuid;
use crate::encryption::kms::MemoryKeyManagementClient;
use crate::encryption::{SensitiveBytes, StandardKeyMetadata};
use crate::io::FileIO;
use crate::spec::{
DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH,
ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, SnapshotRef,
Struct, TableMetadata,
};
use crate::table::Table;
use crate::test_utils::{make_encrypted_table, test_runtime};
use crate::transaction::tests::make_v2_minimal_table;
use crate::transaction::{Transaction, TransactionAction};
use crate::{TableIdent, TableRequirement, TableUpdate};
fn render_template(template: &str, ctx: Value) -> String {
let mut env = Environment::new();
env.set_auto_escape_callback(|_| AutoEscape::None);
env.render_str(template, ctx).unwrap()
}
/// Builds a table whose current snapshot's manifest list contains a data manifest
/// followed by a delete-only manifest (one entry with `ManifestStatus::Deleted`,
/// so `deleted_files_count > 0` while `added_files_count == existing_files_count == 0`).
///
/// Returns the table plus the `manifest_path` of the delete-only manifest so callers
/// can assert whether a subsequent append carries it forward.
async fn make_table_with_delete_only_manifest() -> (Table, TempDir, String) {
let tmp_dir = TempDir::new().unwrap();
let table_location = tmp_dir.path().join("table1");
let manifest_list_location = table_location.join("metadata/manifests_list_1.avro");
let table_metadata_location = table_location.join("metadata/v1.json");
let file_io = FileIO::new_with_fs();
let template = fs::read_to_string(format!(
"{}/testdata/example_table_metadata_v2.json",
env!("CARGO_MANIFEST_DIR")
))
.unwrap();
// The template has two snapshots; point the current one at our manifest list.
let metadata_json = render_template(&template, context! {
table_location => &table_location,
manifest_list_1_location => &manifest_list_location,
manifest_list_2_location => &manifest_list_location,
table_metadata_1_location => &table_metadata_location,
});
let table_metadata = serde_json::from_str::<TableMetadata>(&metadata_json).unwrap();
let table = Table::builder()
.metadata(table_metadata)
.identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
.file_io(file_io)
.metadata_location(table_metadata_location.to_str().unwrap())
.runtime(test_runtime())
.build()
.unwrap();
let current_snapshot = table.metadata().current_snapshot().unwrap();
let schema = current_snapshot.schema(table.metadata()).unwrap();
let partition_spec = table.metadata().default_partition_spec();
let next_manifest_file = |location: &str| {
table
.file_io()
.new_output(format!(
"{}/metadata/manifest_{}.avro",
location,
Uuid::new_v4()
))
.unwrap()
};
let table_location_str = table_location.to_str().unwrap().to_string();
// Data manifest: one Added data file.
let mut data_writer = ManifestWriterBuilder::new(
next_manifest_file(&table_location_str),
Some(current_snapshot.snapshot_id()),
schema.clone(),
partition_spec.as_ref().clone(),
)
.build_v2_data();
data_writer
.add_entry(
ManifestEntry::builder()
.status(ManifestStatus::Added)
.data_file(
DataFileBuilder::default()
.partition_spec_id(0)
.content(DataContentType::Data)
.file_path(format!("{table_location_str}/data.parquet"))
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition(Struct::from_iter([Some(Literal::long(100))]))
.build()
.unwrap(),
)
.build(),
)
.unwrap();
let data_manifest = data_writer.write_manifest_file().await.unwrap();
// Delete-only manifest: a single Deleted entry, nothing added or existing.
let mut delete_writer = ManifestWriterBuilder::new(
next_manifest_file(&table_location_str),
Some(current_snapshot.snapshot_id()),
schema.clone(),
partition_spec.as_ref().clone(),
)
.build_v2_data();
delete_writer
.add_delete_entry(
ManifestEntry::builder()
.status(ManifestStatus::Deleted)
.sequence_number(0)
.file_sequence_number(0)
.data_file(
DataFileBuilder::default()
.partition_spec_id(0)
.content(DataContentType::Data)
.file_path(format!("{table_location_str}/removed.parquet"))
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition(Struct::from_iter([Some(Literal::long(100))]))
.build()
.unwrap(),
)
.build(),
)
.unwrap();
let delete_manifest = delete_writer.write_manifest_file().await.unwrap();
let delete_manifest_path = delete_manifest.manifest_path.clone();
// Sanity: the delete manifest really is delete-only.
assert!(delete_manifest.has_deleted_files());
assert!(!delete_manifest.has_added_files());
assert!(!delete_manifest.has_existing_files());
let mut manifest_list_writer = ManifestListWriter::v2(
table
.file_io()
.new_output(current_snapshot.manifest_list())
.unwrap()
.writer()
.await
.unwrap(),
current_snapshot.snapshot_id(),
current_snapshot.parent_snapshot_id(),
current_snapshot.sequence_number(),
);
manifest_list_writer
.add_manifests(vec![data_manifest, delete_manifest].into_iter())
.unwrap();
manifest_list_writer.close().await.unwrap();
(table, tmp_dir, delete_manifest_path)
}
/// Regression test for #2148: a `fast_append` must carry delete-only manifests
/// forward into the new snapshot. Dropping them lets the files they mark as
/// removed reappear as live data on the next append.
#[tokio::test]
async fn test_fast_append_preserves_delete_only_manifest() {
let (table, _tmp_dir, delete_manifest_path) = make_table_with_delete_only_manifest().await;
// Append a new data file via the public transaction API.
let new_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path(format!("{}/appended.parquet", table.metadata().location()))
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(100))]))
.build()
.unwrap();
let tx = Transaction::new(&table);
let action = tx.fast_append().add_data_files(vec![new_file]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
SnapshotRef::new(snapshot.clone())
} else {
unreachable!("first update of a fast append should be AddSnapshot")
};
let manifest_list = table
.manifest_list_reader(&new_snapshot)
.load()
.await
.unwrap();
assert!(
manifest_list
.entries()
.iter()
.any(|m| m.manifest_path == delete_manifest_path),
"delete-only manifest {delete_manifest_path} was dropped from the new snapshot's \
manifest list; the files it removed would reappear as live data"
);
}
/// 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());
table
.manifest_reader()
.read(&manifest_list.entries()[0])
.await
.unwrap()
.entries()
.iter()
.map(|entry| entry.data_file().clone())
.collect()
}
#[tokio::test]
async fn test_fast_append_writes_encrypted_manifest() {
let table = make_encrypted_table().await;
assert!(
table.encryption_manager().is_some(),
"fixture table should have an EncryptionManager"
);
let new_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("memory:///table/data/00000.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.partition(Struct::empty())
.record_count(100)
.file_size_in_bytes(4096)
.partition_spec_id(table.metadata().default_partition_spec_id())
.build()
.unwrap();
let tx = Transaction::new(&table);
let action = tx.fast_append().add_data_files(vec![new_file]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let new_snapshot: SnapshotRef = updates
.iter()
.find_map(|u| match u {
TableUpdate::AddSnapshot { snapshot } => Some(SnapshotRef::new(snapshot.clone())),
_ => None,
})
.expect("a fast append should emit an AddSnapshot update");
let manifest_list_key_metadata = table
.encryption_manager()
.unwrap()
.decrypt_manifest_list_key_metadata(new_snapshot.encryption_key_id().unwrap())
.await
.unwrap();
let manifest_list_size = table
.file_io()
.new_input(new_snapshot.manifest_list())
.unwrap()
.metadata()
.await
.unwrap()
.size;
assert_eq!(
manifest_list_key_metadata.file_length(),
Some(manifest_list_size)
);
let manifest_list = table
.manifest_list_reader(&new_snapshot)
.load()
.await
.unwrap();
let manifest_file = manifest_list
.entries()
.iter()
.find(|m| m.added_files_count.unwrap_or(0) > 0)
.expect("new snapshot should carry the appended data manifest");
// The manifest list entry must carry decodable key metadata.
let key_metadata_bytes = manifest_file
.key_metadata
.as_ref()
.expect("encrypted manifest must record key metadata");
let key_metadata = StandardKeyMetadata::decode(key_metadata_bytes)
.expect("recorded key metadata must decode as StandardKeyMetadata");
let manifest_size = table
.file_io()
.new_input(&manifest_file.manifest_path)
.unwrap()
.metadata()
.await
.unwrap()
.size;
assert_eq!(key_metadata.file_length(), Some(manifest_size));
assert_eq!(manifest_file.manifest_length, manifest_size as i64);
// The reader self-decrypts using the recorded key metadata and must
// recover the entry we appended. Because the read goes through the
// decryption path, this succeeding also proves the bytes
// on disk were genuinely encrypted (not silently written as plaintext).
let manifest = table.manifest_reader().read(manifest_file).await.unwrap();
assert_eq!(manifest.entries().len(), 1);
assert_eq!(
manifest.entries()[0].data_file().file_path(),
"memory:///table/data/00000.parquet"
);
}
#[tokio::test]
async fn test_empty_data_append_action() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let action = tx.fast_append().add_data_files(vec![]);
assert!(Arc::new(action).commit(&table).await.is_err());
}
/// A `fast_append` must write the manifest list and the manifest
/// files under the `write.metadata.path` prefix when configured,
/// rather than the default `<location>/metadata` directory.
#[tokio::test]
async fn test_fast_append_honors_write_metadata_path() {
let base = make_v2_minimal_table();
let metadata_root = format!("{}/custom-meta", base.metadata().location());
let metadata = base
.metadata()
.clone()
.into_builder(None)
.set_properties(HashMap::from([(
"write.metadata.path".to_string(),
metadata_root.clone(),
)]))
.unwrap()
.build()
.unwrap()
.metadata;
let table = base.with_metadata(Arc::new(metadata));
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path(format!("{}/data/1.parquet", table.metadata().location()))
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(300))]))
.build()
.unwrap();
let tx = Transaction::new(&table);
let action = tx.fast_append().add_data_files(vec![data_file]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
SnapshotRef::new(snapshot.clone())
} else {
unreachable!("first update of a fast append should be AddSnapshot")
};
let prefix = format!("{metadata_root}/");
// Manifest list
assert!(
new_snapshot.manifest_list().starts_with(prefix.as_str()),
"manifest list {} not under configured write.metadata.path {metadata_root}",
new_snapshot.manifest_list()
);
// Manifest files
let manifest_list = table
.manifest_list_reader(&new_snapshot)
.load()
.await
.unwrap();
assert!(
!manifest_list.entries().is_empty(),
"expected at least one manifest entry"
);
for entry in manifest_list.entries() {
assert!(
entry.manifest_path.starts_with(prefix.as_str()),
"manifest {} not under configured write.metadata.path {metadata_root}",
entry.manifest_path
);
}
}
#[tokio::test]
async fn test_set_snapshot_properties() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let mut snapshot_properties = HashMap::new();
snapshot_properties.insert("key".to_string(), "val".to_string());
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("test/1.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(300))]))
.build()
.unwrap();
let action = tx
.fast_append()
.set_snapshot_properties(snapshot_properties)
.add_data_files(vec![data_file]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
// Check customized properties is contained in snapshot summary properties.
let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
snapshot
} else {
unreachable!()
};
assert_eq!(
new_snapshot
.summary()
.additional_properties
.get("key")
.unwrap(),
"val"
);
}
/// See `testdata/manifests_lists/README.md`.
const FIXTURE_MASTER_KEY_ID: &str = "master-1";
const FIXTURE_MASTER_KEY_BYTES: [u8; 16] = [
0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e,
0x0f,
];
async fn make_v3_encrypted_table() -> Table {
let json = fs::read_to_string(format!(
"{}/testdata/table_metadata/TableMetadataV3ValidEncryption.json",
env!("CARGO_MANIFEST_DIR")
))
.unwrap();
let metadata = serde_json::from_str::<TableMetadata>(&json).unwrap();
let kms = MemoryKeyManagementClient::new();
kms.add_master_key_bytes(
FIXTURE_MASTER_KEY_ID,
SensitiveBytes::new(FIXTURE_MASTER_KEY_BYTES),
)
.unwrap();
let file_io = FileIO::new_with_memory();
let manifest_list_bytes = fs::read(format!(
"{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
env!("CARGO_MANIFEST_DIR")
))
.unwrap();
let parent_manifest_list = metadata.current_snapshot().unwrap().manifest_list();
file_io
.new_output(parent_manifest_list)
.unwrap()
.write(manifest_list_bytes.into())
.await
.unwrap();
Table::builder()
.metadata(metadata)
.metadata_location("memory:///table/metadata/v1.json")
.identifier(TableIdent::from_strs(["ns1", "enc"]).unwrap())
.file_io(file_io)
.kms_client(Arc::new(kms))
.runtime(test_runtime())
.build()
.unwrap()
}
#[tokio::test]
async fn test_commit_with_encryption_adds_keys_and_records_snapshot_key_id() {
let table = make_v3_encrypted_table().await;
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("test/1.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::empty())
.build()
.unwrap();
let tx = Transaction::new(&table);
let action = tx.fast_append().add_data_files(vec![data_file]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let added_key_ids: Vec<String> = updates
.iter()
.filter_map(|u| match u {
TableUpdate::AddEncryptionKey { encryption_key } => {
Some(encryption_key.key_id().to_string())
}
_ => None,
})
.collect();
assert!(!added_key_ids.is_empty(), "got {updates:?}");
// Encryption keys are added before the snapshot, so it isn't updates[0] here.
let new_snapshot = updates
.iter()
.find_map(|u| match u {
TableUpdate::AddSnapshot { snapshot } => Some(snapshot),
_ => None,
})
.expect("commit should add a snapshot");
let snapshot_key_id = new_snapshot
.encryption_key_id()
.expect("encrypted snapshot should record its manifest-list key id");
assert!(
added_key_ids.iter().any(|id| id == snapshot_key_id),
"snapshot key id {snapshot_key_id} not in added keys {added_key_ids:?}"
);
let new_snapshot_ref: SnapshotRef = Arc::new(new_snapshot.clone());
let manifest_list = table
.manifest_list_reader(&new_snapshot_ref)
.load()
.await
.expect("newly written encrypted manifest list should decrypt and parse");
assert_eq!(
manifest_list.entries().len(),
1,
"append should record exactly the one new data manifest"
);
}
#[tokio::test]
async fn test_snapshot_properties_cannot_override_computed_metrics() {
// A user-supplied snapshot property must not shadow a computed metric key
// such as `added-data-files`. Matching iceberg-java, the computed value
// wins, so the summary reflects the real count and a bad value can neither
// corrupt the summary nor panic total computation (see #2184-adjacent fix).
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let mut snapshot_properties = HashMap::new();
// Both a benign-but-wrong value and a non-integer value collide with
// computed metric keys; neither should reach the final summary.
snapshot_properties.insert("added-data-files".to_string(), "9999".to_string());
snapshot_properties.insert("added-records".to_string(), "not-a-number".to_string());
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("test/1.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(300))]))
.build()
.unwrap();
let action = tx
.fast_append()
.set_snapshot_properties(snapshot_properties)
.add_data_files(vec![data_file]);
// Must not panic during total computation.
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
snapshot
} else {
unreachable!()
};
let props = &new_snapshot.summary().additional_properties;
// Computed metric wins over the user's colliding values.
assert_eq!(
props.get("added-data-files").unwrap(),
"1",
"computed added-data-files must override the user-supplied value"
);
assert_eq!(
props.get("added-records").unwrap(),
"1",
"computed added-records must override the user-supplied non-integer value"
);
}
#[tokio::test]
async fn test_append_snapshot_properties() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let mut snapshot_properties = HashMap::new();
snapshot_properties.insert("key".to_string(), "val".to_string());
let action = tx
.fast_append()
.set_snapshot_properties(snapshot_properties);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
// Check customized properties is contained in snapshot summary properties.
let new_snapshot = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
snapshot
} else {
unreachable!()
};
assert_eq!(
new_snapshot
.summary()
.additional_properties
.get("key")
.unwrap(),
"val"
);
}
#[tokio::test]
async fn test_fast_append_file_with_incompatible_partition_value() {
let table = make_v2_minimal_table();
let tx = Transaction::new(&table);
let action = tx.fast_append();
// check add data file with incompatible partition value
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("test/3.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::string("test"))]))
.build()
.unwrap();
let action = action.add_data_files(vec![data_file.clone()]);
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();
let tx = Transaction::new(&table);
let action = tx.fast_append();
let data_file = DataFileBuilder::default()
.content(DataContentType::Data)
.file_path("test/3.parquet".to_string())
.file_format(DataFileFormat::Parquet)
.file_size_in_bytes(100)
.record_count(1)
.partition_spec_id(table.metadata().default_partition_spec_id())
.partition(Struct::from_iter([Some(Literal::long(300))]))
.build()
.unwrap();
let action = action.add_data_files(vec![data_file.clone()]);
let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
let updates = action_commit.take_updates();
let requirements = action_commit.take_requirements();
// check updates and requirements
assert!(
matches!((&updates[0],&updates[1]), (TableUpdate::AddSnapshot { snapshot },TableUpdate::SetSnapshotRef { reference,ref_name }) if snapshot.snapshot_id() == reference.snapshot_id && ref_name == MAIN_BRANCH)
);
assert_eq!(
vec![
TableRequirement::UuidMatch {
uuid: table.metadata().uuid()
},
TableRequirement::RefSnapshotIdMatch {
r#ref: MAIN_BRANCH.to_string(),
snapshot_id: table.metadata().current_snapshot_id
}
],
requirements
);
// check manifest list
let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] {
SnapshotRef::new(snapshot.clone())
} else {
unreachable!()
};
let manifest_list = table
.manifest_list_reader(&new_snapshot)
.load()
.await
.unwrap();
assert_eq!(1, manifest_list.entries().len());
assert_eq!(
manifest_list.entries()[0].sequence_number,
new_snapshot.sequence_number()
);
// check manifest
let manifest = table
.manifest_reader()
.read(&manifest_list.entries()[0])
.await
.unwrap();
assert_eq!(1, manifest.entries().len());
assert_eq!(
new_snapshot.sequence_number(),
manifest.entries()[0]
.sequence_number()
.expect("Inherit sequence number by load manifest")
);
assert_eq!(
new_snapshot.snapshot_id(),
manifest.entries()[0].snapshot_id().unwrap()
);
assert_eq!(data_file, *manifest.entries()[0].data_file());
}
}