iceberg/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 async_trait::async_trait;
use uuid::Uuid;

use crate::error::{Result, invalid_data};
use crate::spec::{DataContentType, DataFile};
use crate::table::Table;
use crate::transaction::action::CommitStatus;
use crate::transaction::snapshot::simple::{AppendIntent, SimpleSnapshotProducer};
use crate::transaction::snapshot::validate_files_not_live;
use crate::transaction::{ActionCommit, TransactionAction};
use crate::{Error, ErrorKind};

/// FastAppendAction is a transaction action for fast append data files to the table.
#[derive(Clone)]
pub struct FastAppendAction {
    check_duplicate: bool,
    /// Overrides the commit UUID of the producer created in `new_state`.
    commit_uuid: Option<Uuid>,
    snapshot_properties: HashMap<String, String>,
    /// Free of duplicate paths; see [`Self::add_data_files`].
    added_data_files: Vec<DataFile>,
    /// Paths already in `added_data_files`.
    added_paths: HashSet<String>,
}

impl FastAppendAction {
    pub(crate) fn new() -> Self {
        Self {
            check_duplicate: true,
            commit_uuid: None,
            snapshot_properties: HashMap::default(),
            added_data_files: vec![],
            added_paths: HashSet::new(),
        }
    }

    /// 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.
    ///
    /// A path added more than once keeps its first occurrence, so a manifest
    /// never references the same file twice. This is independent of the
    /// `check_duplicate`-gated check against files already in the table.
    pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self {
        for data_file in data_files {
            if self.added_paths.insert(data_file.file_path.clone()) {
                self.added_data_files.push(data_file);
            }
        }
        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
    }

    /// Checks the parts of this action that do not depend on the table.
    fn validate(&self) -> Result<()> {
        // 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() {
            return Err(Error::new(
                ErrorKind::PreconditionFailed,
                "No added data files or added snapshot properties found when write a manifest file",
            ));
        }
        if let Some(data_file) = self
            .added_data_files
            .iter()
            .find(|f| f.content_type() != DataContentType::Data)
        {
            return Err(invalid_data!(
                "Only data content type is allowed for fast append, got {:?} for {}",
                data_file.content_type(),
                data_file.file_path()
            ));
        }
        Ok(())
    }
}

#[async_trait]
impl TransactionAction for FastAppendAction {
    type State = SimpleSnapshotProducer;

    fn new_state(&self) -> SimpleSnapshotProducer {
        SimpleSnapshotProducer::new(AppendIntent {
            commit_uuid: self.commit_uuid,
            snapshot_properties: self.snapshot_properties.clone(),
            data_files: self.added_data_files.clone(),
        })
    }

    async fn commit(
        &self,
        state: &mut SimpleSnapshotProducer,
        table: &Table,
    ) -> Result<ActionCommit> {
        self.validate()?;
        if self.check_duplicate {
            validate_files_not_live(table, &self.added_data_files).await?;
        }
        state.apply(table).await
    }

    async fn cleanup(self, _state: SimpleSnapshotProducer, _table: &Table, _status: CommitStatus) {
        // TODO(#3417): remove the producer's uncommitted artifacts once terminal
        // cleanup is wired up.
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::fs;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, Ordering};

    use minijinja::{AutoEscape, Environment, Value, context};
    use tempfile::TempDir;
    use uuid::Uuid;

