Improve append

datafusion_iceberg/src/table.rs

@@ -920,7 +920,13 @@ mod tests {
             "INSERT INTO orders (id, customer_id, product_id, date, amount) VALUES 
                 (7, 1, 3, '2020-01-03', 1),
                 (8, 2, 1, '2020-01-03', 2),
-                (9, 2, 2, '2020-01-03', 1);",
+                (9, 2, 2, '2020-01-03', 1),
+                (10, 1, 2, '2020-01-04', 3),
+                (11, 3, 1, '2020-01-04', 2),
+                (12, 2, 3, '2020-01-04', 1),
+                (13, 1, 1, '2020-01-05', 4),
+                (14, 3, 2, '2020-01-05', 2),
+                (15, 2, 3, '2020-01-05', 3);",
         )
         .await
         .expect("Failed to create query plan for insert")
@@ -952,8 +958,9 @@ mod tests {
                 );
                 for (product_id, amount) in product_ids.iter().zip(amounts) {
                     match product_id.unwrap() {
-                        1 => assert_eq!(amount.unwrap(), 3),
-                        2 | 3 => assert_eq!(amount.unwrap(), 1),
+                        1 => assert_eq!(amount.unwrap(), 7),
+                        2 => assert_eq!(amount.unwrap(), 4),
+                        3 => assert_eq!(amount.unwrap(), 1),
                         _ => panic!("Unexpected order id"),
                     }
                 }

iceberg-file-catalog/src/lib.rs

