Change FieldSummary {upper,lower}_bound to ByteBuf (#1369)

## Which issue does this PR close?

I would like to invite everyone to roast my Rust-skills in order for me
to improve myself :)

Unblocks #1328

This aligns closely with PyIceberg and Java and greatly simplifies the
use of the Avro readers in PyIceberg. Otherwise, we would need to update
public APIs.

## What changes are included in this PR?



## Are these changes tested?



---------

Co-authored-by: Kevin Liu <[email protected]>

Author: Fokko

crates/iceberg/src/expr/visitors/manifest_evaluator.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use fnv::FnvHashSet;
+use serde_bytes::ByteBuf;
 
 use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
 use crate::expr::{BoundPredicate, BoundReference};
@@ -42,13 +43,13 @@ impl ManifestEvaluator {
     /// see if this `ManifestFile` could possibly contain data that matches
     /// the scan's filter.
     pub(crate) fn eval(&self, manifest_file: &ManifestFile) -> Result<bool> {
-        if manifest_file.partitions.is_empty() {
-            return Ok(true);
+        match &manifest_file.partitions {
+            Some(p) if !p.is_empty() => {
+                let mut evaluator = ManifestFilterVisitor::new(p);
+                visit(&mut evaluator, &self.partition_filter)
+            }
+            _ => Ok(true),
         }
-
-        let mut evaluator = ManifestFilterVisitor::new(&manifest_file.partitions);
-
-        visit(&mut evaluator, &self.partition_filter)
     }
 }
 
@@ -154,9 +155,19 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
         _predicate: &BoundPredicate,
     ) -> crate::Result<bool> {
         let field = self.field_summary_for_reference(reference);
+
         match &field.lower_bound {
-            Some(bound) if datum <= bound => ROWS_CANNOT_MATCH,
-            Some(_) => ROWS_MIGHT_MATCH,
+            Some(bound_bytes) => {
+                let bound = ManifestFilterVisitor::bytes_to_datum(
+                    bound_bytes,
+                    *reference.field().field_type.clone(),
+                );
+                if datum <= &bound {
+                    ROWS_CANNOT_MATCH
+                } else {
+                    ROWS_MIGHT_MATCH
+                }
+            }
             None => ROWS_CANNOT_MATCH,
         }
     }
@@ -169,8 +180,17 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
     ) -> crate::Result<bool> {
         let field = self.field_summary_for_reference(reference);
         match &field.lower_bound {
-            Some(bound) if datum < bound => ROWS_CANNOT_MATCH,
-            Some(_) => ROWS_MIGHT_MATCH,
+            Some(bound_bytes) => {
+                let bound = ManifestFilterVisitor::bytes_to_datum(
+                    bound_bytes,
+                    *reference.field().field_type.clone(),
+                );
+                if datum < &bound {
+                    ROWS_CANNOT_MATCH
+                } else {
+                    ROWS_MIGHT_MATCH
+                }
+            }
             None => ROWS_CANNOT_MATCH,
         }
     }
@@ -183,8 +203,17 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
     ) -> crate::Result<bool> {
         let field = self.field_summary_for_reference(reference);
         match &field.upper_bound {
-            Some(bound) if datum >= bound => ROWS_CANNOT_MATCH,
-            Some(_) => ROWS_MIGHT_MATCH,
+            Some(bound_bytes) => {
+                let bound = ManifestFilterVisitor::bytes_to_datum(
+                    bound_bytes,
+                    *reference.field().field_type.clone(),
+                );
+                if datum >= &bound {
+                    ROWS_CANNOT_MATCH
+                } else {
+                    ROWS_MIGHT_MATCH
+                }
+            }
             None => ROWS_CANNOT_MATCH,
         }
     }