    use crate::catalog::MockCatalog;
    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::{commit_with_fresh_state, make_encrypted_table, test_runtime};
    use crate::transaction::tests::{
        make_v2_minimal_table, make_v3_minimal_table, with_retry_properties,
    };
    use crate::transaction::{ApplyTransactionAction, FastAppendAction, Transaction};
    use crate::{ErrorKind, TableCommit, 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 = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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_fast_append_rejects_delete_files() {
        let table = make_v2_minimal_table();
        let delete_file = DataFileBuilder::default()
            .content(DataContentType::EqualityDeletes)
            .file_path("test/1-deletes.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))]))
            .equality_ids(Some(vec![1]))
            .build()
            .unwrap();

        let tx = Transaction::new(&table);
        let action = tx.fast_append().add_data_files(vec![delete_file]);
        let err = commit_with_fresh_state(action, &table)
            .await
            .err()
            .expect("delete files must be rejected");
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
    }

    #[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!(commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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_snapshot_summary_records_writer_identity() {
        let table = make_v2_minimal_table();
        let tx = Transaction::new(&table);

        let snapshot_properties = HashMap::from([
            ("key".to_string(), "val".to_string()),
            ("iceberg-version".to_string(), "caller".to_string()),
        ]);

        let action = tx
            .fast_append()
            .set_snapshot_properties(snapshot_properties);
        let mut action_commit = commit_with_fresh_state(action, &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;
        let version = env!("CARGO_PKG_VERSION");

        assert_eq!(
            props.get("iceberg-version").unwrap(),
            &format!("Apache Iceberg Rust {version}")
        );
        assert_eq!(props.get("key").unwrap(), "val");
    }

    #[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 = commit_with_fresh_state(action, &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!(commit_with_fresh_state(action, &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, across two `add_data_files` calls: 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)])
            .add_data_files(vec![make_file(300, 30)]);
        let mut action_commit = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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 = commit_with_fresh_state(action, &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());
    }

    /// Two writers append concurrently. Writer A starts from S0, but writer B
    /// commits S1 first, so A's first attempt fails the `RefSnapshotIdMatch`
    /// requirement and A rebases onto S1 to commit S2.
    ///
    /// The producer state survives the retry: A's manifest is written once and
    /// reused, while parent, sequence number, manifest list and summary are
    /// recomputed against S1 (see #3418).
    #[tokio::test]
    async fn test_fast_append_retry_rebases_onto_concurrent_commit() {
        let (s1, s2) = rebase_onto_concurrent_commit(make_v2_minimal_table()).await;
        assert_eq!(s1.first_row_id(), None);
        assert_eq!(s2.first_row_id(), None);
    }

    /// On a v3 table the retry must also move the row range: S2's first row id
    /// is S1's `next_row_id`, not S0's.
    #[tokio::test]
    async fn test_fast_append_retry_rebases_row_range_onto_concurrent_commit() {
        let (s1, s2) = rebase_onto_concurrent_commit(make_v3_minimal_table()).await;

        // B's single-row file takes row 0 in S1, so A's file starts at row 1.
        assert_eq!(s1.row_range(), Some((0, 1)));
        assert_eq!(s2.row_range(), Some((1, 1)));
    }

    /// Runs the concurrent-append rebase scenario on `s0` and returns the
    /// snapshots committed by writer B (S1) and writer A (S2).
    async fn rebase_onto_concurrent_commit(s0: Table) -> (SnapshotRef, SnapshotRef) {
        // `TableCommit::apply` derives the next metadata location from the
        // current one, so it needs a well-formed name.
        let s0 = with_retry_properties(s0, "3").with_metadata_location(
            "memory:///table/metadata/00000-9c12d441-03fe-4693-9a96-a0705ddf69c1.metadata.json"
                .to_string(),
        );
        let data_file = |path: &str| {
            DataFileBuilder::default()
                .content(DataContentType::Data)
                .file_path(path.to_string())
                .file_format(DataFileFormat::Parquet)
                .file_size_in_bytes(100)
                .record_count(1)
                .partition_spec_id(s0.metadata().default_partition_spec_id())
                .partition(Struct::from_iter([Some(Literal::long(300))]))
                .build()
                .unwrap()
        };

        // Writer B commits S1 on top of S0.
        let mut b_commit = commit_with_fresh_state(
            FastAppendAction::new().add_data_files(vec![data_file("test/b.parquet")]),
            &s0,
        )
        .await
        .unwrap();
        let s1 = TableCommit::builder()
            .ident(s0.identifier().clone())
            .updates(b_commit.take_updates())
            .requirements(b_commit.take_requirements())
            .build()
            .apply(s0.clone())
            .unwrap();
        let s1_snapshot = s1.metadata().current_snapshot().unwrap().clone();

        // The catalog serves S0 to A's first attempt and S1 afterwards; every
        // update is validated against S1, so the first attempt's
        // `RefSnapshotIdMatch { snapshot_id: None }` fails with a real conflict.
        let mut mock_catalog = MockCatalog::new();
        let loads = AtomicU32::new(0);
        let (load_s0, load_s1) = (s0.clone(), s1.clone());
        mock_catalog
            .expect_load_table()
            .times(2)
            .returning_st(move |_| {
                let table = if loads.fetch_add(1, Ordering::SeqCst) == 0 {
                    load_s0.clone()
                } else {
                    load_s1.clone()
                };
                Box::pin(async move { Ok(table) })
            });
        let apply_s1 = s1.clone();
        mock_catalog
            .expect_update_table()
            .times(2)
            .returning_st(move |commit| {
                let s1 = apply_s1.clone();
                Box::pin(async move { commit.apply(s1) })
            });

        // Writer A starts from S0.
        let a_uuid = Uuid::now_v7();
        let tx = Transaction::new(&s0);
        let tx = tx
            .fast_append()
            .set_commit_uuid(a_uuid)
            .add_data_files(vec![data_file("test/a.parquet")])
            .apply(tx)
            .unwrap();
        let s2 = tx.commit(&mock_catalog).await.unwrap();
        let s2_snapshot = s2.metadata().current_snapshot().unwrap().clone();

        // S2 is rebased onto S1, not S0.
        assert_eq!(
            s2_snapshot.parent_snapshot_id(),
            Some(s1_snapshot.snapshot_id())
        );
        assert_eq!(
            s2_snapshot.sequence_number(),
            s1_snapshot.sequence_number() + 1
        );
        // The committed manifest list is A's second attempt.
        assert!(
            s2_snapshot
                .manifest_list()
                .ends_with(&format!("-1-{a_uuid}.avro")),
            "unexpected manifest list path {}",
            s2_snapshot.manifest_list()
        );

        // S2 carries B's manifest forward and adds A's. A's manifest is the one
        // written during the first attempt (counter 0), reused after rebase.
        let manifest_list = s2.manifest_list_reader(&s2_snapshot).load().await.unwrap();
        assert_eq!(manifest_list.entries().len(), 2);
        let b_manifest = manifest_list
            .entries()
            .iter()
            .find(|m| m.added_snapshot_id == s1_snapshot.snapshot_id())
            .expect("S2 should carry B's manifest forward");
        let a_manifest = manifest_list
            .entries()
            .iter()
            .find(|m| m.added_snapshot_id == s2_snapshot.snapshot_id())
            .expect("S2 should add A's manifest");
        assert!(
            a_manifest
                .manifest_path
                .ends_with(&format!("{a_uuid}-m0.avro")),
            "unexpected manifest path {}",
            a_manifest.manifest_path
        );
        assert_eq!(a_manifest.sequence_number, s2_snapshot.sequence_number());
        assert_eq!(b_manifest.sequence_number, s1_snapshot.sequence_number());

        // Summary totals are computed against S1, not S0.
        let summary = &s2_snapshot.summary().additional_properties;
        assert_eq!(summary.get("added-data-files").unwrap(), "1");
        assert_eq!(summary.get("total-data-files").unwrap(), "2");
        assert_eq!(summary.get("total-records").unwrap(), "2");

        (s1_snapshot, s2_snapshot)
    }
}