refactor(rust/sedona-spatial-join): Support wraparound rectangles in EvaluatedGeometryArray (apache#799)

Co-authored-by: Copilot <copilot@github.com>

Author: paleolimbot

Cargo.lock

@@ -16,7 +16,6 @@
 // under the License.
 
 use arrow_schema::DataType;
-use geo_types::Rect;
 
 mod error;
 #[cfg(gpu_available)]
@@ -113,9 +112,10 @@ mod sys {
         /// Inserts a batch of bounding boxes into the index.
         /// Each rectangle is represented as a `Rect<f32>` with minimum and maximum x and y coordinates.
         /// This method accumulates these rectangles until `finish_building` is called to finalize the index.
-        /// The method can be called multiple times to insert data in batches before finalizing.
-        pub fn push_build(&mut self, rects: &[Rect<f32>]) -> Result<()> {
-            // Re-interpreting Rect<f32> as flat f32 array (xmin, ymin, xmax, ymax)
+        /// The method can be called multiple times to insert data in batches before finalizing. The values
+        /// in rects are ordered (xmin, ymin, xmax, ymax).
+        pub fn push_build(&mut self, rects: &[[f32; 4]]) -> Result<()> {
+            // Re-interpreting rects as a flat f32 array (xmin, ymin, xmax, ymax)
             let raw_ptr = rects.as_ptr() as *const f32;
             self.inner.push_build(raw_ptr, rects.len() as u32)
         }
@@ -126,7 +126,8 @@ mod sys {
         }
 
         /// Probes the spatial index with a batch of rectangles and returns pairs of matching indices from the build and probe sets.
-        pub fn probe(&self, rects: &[Rect<f32>]) -> Result<(Vec<u32>, Vec<u32>)> {
+        ///  The values in rects are ordered (xmin, ymin, xmax, ymax).
+        pub fn probe(&self, rects: &[[f32; 4]]) -> Result<(Vec<u32>, Vec<u32>)> {
             let raw_ptr = rects.as_ptr() as *const f32;
             self.inner.probe(raw_ptr, rects.len() as u32)
         }
@@ -203,13 +204,13 @@ mod sys {
             Err(GpuSpatialError::GpuNotAvailable)
         }
         pub fn clear(&mut self) {}
-        pub fn push_build(&mut self, _r: &[Rect<f32>]) -> Result<()> {
+        pub fn push_build(&mut self, _r: &[[f32; 4]]) -> Result<()> {
             Err(GpuSpatialError::GpuNotAvailable)
         }
         pub fn finish_building(&mut self) -> Result<GpuSpatialIndex> {
             Err(GpuSpatialError::GpuNotAvailable)
         }
-        pub fn probe(&self, _r: &[Rect<f32>]) -> Result<(Vec<u32>, Vec<u32>)> {
+        pub fn probe(&self, _r: &[[f32; 4]]) -> Result<(Vec<u32>, Vec<u32>)> {
             Err(GpuSpatialError::GpuNotAvailable)
         }
     }
@@ -268,13 +269,16 @@ mod tests {
             Some("POLYGON ((30 10, 40 40, 20 40, 10 20, 30 10))"),
             Some("POLYGON ((35 10, 45 45, 15 40, 10 20, 35 10), (20 30, 35 35, 30 20, 20 30))"),
         ];
-        let rects: Vec<Rect<f32>> = polygon_values
+        let rects: Vec<_> = polygon_values
             .iter()
             .map(|w| {
-                Polygon::try_from_wkt_str(w.unwrap())
+                let rect = Polygon::try_from_wkt_str(w.unwrap())
                     .unwrap()
                     .bounding_rect()
-                    .unwrap()
+                    .unwrap();
+                let min = rect.min();
+                let max = rect.max();
+                (min.x, min.y, max.x, max.y)
             })
             .collect();
 
@@ -286,9 +290,14 @@ mod tests {
 
         // 4. Probe (Index is immutable and safe)
         let point_values = &[Some("POINT (30 20)")];
-        let points: Vec<Rect<f32>> = point_values
+        let points: Vec<_> = point_values
             .iter()
-            .map(|w| Point::try_from_wkt_str(w.unwrap()).unwrap().bounding_rect())
+            .map(|w| {
+                let rect = Point::try_from_wkt_str(w.unwrap()).unwrap().bounding_rect();
+                let min = rect.min();
+                let max = rect.max();
+                (min.x, min.y, max.x, max.y)
+            })
             .collect();
 
         let (build_idx, probe_idx) = index.probe(&points).unwrap();

c/sedona-libgpuspatial/src/lib.rs

@@ -26,15 +26,17 @@ use arrow_schema::{DataType, Field, FieldRef};
 use datafusion_common::{
     cast::as_binary_array,
     error::{DataFusionError, Result},
-    ScalarValue,
+    exec_datafusion_err, ScalarValue,
 };
 use datafusion_expr::Volatility;
 use datafusion_expr::{Accumulator, ColumnarValue};
 use sedona_common::{sedona_internal_datafusion_err, sedona_internal_err};
 use sedona_expr::aggregate_udf::{SedonaAccumulatorRef, SedonaAggregateUDF};
 use sedona_expr::item_crs::ItemCrsSedonaAccumulator;
 use sedona_expr::{aggregate_udf::SedonaAccumulator, statistics::GeoStatistics};
-use sedona_geometry::analyze::GeometrySummary;
+use sedona_geometry::analyze::{analyze_wkb, GeometrySummary};
+use sedona_geometry::bounding_box::BoundingBox;
+use sedona_geometry::bounds::geo_traits_bounds_xy;
 use sedona_geometry::interval::IntervalTrait;
 use sedona_geometry::types::{GeometryTypeAndDimensions, GeometryTypeAndDimensionsSet};
 use sedona_schema::{datatypes::SedonaType, matchers::ArgMatcher};
@@ -77,12 +79,9 @@ impl SedonaAccumulator for STAnalyzeAgg {
     fn accumulator(
         &self,
         args: &[SedonaType],
-        output_type: &SedonaType,
+        _output_type: &SedonaType,
     ) -> Result<Box<dyn Accumulator>> {
-        Ok(Box::new(AnalyzeAccumulator::new(
-            args[0].clone(),
-            output_type.clone(),
-        )))
+        Ok(Box::new(AnalyzeAccumulator::new(args[0].clone())))
     }
 
     fn state_fields(&self, _args: &[SedonaType]) -> Result<Vec<FieldRef>> {
@@ -223,38 +222,42 @@ impl STAnalyzeAgg {
 #[derive(Debug)]
 pub struct AnalyzeAccumulator {
     input_type: SedonaType,
-    _output_type: SedonaType,
     stats: GeoStatistics,
 }
 
 impl AnalyzeAccumulator {
-    pub fn new(input_type: SedonaType, output_type: SedonaType) -> Self {
+    pub fn new(input_type: SedonaType) -> Self {
         Self {
             input_type,
-            _output_type: output_type,
             stats: GeoStatistics::empty(),
         }
     }
 
-    pub fn update_statistics(&mut self, geom: &Wkb) -> Result<()> {
-        // Get geometry analysis information
-        let summary = sedona_geometry::analyze::analyze_geometry(geom)
-            .map_err(|e| DataFusionError::External(Box::new(e)))?;
+    pub fn update_statistics_with_bbox(&mut self, geom: &Wkb, bbox: &BoundingBox) -> Result<()> {
+        let summary = analyze_wkb(geom).map_err(|e| DataFusionError::External(Box::new(e)))?;
 
-        self.ingest_geometry_summary(&summary);
+        self.ingest_geometry_summary(&summary, bbox);
+        Ok(())
+    }
 
+    fn update_statistics(&mut self, geom: &Wkb) -> Result<()> {
+        let bbox =
+            geo_traits_bounds_xy(geom).map_err(|e| exec_datafusion_err!("Bounding error: {e}"))?;
+        let summary = analyze_wkb(geom).map_err(|e| exec_datafusion_err!("Analysis error: {e}"))?;
+
+        self.ingest_geometry_summary(&summary, &bbox);
         Ok(())
     }
 
-    pub fn ingest_geometry_summary(&mut self, summary: &GeometrySummary) {
+    fn ingest_geometry_summary(&mut self, summary: &GeometrySummary, bbox: &BoundingBox) {
         // Start with a clone of the current stats
         let mut stats = self.stats.clone();
 
         // Update each component of the statistics
         stats = self.update_basic_counts(stats, summary.size_bytes);
         stats = self.update_geometry_type_counts(stats, summary);
         stats = self.update_point_count(stats, summary.point_count);
-        stats = self.update_envelope_info(stats, summary);
+        stats = self.update_envelope_info(stats, bbox);
         stats = self.update_geometry_types(stats, summary.geometry_type);
 
         // Assign the updated stats back to self.stats
@@ -300,14 +303,7 @@ impl AnalyzeAccumulator {
     }
 
     // Update envelope dimensions and bounding box
-    fn update_envelope_info(
-        &self,
-        stats: GeoStatistics,
-        analysis: &GeometrySummary,
-    ) -> GeoStatistics {
-        // The bbox is directly available on analysis, not wrapped in an Option
-        let bbox = &analysis.bbox;
-
+    fn update_envelope_info(&self, stats: GeoStatistics, bbox: &BoundingBox) -> GeoStatistics {
         // Calculate envelope width and height from the bbox
         let envelope_width = if bbox.x().is_empty() {
             0.0