@@ -197,8 +226,17 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
     ) -> crate::Result<bool> {
         let field = self.field_summary_for_reference(reference);
         match &field.upper_bound {
-            Some(bound) if datum > bound => ROWS_CANNOT_MATCH,
-            Some(_) => ROWS_MIGHT_MATCH,
+            Some(bound_bytes) => {
+                let bound = ManifestFilterVisitor::bytes_to_datum(
+                    bound_bytes,
+                    *reference.field().field_type.clone(),
+                );
+                if datum > &bound {
+                    ROWS_CANNOT_MATCH
+                } else {
+                    ROWS_MIGHT_MATCH
+                }
+            }
             None => ROWS_CANNOT_MATCH,
         }
     }
@@ -215,14 +253,22 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
             return ROWS_CANNOT_MATCH;
         }
 
-        if let Some(lower_bound) = &field.lower_bound {
-            if lower_bound > datum {
+        if let Some(lower_bound_bytes) = &field.lower_bound {
+            let lower_bound = ManifestFilterVisitor::bytes_to_datum(
+                lower_bound_bytes,
+                *reference.field().field_type.clone(),
+            );
+            if &lower_bound > datum {
                 return ROWS_CANNOT_MATCH;
             }
         }
 
-        if let Some(upper_bound) = &field.upper_bound {
-            if upper_bound < datum {
+        if let Some(upper_bound_bytes) = &field.upper_bound {
+            let upper_bound = ManifestFilterVisitor::bytes_to_datum(
+                upper_bound_bytes,
+                *reference.field().field_type.clone(),
+            );
+            if &upper_bound < datum {
                 return ROWS_CANNOT_MATCH;
             }
         }
@@ -260,23 +306,15 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
         let prefix_len = prefix.len();
 
         if let Some(lower_bound) = &field.lower_bound {
-            let lower_bound_str = ManifestFilterVisitor::datum_as_str(
-                lower_bound,
-                "Cannot perform starts_with on non-string lower bound",
-            )?;
-            let min_len = lower_bound_str.len().min(prefix_len);
-            if prefix.as_bytes().lt(&lower_bound_str.as_bytes()[..min_len]) {
+            let min_len = lower_bound.len().min(prefix_len);
+            if prefix.as_bytes().lt(&lower_bound[..min_len]) {
                 return ROWS_CANNOT_MATCH;
             }
         }
 
         if let Some(upper_bound) = &field.upper_bound {
-            let upper_bound_str = ManifestFilterVisitor::datum_as_str(
-                upper_bound,
-                "Cannot perform starts_with on non-string upper bound",
-            )?;
-            let min_len = upper_bound_str.len().min(prefix_len);
-            if prefix.as_bytes().gt(&upper_bound_str.as_bytes()[..min_len]) {
+            let min_len = upper_bound.len().min(prefix_len);
+            if prefix.as_bytes().gt(&upper_bound[..min_len]) {
                 return ROWS_CANNOT_MATCH;
             }
         }
@@ -305,35 +343,19 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
         // not_starts_with will match unless all values must start with the prefix. This happens when
         // the lower and upper bounds both start with the prefix.
         if let Some(lower_bound) = &field.lower_bound {
-            let lower_bound_str = ManifestFilterVisitor::datum_as_str(
-                lower_bound,
-                "Cannot perform not_starts_with on non-string lower bound",
-            )?;
-
             // if lower is shorter than the prefix then lower doesn't start with the prefix
-            if prefix_len > lower_bound_str.len() {
+            if prefix_len > lower_bound.len() {
                 return ROWS_MIGHT_MATCH;
             }
 
-            if prefix
-                .as_bytes()
-                .eq(&lower_bound_str.as_bytes()[..prefix_len])
-            {
+            if prefix.as_bytes().eq(&lower_bound[..prefix_len]) {
                 if let Some(upper_bound) = &field.upper_bound {
-                    let upper_bound_str = ManifestFilterVisitor::datum_as_str(
-                        upper_bound,
-                        "Cannot perform not_starts_with on non-string upper bound",
-                    )?;
-
                     // if upper is shorter than the prefix then upper can't start with the prefix
-                    if prefix_len > upper_bound_str.len() {
+                    if prefix_len > upper_bound.len() {
                         return ROWS_MIGHT_MATCH;
                     }
 
-                    if prefix
-                        .as_bytes()
-                        .eq(&upper_bound_str.as_bytes()[..prefix_len])
-                    {
+                    if prefix.as_bytes().eq(&upper_bound[..prefix_len]) {
                         return ROWS_CANNOT_MATCH;
                     }
                 }
@@ -359,13 +381,21 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
         }
 
         if let Some(lower_bound) = &field.lower_bound {
-            if literals.iter().all(|datum| lower_bound > datum) {
+            let lower_bound = ManifestFilterVisitor::bytes_to_datum(
+                lower_bound,
+                *reference.field().clone().field_type,
+            );
+            if literals.iter().all(|datum| &lower_bound > datum) {
                 return ROWS_CANNOT_MATCH;
             }
         }
 
         if let Some(upper_bound) = &field.upper_bound {
-            if literals.iter().all(|datum| upper_bound < datum) {
+            let upper_bound = ManifestFilterVisitor::bytes_to_datum(
+                upper_bound,
+                *reference.field().clone().field_type,
+            );
+            if literals.iter().all(|datum| &upper_bound < datum) {
                 return ROWS_CANNOT_MATCH;
             }
         }
@@ -414,6 +444,11 @@ impl ManifestFilterVisitor<'_> {
         };
         Ok(bound)
     }
+
+    fn bytes_to_datum(bytes: &ByteBuf, t: Type) -> Datum {
+        let p = t.as_primitive_type().unwrap();
+        Datum::try_from_bytes(bytes, p.clone()).unwrap()
+    }
 }
 
 #[cfg(test)]
@@ -520,8 +555,8 @@ mod test {
             FieldSummary {
                 contains_null: false,
                 contains_nan: None,
-                lower_bound: Some(Datum::int(INT_MIN_VALUE)),
-                upper_bound: Some(Datum::int(INT_MAX_VALUE)),
+                lower_bound: Some(Datum::int(INT_MIN_VALUE).to_bytes().unwrap()),
+                upper_bound: Some(Datum::int(INT_MAX_VALUE).to_bytes().unwrap()),
             },
             // all_nulls_missing_nan
             FieldSummary {
@@ -534,22 +569,22 @@ mod test {
             FieldSummary {
                 contains_null: true,
                 contains_nan: None,
-                lower_bound: Some(Datum::string(STRING_MIN_VALUE)),
-                upper_bound: Some(Datum::string(STRING_MAX_VALUE)),
+                lower_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
+                upper_bound: Some(Datum::string(STRING_MAX_VALUE).to_bytes().unwrap()),
             },
             // no_nulls
             FieldSummary {
                 contains_null: false,
                 contains_nan: None,
-                lower_bound: Some(Datum::string(STRING_MIN_VALUE)),
-                upper_bound: Some(Datum::string(STRING_MAX_VALUE)),
+                lower_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
+                upper_bound: Some(Datum::string(STRING_MAX_VALUE).to_bytes().unwrap()),
             },
             // float
             FieldSummary {
                 contains_null: true,
                 contains_nan: None,
-                lower_bound: Some(Datum::float(0.0)),
-                upper_bound: Some(Datum::float(20.0)),
+                lower_bound: Some(Datum::float(0.0).to_bytes().unwrap()),
+                upper_bound: Some(Datum::float(20.0).to_bytes().unwrap()),
             },
             // all_nulls_double
             FieldSummary {
@@ -583,8 +618,8 @@ mod test {
             FieldSummary {
                 contains_null: false,
                 contains_nan: Some(false),
-                lower_bound: Some(Datum::float(0.0)),
-                upper_bound: Some(Datum::float(20.0)),
+                lower_bound: Some(Datum::float(0.0).to_bytes().unwrap()),
+                upper_bound: Some(Datum::float(20.0).to_bytes().unwrap()),
             },
             // all_nulls_missing_nan_float
             FieldSummary {
@@ -597,15 +632,15 @@ mod test {
             FieldSummary {
                 contains_null: true,
                 contains_nan: None,
-                lower_bound: Some(Datum::string(STRING_MIN_VALUE)),
-                upper_bound: Some(Datum::string(STRING_MIN_VALUE)),
+                lower_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
+                upper_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
             },
             // no_nulls_same_value_a
             FieldSummary {
                 contains_null: false,
                 contains_nan: None,
-                lower_bound: Some(Datum::string(STRING_MIN_VALUE)),
-                upper_bound: Some(Datum::string(STRING_MIN_VALUE)),
+                lower_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
+                upper_bound: Some(Datum::string(STRING_MIN_VALUE).to_bytes().unwrap()),
             },
         ]
     }
@@ -625,7 +660,7 @@ mod test {
             added_rows_count: None,
             existing_rows_count: None,
             deleted_rows_count: None,
-            partitions,
+            partitions: Some(partitions),
             key_metadata: vec![],
         }
     }

crates/iceberg/src/inspect/manifests.rs

@@ -29,7 +29,7 @@ use futures::{StreamExt, stream};
 use crate::Result;
 use crate::arrow::schema_to_arrow_schema;
 use crate::scan::ArrowRecordBatchStream;
-use crate::spec::{FieldSummary, ListType, NestedField, PrimitiveType, StructType, Type};
+use crate::spec::{Datum, FieldSummary, ListType, NestedField, PrimitiveType, StructType, Type};
 use crate::table::Table;
 
 /// Manifests table.
@@ -181,7 +181,20 @@ impl<'a> ManifestsTable<'a> {
                     .append_value(manifest.existing_files_count.unwrap_or(0) as i32);
                 deleted_delete_files_count
                     .append_value(manifest.deleted_files_count.unwrap_or(0) as i32);
-                self.append_partition_summaries(&mut partition_summaries, &manifest.partitions);
+
+                let spec = self
+                    .table
+                    .metadata()
+                    .partition_spec_by_id(manifest.partition_spec_id)
+                    .unwrap();
+                let spec_struct = spec
+                    .partition_type(self.table.metadata().current_schema())
+                    .unwrap();
+                self.append_partition_summaries(
+                    &mut partition_summaries,
+                    &manifest.partitions.clone().unwrap_or_else(Vec::new),
+                    spec_struct,
+                );
             }
         }
 
@@ -230,9 +243,10 @@ impl<'a> ManifestsTable<'a> {
         &self,
         builder: &mut GenericListBuilder<i32, StructBuilder>,
         partitions: &[FieldSummary],
+        partition_struct: StructType,
     ) {
         let partition_summaries_builder = builder.values();
-        for summary in partitions {
+        for (summary, field) in partitions.iter().zip(partition_struct.fields()) {
             partition_summaries_builder
                 .field_builder::<BooleanBuilder>(0)
                 .unwrap()
@@ -241,14 +255,23 @@ impl<'a> ManifestsTable<'a> {
                 .field_builder::<BooleanBuilder>(1)
                 .unwrap()
                 .append_option(summary.contains_nan);
+
             partition_summaries_builder
                 .field_builder::<StringBuilder>(2)
                 .unwrap()
-                .append_option(summary.lower_bound.as_ref().map(|v| v.to_string()));
+                .append_option(summary.lower_bound.as_ref().map(|v| {
+                    Datum::try_from_bytes(v, field.field_type.as_primitive_type().unwrap().clone())
+                        .unwrap()
+                        .to_string()
+                }));
             partition_summaries_builder
                 .field_builder::<StringBuilder>(3)
                 .unwrap()
-                .append_option(summary.upper_bound.as_ref().map(|v| v.to_string()));
+                .append_option(summary.upper_bound.as_ref().map(|v| {
+                    Datum::try_from_bytes(v, field.field_type.as_primitive_type().unwrap().clone())
+                        .unwrap()
+                        .to_string()
+                }));
             partition_summaries_builder.append(true);
         }
         builder.append(true);