Merge pull request JanKaul#255 from JanKaul/feat-snapshot-column-bounds

compute snapshot column bounds

Author: JanKaul

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

@@ -19,6 +19,7 @@ use std::{collections::HashMap, fmt, ops::Index, slice::Iter};
 
 use derive_builder::Builder;
 
+use itertools::Itertools;
 use serde::{
     de::{self, Error as SerdeError, IntoDeserializer, MapAccess, Visitor},
     Deserialize, Deserializer, Serialize, Serializer,
@@ -360,6 +361,29 @@ impl StructType {
     pub fn iter(&self) -> Iter<'_, StructField> {
         self.fields.iter()
     }
+
+    /// Returns an iterator over all field IDs in this struct, sorted in ascending order
+    ///
+    /// # Returns
+    /// * An iterator yielding field IDs (i32) in sorted order
+    pub fn field_ids(&self) -> impl Iterator<Item = i32> {
+        self.lookup.keys().map(ToOwned::to_owned).sorted()
+    }
+
+    /// Returns an iterator over field IDs of primitive-type fields only, sorted in ascending order
+    ///
+    /// This method filters the struct's fields to return only those with primitive types
+    /// (boolean, numeric, string, etc.), excluding complex types like structs, lists, and maps.
+    ///
+    /// # Returns
+    /// * An iterator yielding field IDs (i32) of primitive fields in sorted order
+    pub fn primitive_field_ids(&self) -> impl Iterator<Item = i32> {
+        self.lookup
+            .iter()
+            .filter(|(_, x)| matches!(self.fields[**x].field_type, Type::Primitive(_)))
+            .map(|x| x.0.to_owned())
+            .sorted()
+    }
 }
 
 impl Index<usize> for StructType {

iceberg-rust/src/table/manifest_list.rs

@@ -12,7 +12,7 @@ use std::{
 use apache_avro::{
     types::Value as AvroValue, Reader as AvroReader, Schema as AvroSchema, Writer as AvroWriter,
 };
-use futures::future::join_all;
+use futures::{future::join_all, TryStreamExt};
 use iceberg_rust_spec::{
     manifest::{partition_value_schema, DataFile, ManifestEntry, Status},
     manifest_list::{
@@ -28,7 +28,8 @@ use smallvec::SmallVec;
 
 use crate::{
     error::Error,
-    util::{summary_to_rectangle, Rectangle},
+    table::datafiles,
+    util::{summary_to_rectangle, Rectangle, Vec4},
 };
 
 use super::{
@@ -213,6 +214,81 @@ pub async fn snapshot_partition_bounds(
     })
 }
 
+/// Computes the column bounds (minimum and maximum values) for all primitive fields
+/// across all data files in a snapshot.
+///
+/// This function reads all manifests in the snapshot, extracts data files from them,
+/// and computes a bounding rectangle that encompasses the lower and upper bounds
+/// of all primitive columns across all data files.
+///
+/// # Arguments
+///
+/// * `snapshot` - The snapshot to compute column bounds for
+/// * `table_metadata` - Metadata of the table containing schema information
+/// * `object_store` - Object store implementation for reading manifest files
+///
+/// # Returns
+///
+/// Returns `Ok(Some(Rectangle))` containing the computed bounds, or `Ok(None)` if
+/// no data files are found. Returns an error if:
+/// - Schema cannot be resolved for the snapshot
+/// - Manifest files cannot be read
+/// - Column bounds are missing for any primitive field in any data file
+///
+/// # Errors
+///
+/// * `Error::NotFound` - When column bounds are missing for a primitive field
+/// * Other I/O errors from reading manifest or data files
+pub async fn snapshot_column_bounds(
+    snapshot: &Snapshot,
+    table_metadata: &TableMetadata,
+    object_store: Arc<dyn ObjectStore>,
+) -> Result<Option<Rectangle>, Error> {
+    let schema = table_metadata
+        .schema(*snapshot.snapshot_id())
+        .or(table_metadata.current_schema(None))?;
+    let manifests = read_snapshot(snapshot, table_metadata, object_store.clone())
+        .await?
+        .collect::<Result<Vec<_>, _>>()?;
+    let datafiles = datafiles(object_store, &manifests, None, (None, None)).await?;
+
+    let primitive_field_ids = schema.primitive_field_ids().collect::<Vec<_>>();
+    let n = primitive_field_ids.len();
+    datafiles
+        .try_fold(None::<Rectangle>, |acc, (_, manifest)| {
+            let primitive_field_ids = &primitive_field_ids;
+            async move {
+                let mut mins = Vec4::with_capacity(n);
+                let mut maxs = Vec4::with_capacity(n);
+                for (i, id) in primitive_field_ids.iter().enumerate() {
+                    let min = manifest
+                        .data_file()
+                        .lower_bounds()
+                        .as_ref()
+                        .and_then(|x| x.get(id));
+                    let max = manifest
+                        .data_file()
+                        .upper_bounds()
+                        .as_ref()
+                        .and_then(|x| x.get(id));
+                    let (Some(min), Some(max)) = (min, max) else {
+                        return Err(Error::NotFound("column bounds".to_string()));
+                    };
+                    mins[i] = min.clone();
+                    maxs[i] = max.clone();
+                }
+                let rect = Rectangle::new(mins, maxs);
+                if let Some(mut acc) = acc {
+                    acc.expand(&rect);
+                    Ok(Some(acc))
+                } else {
+                    Ok(Some(rect))
+                }
+            }
+        })
+        .await
+}
+
 /// A writer for Iceberg manifest list files that manages the creation and updating of manifest lists.
 ///
 /// The ManifestListWriter is responsible for:

iceberg-rust/src/util/mod.rs

@@ -8,7 +8,7 @@ use smallvec::SmallVec;
 
 use crate::error::Error;
 
-type Vec4<T> = SmallVec<[T; 4]>;
+pub(crate) type Vec4<T> = SmallVec<[T; 4]>;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Rectangle {