refactor: iceberg::spec::values::Struct to remove bitvec (#1203)

Author: xxchan

.github/workflows/ci.yml

@@ -155,7 +155,7 @@ jobs:
           cargo generate-lockfile -Z direct-minimal-versions -Z minimal-versions
           # Some dependencies don't correctly specify a minimal version for their dependencies and will fail to build.
           # So we update these transitive dependencies here.
-          cargo update tap faststr metainfo linkedbytes
+          cargo update faststr metainfo linkedbytes
           # munge 0.4.2 will cause fail to use 1.7.1 so specify the correct version here.
           cargo update -p munge --precise 0.4.1
       - name: Setup MSRV Rust toolchain

Cargo.lock

@@ -55,7 +55,6 @@ async-trait = "0.1.86"
 aws-config = "1"
 aws-sdk-glue = "1.39"
 bimap = "0.6"
-bitvec = "1.0.1"
 bytes = "1.6"
 chrono = "0.4.38"
 ctor = "0.2.8"

Cargo.toml

@@ -35,8 +35,8 @@ storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
 storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]
-storage-s3 = ["opendal/services-s3"]
 storage-oss = ["opendal/services-oss"]
+storage-s3 = ["opendal/services-s3"]
 
 async-std = ["dep:async-std"]
 tokio = ["tokio/rt-multi-thread"]
@@ -56,7 +56,6 @@ arrow-string = { workspace = true }
 async-std = { workspace = true, optional = true, features = ["attributes"] }
 async-trait = { workspace = true }
 bimap = { workspace = true }
-bitvec = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true }
 derive_builder = { workspace = true }

crates/iceberg/Cargo.toml

@@ -28,7 +28,6 @@ use arrow_array::{
     Int64Array, PrimitiveArray, Scalar, StringArray, TimestampMicrosecondArray,
 };
 use arrow_schema::{DataType, Field, Fields, Schema as ArrowSchema, TimeUnit};
-use bitvec::macros::internal::funty::Fundamental;
 use num_bigint::BigInt;
 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
 use parquet::file::statistics::Statistics;
