support merge append (#29)

Co-authored-by: ZENOTME <[email protected]>
Signed-off-by: xxchan <[email protected]>

Author: ZENOTME

crates/iceberg/src/spec/manifest/entry.rs

@@ -147,6 +147,12 @@ impl ManifestEntry {
     pub fn data_file(&self) -> &DataFile {
         &self.data_file
     }
+
+    /// File sequence number indicating when the file was added. Inherited when null and status is 1 (added).
+    #[inline]
+    pub fn file_sequence_number(&self) -> Option<i64> {
+        self.file_sequence_number
+    }
 }
 
 /// Used to track additions and deletions in ManifestEntry.

crates/iceberg/src/transaction/append.rs

@@ -24,12 +24,22 @@ use uuid::Uuid;
 use crate::error::Result;
 use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation};
 use crate::transaction::snapshot::{
-    DefaultManifestProcess, SnapshotProduceAction, SnapshotProduceOperation,
+    DefaultManifestProcess, MergeManifestProcess, SnapshotProduceAction, SnapshotProduceOperation,
 };
 use crate::transaction::Transaction;
 use crate::writer::file_writer::ParquetWriter;
 use crate::{Error, ErrorKind};
 
+/// Target size of manifest file when merging manifests.
+pub const MANIFEST_TARGET_SIZE_BYTES: &str = "commit.manifest.target-size-bytes";
+const MANIFEST_TARGET_SIZE_BYTES_DEFAULT: u32 = 8 * 1024 * 1024; // 8 MB
+/// Minimum number of manifests to merge.
+pub const MANIFEST_MIN_MERGE_COUNT: &str = "commit.manifest.min-count-to-merge";
+const MANIFEST_MIN_MERGE_COUNT_DEFAULT: u32 = 100;
+/// Whether allow to merge manifests.
+pub const MANIFEST_MERGE_ENABLED: &str = "commit.manifest-merge.enabled";
+const MANIFEST_MERGE_ENABLED_DEFAULT: bool = false;
+
 /// FastAppendAction is a transaction action for fast append data files to the table.
 pub struct FastAppendAction<'a> {
     snapshot_produce_action: SnapshotProduceAction<'a>,
@@ -204,6 +214,84 @@ impl SnapshotProduceOperation for FastAppendOperation {
     }
 }
 
+/// MergeAppendAction is a transaction action similar to fast append except that it will merge manifests
+/// based on the target size.
+pub struct MergeAppendAction<'a> {
+    snapshot_produce_action: SnapshotProduceAction<'a>,
+    target_size_bytes: u32,
+    min_count_to_merge: u32,
+    merge_enabled: bool,
+}
+
+impl<'a> MergeAppendAction<'a> {
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn new(
+        tx: Transaction<'a>,
+        snapshot_id: i64,
+        commit_uuid: Uuid,
+        key_metadata: Vec<u8>,
+        snapshot_properties: HashMap<String, String>,
+    ) -> Result<Self> {
+        let target_size_bytes: u32 = tx
+            .current_table
+            .metadata()
+            .properties()
+            .get(MANIFEST_TARGET_SIZE_BYTES)
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(MANIFEST_TARGET_SIZE_BYTES_DEFAULT);
+        let min_count_to_merge: u32 = tx
+            .current_table
+            .metadata()
+            .properties()
+            .get(MANIFEST_MIN_MERGE_COUNT)
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(MANIFEST_MIN_MERGE_COUNT_DEFAULT);
+        let merge_enabled = tx
+            .current_table
+            .metadata()
+            .properties()
+            .get(MANIFEST_MERGE_ENABLED)
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(MANIFEST_MERGE_ENABLED_DEFAULT);
+        Ok(Self {
+            snapshot_produce_action: SnapshotProduceAction::new(
+                tx,
+                snapshot_id,
+                key_metadata,
+                commit_uuid,
+                snapshot_properties,
+            )?,
+            target_size_bytes,
+            min_count_to_merge,
+            merge_enabled,
+        })
+    }
+
+    /// Add data files to the snapshot.
+    pub fn add_data_files(
+        &mut self,
+        data_files: impl IntoIterator<Item = DataFile>,
+    ) -> Result<&mut Self> {
+        self.snapshot_produce_action.add_data_files(data_files)?;
+        Ok(self)
+    }
+
+    /// Finished building the action and apply it to the transaction.
+    pub async fn apply(self) -> Result<Transaction<'a>> {
+        if self.merge_enabled {
+            let process =
+                MergeManifestProcess::new(self.target_size_bytes, self.min_count_to_merge);
+            self.snapshot_produce_action
+                .apply(FastAppendOperation, process)
+                .await
+        } else {
+            self.snapshot_produce_action
+                .apply(FastAppendOperation, DefaultManifestProcess)
+                .await
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::scan::tests::TableTestFixture;

crates/iceberg/src/transaction/mod.rs

@@ -18,7 +18,7 @@
 //! This module contains transaction api.
 
 mod append;
-pub mod remove_snapshots;
+mod remove_snapshots;
 mod snapshot;
 mod sort_order;
 
@@ -27,13 +27,14 @@ use std::collections::HashMap;
 use std::mem::discriminant;
 use std::sync::Arc;
 
+pub use append::{MANIFEST_MERGE_ENABLED, MANIFEST_MIN_MERGE_COUNT, MANIFEST_TARGET_SIZE_BYTES};
 use remove_snapshots::RemoveSnapshotAction;
 use uuid::Uuid;
 
 use crate::error::Result;
 use crate::spec::FormatVersion;
 use crate::table::Table;
-use crate::transaction::append::FastAppendAction;
+use crate::transaction::append::{FastAppendAction, MergeAppendAction};
 use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::TableUpdate::UpgradeFormatVersion;
 use crate::{Catalog, Error, ErrorKind, TableCommit, TableRequirement, TableUpdate};
@@ -184,6 +185,22 @@ impl<'a> Transaction<'a> {
         )
     }
 
+    /// Creates a merge append action.
+    pub fn merge_append(
+        self,
+        commit_uuid: Option<Uuid>,
+        key_metadata: Vec<u8>,
+    ) -> Result<MergeAppendAction<'a>> {
+        let snapshot_id = self.generate_unique_snapshot_id();
+        MergeAppendAction::new(
+            self,
+            snapshot_id,
+            commit_uuid.unwrap_or_else(Uuid::now_v7),
+            key_metadata,
+            HashMap::new(),
+        )
+    }
+
     /// Creates replace sort order action.
     pub fn replace_sort_order(self) -> ReplaceSortOrderAction<'a> {
         ReplaceSortOrderAction {