@@ -113,7 +113,7 @@ impl Catalog for FileCatalog {
             .map_err(IcebergError::from)
             .map_ok(|x| {
                 let path = x.location.as_ref();
-                self.identifier(&path)
+                self.identifier(path)
             })
             .try_collect()
             .await
@@ -501,15 +501,15 @@ impl Catalog for FileCatalog {
 
 impl FileCatalog {
     fn namespace_path(&self, namespace: &str) -> String {
-        self.path.as_str().trim_end_matches('/').to_owned() + "/" + &self.name + "/" + &namespace
+        self.path.as_str().trim_end_matches('/').to_owned() + "/" + &self.name + "/" + namespace
     }
 
     fn tabular_path(&self, namespace: &str, name: &str) -> String {
         self.path.as_str().trim_end_matches('/').to_owned()
             + "/"
             + &self.name
             + "/"
-            + &namespace
+            + namespace
             + "/"
             + name
     }
@@ -523,7 +523,7 @@ impl FileCatalog {
             .try_filter(|x| {
                 future::ready(
                     x.ends_with("metadata.json")
-                        && x.starts_with(&(path.clone() + "/v").trim_start_matches('/')),
+                        && x.starts_with((path.clone() + "/v").trim_start_matches('/')),
                 )
             })
             .try_collect()
@@ -551,9 +551,7 @@ impl FileCatalog {
         let parts = path
             .trim_start_matches(self.path.trim_start_matches('/'))
             .trim_start_matches('/')
-            .split('/')
-            .skip(1)
-            .next()
+            .split('/').nth(1)
             .ok_or(IcebergError::InvalidFormat("Namespace in path".to_owned()))?
             .to_owned();
         Namespace::try_new(&[parts]).map_err(IcebergError::from)

iceberg-rust-spec/src/spec/manifest.rs

@@ -189,8 +189,10 @@ pub struct ManifestEntry {
     status: Status,
     /// Snapshot id where the file was added, or deleted if status is 2.
     /// Inherited when null.
+    #[builder(setter(strip_option), default)]
     snapshot_id: Option<i64>,
     /// Sequence number when the file was added. Inherited when null.
+    #[builder(setter(strip_option), default)]
     sequence_number: Option<i64>,
     /// File path, partition tuple, metrics, …
     data_file: DataFile,

iceberg-rust-spec/src/spec/manifest_list.rs

@@ -49,7 +49,7 @@ impl<'a, 'metadata, R: Read> ManifestListReader<'a, 'metadata, R> {
             FormatVersion::V2 => manifest_list_schema_v2(),
         };
         Ok(Self {
-            reader: AvroReader::with_schema(&schema, reader)?
+            reader: AvroReader::with_schema(schema, reader)?
                 .zip(repeat(table_metadata))
                 .map(avro_value_to_manifest_file),
         })
@@ -773,15 +773,15 @@ mod tests {
             partitions: Some(vec![FieldSummary {
                 contains_null: true,
                 contains_nan: Some(false),
-                lower_bound: Some(Value::Date(1234)),
-                upper_bound: Some(Value::Date(76890)),
+                lower_bound: Some(Value::Int(1234)),
+                upper_bound: Some(Value::Int(76890)),
             }]),
             key_metadata: None,
         };
 
         let schema = manifest_list_schema_v2();
 
-        let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+        let mut writer = apache_avro::Writer::new(schema, Vec::new());
 
         writer.append_ser(manifest_file.clone()).unwrap();
 
@@ -853,15 +853,15 @@ mod tests {
             partitions: Some(vec![FieldSummary {
                 contains_null: true,
                 contains_nan: Some(false),
-                lower_bound: Some(Value::Date(1234)),
-                upper_bound: Some(Value::Date(76890)),
+                lower_bound: Some(Value::Int(1234)),
+                upper_bound: Some(Value::Int(76890)),
             }]),
             key_metadata: None,
         };
 
         let schema = manifest_list_schema_v1();
 
-        let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+        let mut writer = apache_avro::Writer::new(schema, Vec::new());
 
         writer.append_ser(manifest_file.clone()).unwrap();
 

iceberg-rust-spec/src/spec/partition.rs

@@ -183,10 +183,13 @@ impl PartitionSpec {
             .map(|field| {
                 schema
                     .get(field.source_id as usize)
-                    .map(|x| x.field_type.clone())
+                    .ok_or(Error::NotFound(
+                        "Partition field".to_owned(),
+                        field.name.clone(),
+                    ))
+                    .and_then(|x| x.field_type.clone().tranform(&field.transform))
             })
-            .collect::<Option<Vec<_>>>()
-            .ok_or(Error::InvalidFormat("partition spec".to_string()))
+            .collect::<Result<Vec<_>, Error>>()
     }
 }
 

iceberg-rust-spec/src/spec/table_metadata.rs

@@ -733,7 +733,7 @@ pub struct SnapshotLog {
     pub timestamp_ms: i64,
 }
 
-#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone)]
+#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone, Copy)]
 #[repr(u8)]
 /// Iceberg format version
 #[derive(Default)]

iceberg-rust-spec/src/spec/values.rs

@@ -6,11 +6,14 @@ use std::{
     any::Any,
     collections::{btree_map::Keys, BTreeMap, HashMap},
     fmt,
+    hash::{DefaultHasher, Hash, Hasher},
     io::Cursor,
+    ops::{Add, Sub},
     slice::Iter,
 };
 
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc};
+use itertools::Itertools;
 use ordered_float::OrderedFloat;
 use rust_decimal::Decimal;
 use serde::{
@@ -128,7 +131,7 @@ impl fmt::Display for Value {
 /// The partition struct stores the tuple of partition values for each file.
 /// Its type is derived from the partition fields of the partition spec used to write the manifest file.
 /// In v2, the partition struct’s field ids must match the ids from the partition spec.
-#[derive(Debug, Clone, Eq, Hash, PartialOrd, Ord)]
+#[derive(Debug, Clone, Eq, PartialOrd, Ord)]
 pub struct Struct {
     /// Vector to store the field values
     pub fields: Vec<Option<Value>>,
@@ -273,6 +276,15 @@ impl PartialEq for Struct {
     }
 }
 
+impl Hash for Struct {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        for key in self.keys().sorted() {
+            key.hash(state);
+            self.get(key).hash(state);
+        }
+    }
+}
+
 impl Value {
     /// Perform a partition transformation for the given value
     pub fn tranform(&self, transform: &Transform) -> Result<Value, Error> {
@@ -817,6 +829,95 @@ mod datetime {
     }
 }
 
+pub trait TryAdd: Sized {
+    fn try_add(&self, other: &Self) -> Result<Self, Error>;
+}
+pub trait TrySub: Sized {
+    fn try_sub(&self, other: &Self) -> Result<Self, Error>;
+}
+
+impl<T: Add<Output = T> + Copy> TryAdd for T {
+    fn try_add(&self, other: &Self) -> Result<Self, Error> {
+        Ok(*self + *other)
+    }
+}
+
+impl<T: Sub<Output = T> + Copy> TrySub for T {
+    fn try_sub(&self, other: &Self) -> Result<Self, Error> {
+        Ok(*self - *other)
+    }
+}
+
+impl TryAdd for Value {
+    fn try_add(&self, other: &Self) -> Result<Self, Error> {
+        match (self, other) {
+            (Value::Int(own), Value::Int(other)) => Ok(Value::Int(own + other)),
+            (Value::LongInt(own), Value::LongInt(other)) => Ok(Value::LongInt(own + other)),
+            (Value::Float(own), Value::Float(other)) => Ok(Value::Float(*own + *other)),
+            (Value::Double(own), Value::Double(other)) => Ok(Value::Double(*own + *other)),
+            (Value::Date(own), Value::Date(other)) => Ok(Value::Date(own + other)),
+            (Value::Time(own), Value::Time(other)) => Ok(Value::Time(own + other)),
+            (Value::Timestamp(own), Value::Timestamp(other)) => Ok(Value::Timestamp(own + other)),
+            (Value::TimestampTZ(own), Value::TimestampTZ(other)) => {
+                Ok(Value::TimestampTZ(own + other))
+            }
+            (x, y) => Err(Error::Type(
+                x.datatype().to_string(),
+                y.datatype().to_string(),
+            )),
+        }
+    }
+}
+
+impl TrySub for Value {
+    fn try_sub(&self, other: &Self) -> Result<Self, Error> {
+        match (self, other) {
+            (Value::Int(own), Value::Int(other)) => Ok(Value::Int(own - other)),
+            (Value::LongInt(own), Value::LongInt(other)) => Ok(Value::LongInt(own - other)),
+            (Value::Float(own), Value::Float(other)) => Ok(Value::Float(*own - *other)),
+            (Value::Double(own), Value::Double(other)) => Ok(Value::Double(*own - *other)),
+            (Value::Date(own), Value::Date(other)) => Ok(Value::Date(own - other)),
+            (Value::Time(own), Value::Time(other)) => Ok(Value::Time(own - other)),
+            (Value::Timestamp(own), Value::Timestamp(other)) => Ok(Value::Timestamp(own - other)),
+            (Value::TimestampTZ(own), Value::TimestampTZ(other)) => {
+                Ok(Value::TimestampTZ(own - other))
+            }
+            (Value::String(own), Value::String(other)) => {
+                Ok(Value::LongInt(sub_string(own, other) as i64))
+            }
+            (x, y) => Err(Error::Type(
+                x.datatype().to_string(),
+                y.datatype().to_string(),
+            )),
+        }
+    }
+}
+
+fn sub_string(left: &str, right: &str) -> u64 {
+    if let Some(distance) = left
+        .chars()
+        .zip(right.chars())
+        .take(256)
+        .skip_while(|(l, r)| l == r)
+        .try_fold(0, |acc, (l, r)| {
+            if let (Some(l), Some(r)) = (l.to_digit(36), r.to_digit(36)) {
+                Some(acc + (l - r).pow(2))
+            } else {
+                None
+            }
+        })
+    {
+        distance as u64
+    } else {
+        let mut hasher = DefaultHasher::new();
+        hasher.write(left.as_bytes());
+        let left = hasher.finish();
+        hasher.write(right.as_bytes());
+        let right = hasher.finish();
+        left - right
+    }
+}
+
 #[cfg(test)]
 mod tests {
 

iceberg-rust/Cargo.toml

@@ -30,4 +30,5 @@ thrift = { version = "0.17.0", default-features = false }
 thiserror = { workspace = true }
 derive-getters = { workspace = true }
 iceberg-rust-spec = { path = "../iceberg-rust-spec", version = "0.5.8" }
+smallvec = { version = "1.13.2", features = ["const_generics"] }
 

iceberg-rust/src/lib.rs

@@ -10,4 +10,5 @@ pub mod materialized_view;
 pub mod spec;
 pub mod sql;
 pub mod table;
+pub(crate) mod util;
 pub mod view;