@@ -662,10 +661,10 @@ pub(crate) fn get_arrow_datum(datum: &Datum) -> Result<Box<dyn ArrowDatum + Send
             Ok(Box::new(Int64Array::new_scalar(*value)))
         }
         (PrimitiveType::Float, PrimitiveLiteral::Float(value)) => {
-            Ok(Box::new(Float32Array::new_scalar(value.as_f32())))
+            Ok(Box::new(Float32Array::new_scalar(value.to_f32().unwrap())))
         }
         (PrimitiveType::Double, PrimitiveLiteral::Double(value)) => {
-            Ok(Box::new(Float64Array::new_scalar(value.as_f64())))
+            Ok(Box::new(Float64Array::new_scalar(value.to_f64().unwrap())))
         }
         (PrimitiveType::String, PrimitiveLiteral::String(value)) => {
             Ok(Box::new(StringArray::new_scalar(value.as_str())))

crates/iceberg/src/arrow/schema.rs

@@ -58,20 +58,18 @@ impl StructAccessor {
 
     pub(crate) fn get<'a>(&'a self, container: &'a Struct) -> Result<Option<Datum>> {
         match &self.inner {
-            None => {
-                if container.is_null_at_index(self.position) {
-                    Ok(None)
-                } else if let Literal::Primitive(literal) = &container[self.position] {
+            None => match &container[self.position] {
+                None => Ok(None),
+                Some(Literal::Primitive(literal)) => {
                     Ok(Some(Datum::new(self.r#type().clone(), literal.clone())))
-                } else {
-                    Err(Error::new(
-                        ErrorKind::Unexpected,
-                        "Expected Literal to be Primitive",
-                    ))
                 }
-            }
+                Some(_) => Err(Error::new(
+                    ErrorKind::Unexpected,
+                    "Expected Literal to be Primitive",
+                )),
+            },
             Some(inner) => {
-                if let Literal::Struct(wrapped) = &container[self.position] {
+                if let Some(Literal::Struct(wrapped)) = &container[self.position] {
                     inner.get(wrapped)
                 } else {
                     Err(Error::new(

crates/iceberg/src/expr/accessor.rs

@@ -163,11 +163,7 @@ impl PartitionSpec {
             .iter()
             .enumerate()
             .map(|(i, field)| {
-                let value = if data.is_null_at_index(i) {
-                    None
-                } else {
-                    Some(&data.fields()[i])
-                };
+                let value = data[i].as_ref();
                 format!(
                     "{}={}",
                     field.name,

crates/iceberg/src/spec/partition.rs

@@ -27,7 +27,6 @@ use std::ops::Index;
 use std::str::FromStr;
 
 pub use _serde::RawLiteral;
-use bitvec::vec::BitVec;
 use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc};
 use num_bigint::BigInt;
 use ordered_float::OrderedFloat;
@@ -1726,102 +1725,53 @@ impl Literal {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct Struct {
     /// Vector to store the field values
-    fields: Vec<Literal>,
-    /// Null bitmap
-    null_bitmap: BitVec,
+    fields: Vec<Option<Literal>>,
 }
 
 impl Struct {
     /// Create a empty struct.
     pub fn empty() -> Self {
-        Self {
-            fields: Vec::new(),
-            null_bitmap: BitVec::new(),
-        }
+        Self { fields: Vec::new() }
     }
 
     /// Create a iterator to read the field in order of field_value.
     pub fn iter(&self) -> impl ExactSizeIterator<Item = Option<&Literal>> {
-        self.null_bitmap.iter().zip(self.fields.iter()).map(
-            |(null, value)| {
-                if *null {
-                    None
-                } else {
-                    Some(value)
-                }
-            },
-        )
+        self.fields.iter().map(|field| field.as_ref())
     }
 
     /// returns true if the field at position `index` is null
     pub fn is_null_at_index(&self, index: usize) -> bool {
-        self.null_bitmap[index]
+        self.fields[index].is_none()
     }
 
     /// Return fields in the struct.
-    pub fn fields(&self) -> &[Literal] {
+    pub fn fields(&self) -> &[Option<Literal>] {
         &self.fields
     }
 }
 
 impl Index<usize> for Struct {
-    type Output = Literal;
+    type Output = Option<Literal>;
 
     fn index(&self, idx: usize) -> &Self::Output {
         &self.fields[idx]
     }
 }
 
-/// An iterator that moves out of a struct.
-pub struct StructValueIntoIter {
-    null_bitmap: bitvec::boxed::IntoIter,
-    fields: std::vec::IntoIter<Literal>,
-}
-
-impl Iterator for StructValueIntoIter {
-    type Item = Option<Literal>;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        match (self.null_bitmap.next(), self.fields.next()) {
-            (Some(null), Some(value)) => Some(if null { None } else { Some(value) }),
-            _ => None,
-        }
-    }
-}
-
 impl IntoIterator for Struct {
     type Item = Option<Literal>;
 
-    type IntoIter = StructValueIntoIter;
+    type IntoIter = std::vec::IntoIter<Option<Literal>>;
 
     fn into_iter(self) -> Self::IntoIter {
-        StructValueIntoIter {
-            null_bitmap: self.null_bitmap.into_iter(),
-            fields: self.fields.into_iter(),
-        }
+        self.fields.into_iter()
     }
 }
 
 impl FromIterator<Option<Literal>> for Struct {
     fn from_iter<I: IntoIterator<Item = Option<Literal>>>(iter: I) -> Self {
-        let mut fields = Vec::new();
-        let mut null_bitmap = BitVec::new();
-
-        for value in iter.into_iter() {
-            match value {
-                Some(value) => {
-                    fields.push(value);
-                    null_bitmap.push(false)
-                }
-                None => {
-                    fields.push(Literal::Primitive(PrimitiveLiteral::Boolean(false)));
-                    null_bitmap.push(true)
-                }
-            }
-        }
         Struct {
-            fields,
-            null_bitmap,
+            fields: iter.into_iter().collect(),
         }
     }
 }
@@ -3402,7 +3352,7 @@ mod tests {
             match (&desered_literal, &struct_literal) {
                 (Literal::Struct(desered), Literal::Struct(expected)) => {
                     match (&desered.fields[0], &expected.fields[0]) {
-                        (Literal::Map(desered), Literal::Map(expected)) => {
+                        (Some(Literal::Map(desered)), Some(Literal::Map(expected))) => {
                             assert!(desered.has_same_content(expected))
                         }
                         _ => {

crates/iceberg/src/spec/values.rs

@@ -105,21 +105,19 @@ impl<'a> SnapshotProduceAction<'a> {
         }
 
         for (value, field) in partition_value.fields().iter().zip(partition_type.fields()) {
-            if !field
-                .field_type
-                .as_primitive_type()
-                .ok_or_else(|| {
-                    Error::new(
-                        ErrorKind::Unexpected,
-                        "Partition field should only be primitive type.",
-                    )
-                })?
-                .compatible(&value.as_primitive_literal().unwrap())
-            {
-                return Err(Error::new(
-                    ErrorKind::DataInvalid,
-                    "Partition value is not compatible partition type",
-                ));
+            let field = field.field_type.as_primitive_type().ok_or_else(|| {
+                Error::new(
+                    ErrorKind::Unexpected,
+                    "Partition field should only be primitive type.",
+                )
+            })?;
+            if let Some(value) = value {
+                if !field.compatible(&value.as_primitive_literal().unwrap()) {
+                    return Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        "Partition value is not compatible partition type",
+                    ));
+                }
             }
         }
         Ok(())