- add parse_property to refine parse property logic

- rename FastAppendOperation to AppendOperation
- use iterable input for pack to be more memory efficient

Author: ZENOTME

crates/iceberg/src/transaction/append.rs

@@ -103,14 +103,14 @@ impl TransactionAction for FastAppendAction {
         }
 
         snapshot_producer
-            .commit(FastAppendOperation, DefaultManifestProcess)
+            .commit(AppendOperation, DefaultManifestProcess)
             .await
     }
 }
 
-pub(crate) struct FastAppendOperation;
+pub(crate) struct AppendOperation;
 
-impl SnapshotProduceOperation for FastAppendOperation {
+impl SnapshotProduceOperation for AppendOperation {
     fn operation(&self) -> Operation {
         Operation::Append
     }

crates/iceberg/src/transaction/merge_append.rs

@@ -26,10 +26,11 @@ use crate::Result;
 use crate::io::FileIO;
 use crate::spec::{DataFile, ManifestContentType, ManifestFile, ManifestStatus, ManifestWriter};
 use crate::table::Table;
-use crate::transaction::append::FastAppendOperation;
+use crate::transaction::append::AppendOperation;
 use crate::transaction::snapshot::{DefaultManifestProcess, ManifestProcess, SnapshotProducer};
 use crate::transaction::{ActionCommit, TransactionAction};
 use crate::utils::bin::ListPacker;
+use crate::utils::parse_property;
 
 /// MergeAppendAction is a transaction action similar to fast append except that it will merge manifests
 /// based on the target size.
@@ -57,24 +58,21 @@ const MANIFEST_MERGE_ENABLED_DEFAULT: bool = false;
 
 impl MergeAppendAction {
     pub(crate) fn new(table: &Table) -> Result<Self> {
-        let target_size_bytes: u32 = 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 = table
-            .metadata()
-            .properties()
-            .get(MANIFEST_MIN_MERGE_COUNT)
-            .and_then(|s| s.parse().ok())
-            .unwrap_or(MANIFEST_MIN_MERGE_COUNT_DEFAULT);
-        let merge_enabled = table
-            .metadata()
-            .properties()
-            .get(MANIFEST_MERGE_ENABLED)
-            .and_then(|s| s.parse().ok())
-            .unwrap_or(MANIFEST_MERGE_ENABLED_DEFAULT);
+        let target_size_bytes: u32 = parse_property(
+            table.metadata().properties(),
+            MANIFEST_TARGET_SIZE_BYTES,
+            MANIFEST_TARGET_SIZE_BYTES_DEFAULT,
+        )?;
+        let min_count_to_merge: u32 = parse_property(
+            table.metadata().properties(),
+            MANIFEST_MIN_MERGE_COUNT,
+            MANIFEST_MIN_MERGE_COUNT_DEFAULT,
+        )?;
+        let merge_enabled = parse_property(
+            table.metadata().properties(),
+            MANIFEST_MERGE_ENABLED,
+            MANIFEST_MERGE_ENABLED_DEFAULT,
+        )?;
         Ok(Self {
             check_duplicate: true,
             target_size_bytes,
@@ -140,14 +138,14 @@ impl TransactionAction for MergeAppendAction {
 
         if self.merge_enabled {
             snapshot_producer
-                .commit(FastAppendOperation, MergeManifsetProcess {
+                .commit(AppendOperation, MergeManifsetProcess {
                     target_size_bytes: self.target_size_bytes,
                     min_count_to_merge: self.min_count_to_merge,
                 })
                 .await
         } else {
             snapshot_producer
-                .commit(FastAppendOperation, DefaultManifestProcess)
+                .commit(AppendOperation, DefaultManifestProcess)
                 .await
         }
     }
@@ -228,7 +226,6 @@ impl MergeManifestManager {
             packer.pack(group_manifests, |manifest| manifest.manifest_length as u32);
 
         let manifest_merge_futures = manifest_bins
-            .into_iter()
             .map(|manifest_bin| {
                 if manifest_bin.len() == 1 {
                     Ok(Box::pin(async { Ok(manifest_bin) })

crates/iceberg/src/utils.rs

@@ -15,7 +15,11 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
 use std::num::NonZeroUsize;
+use std::str::FromStr;
+
+use crate::{Error, ErrorKind, Result};
 
 // Use a default value of 1 as the safest option.
 // See https://doc.rust-lang.org/std/thread/fn.available_parallelism.html#limitations
@@ -41,12 +45,31 @@ pub(crate) fn available_parallelism() -> NonZeroUsize {
     })
 }
 
+pub(crate) fn parse_property<T>(
+    properties: &HashMap<String, String>,
+    key: &str,
+    default_value: T,
+) -> Result<T>
+where
+    T: FromStr,
+    T::Err: std::error::Error + Send + Sync + 'static,
+{
+    match properties.get(key) {
+        Some(s) => s.parse::<T>().map_err(|e| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                format!("Invalid property value for key: {}", key),
+            )
+            .with_source(e)
+        }),
+        None => Ok(default_value),
+    }
+}
+
 pub mod bin {
     use std::iter::Iterator;
     use std::marker::PhantomData;
 
-    use itertools::Itertools;
-
     struct Bin<T> {
         bin_weight: u32,
         target_weight: u32,
@@ -91,8 +114,11 @@ pub mod bin {
             }
         }
 
-        pub fn pack<F>(&self, items: Vec<T>, weight_func: F) -> Vec<Vec<T>>
-        where F: Fn(&T) -> u32 {
+        pub fn pack<I, F>(&self, items: I, weight_func: F) -> impl Iterator<Item = Vec<T>>
+        where
+            I: IntoIterator<Item = T>,
+            F: Fn(&T) -> u32,
+        {
             let mut bins: Vec<Bin<T>> = vec![];
             for item in items {
                 let cur_weight = weight_func(&item);
@@ -106,7 +132,7 @@ pub mod bin {
                 addable_bin.add(item, cur_weight);
             }
 
-            bins.into_iter().map(|bin| bin.into_vec()).collect_vec()
+            bins.into_iter().map(|bin| bin.into_vec())
         }
     }
 
@@ -119,7 +145,7 @@ pub mod bin {
             let packer = ListPacker::new(10);
             let items = vec![3, 4, 5, 6, 2, 1];
 
-            let packed = packer.pack(items, |&x| x);
+            let packed: Vec<Vec<u32>> = packer.pack(items, |&x| x).collect();
 
             assert_eq!(packed.len(), 3);
             assert!(packed[0].iter().sum::<u32>() == 10);
@@ -155,7 +181,7 @@ pub mod bin {
                 },
             ];
 
-            let packed = packer.pack(items, |item| item.size);
+            let packed: Vec<Vec<Item>> = packer.pack(items, |item| item.size).collect();
 
             assert_eq!(packed.len(), 2);
             assert!(packed[0].iter().map(|x| x.size).sum::<u32>() <= 15);
@@ -167,7 +193,7 @@ pub mod bin {
             let packer = ListPacker::new(10);
             let items = vec![15, 5, 3];
 
-            let packed = packer.pack(items, |&x| x);
+            let packed: Vec<Vec<u32>> = packer.pack(items, |&x| x).collect();
 
             assert_eq!(packed.len(), 2);
             assert!(packed[0].contains(&15));
@@ -179,7 +205,7 @@ pub mod bin {
             let packer = ListPacker::new(10);
             let items: Vec<u32> = vec![];
 
-            let packed = packer.pack(items, |&x| x);
+            let packed: Vec<Vec<u32>> = packer.pack(items, |&x| x).collect();
 
             assert_eq!(packed.len(), 0);
         }