perf(vortex-geo): short-circuit predicates with bounding boxes (#9076)

## Rationale for this change

Constant-vs-column geometry predicates currently run the exact spatial
predicate for every valid row, even when the constant and row bounding
rectangles already prove that the result is false. For complex constant
geometries and spatially selective queries, this repeats substantially
more work than necessary.

## What changes are included in this PR?

- Add an optional bounding-rectangle pre-check to the shared
null-propagating binary geometry executor.
- Short-circuit `vortex.geo.intersects` when operand rectangles are
disjoint.
- Short-circuit `vortex.geo.contains` when the container rectangle
cannot cover the contained rectangle, preserving operand order.
- Keep `vortex.geo.distance` on exact evaluation because rectangles
cannot determine the result.
- Preserve null propagation and fall back to exact evaluation for empty
geometries or overlapping rectangles.
- Add regression coverage for short-circuiting, nulls, operand order,
empty geometries, exact-result equivalence, containment, and distance.
- Add the `predicate_bbox` Divan microbenchmark.

### Microbenchmark

Command:

```bash
cargo bench -p vortex-geo --bench predicate_bbox -- --min-time 0.1 --sample-count 10
```

The benchmark evaluates 8,192 points against a 128-vertex constant
polygon. Median results:

| Predicate / workload | Exact | Bounding-box pre-check | Change |
| --- | ---: | ---: | ---: |
| `contains`, disjoint | 689.0 us | 16.87 us | 40.8x faster |
| `contains`, candidate | 674.4 us | 667.4 us | 1.0% faster |
| `intersects`, disjoint | 670.5 us | 16.79 us | 39.9x faster |
| `intersects`, candidate | 653.5 us | 670.4 us | 2.6% slower |

---------

Signed-off-by: Nemo Yu <zyu379@wisc.edu>

Author: HarukiMoriarty

vortex-bench/src/spatialbench/benchmark.rs

@@ -148,7 +148,7 @@ impl Benchmark for SpatialBenchBenchmark {
         // shorter vec means the runner skips them).
         match self.scale_factor.as_str() {
             "1.0" => Some(vec![0, 94, 1, 22, 258, 316691, 3, 6000000, 369, 37]),
-            "10.0" => Some(vec![0, 994, 1, 79, 231, 3144328, 3, 60000000, 9357, 573]),
+            "10.0" => Some(vec![0, 994, 1, 79, 243, 3144328, 3, 60000000, 9357, 573]),
             _ => None,
         }
     }

vortex-geo/Cargo.toml

@@ -45,5 +45,9 @@ vortex-layout = { workspace = true }
 name = "envelope"
 harness = false
 
+[[bench]]
+name = "predicate_bbox"
+harness = false
+
 [lints]
 workspace = true

vortex-geo/benches/predicate_bbox.rs

@@ -0,0 +1,159 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Microbenchmark for the bounding-box pre-check used by constant-vs-column geometry predicates.
+//!
+//! `disjoint` rows are rejected by their bounding boxes, while `candidate` rows overlap the
+//! constant's box and therefore pay for both the pre-check and the exact predicate. Comparing the
+//! candidate cases tracks the worst-case overhead of the optimization.
+//!
+//! Run with `cargo bench -p vortex-geo --bench predicate_bbox`.
+
+use std::f64::consts::TAU;
+use std::sync::LazyLock;
+
+use divan::Bencher;
+use divan::counter::ItemsCount;
+use geo::BoundingRect;
+use geo::Contains;
+use geo::Intersects;
+use geo_types::Geometry;
+use geo_types::LineString;
+use geo_types::Point;
+use geo_types::Polygon;
+
+fn main() {
+    divan::main();
+}
+
+const ROWS: usize = 1 << 13;
+const VERTICES: usize = 128;
+
+static QUERY: LazyLock<Geometry<f64>> = LazyLock::new(|| {
+    let mut ring: Vec<_> = (0..VERTICES)
+        .map(|i| {
+            let theta = TAU * i as f64 / VERTICES as f64;
+            (10.0 * theta.cos(), 10.0 * theta.sin())
+        })
+        .collect();
+    ring.push(ring[0]);
+    Geometry::Polygon(Polygon::new(LineString::from(ring), vec![]))
+});
+
+static DISJOINT: LazyLock<Vec<Geometry<f64>>> = LazyLock::new(|| {
+    (0..ROWS)
+        .map(|i| Geometry::Point(Point::new(100.0 + ordinate(i), 100.0 + ordinate(i + 1))))
+        .collect()
+});
+
+static CANDIDATES: LazyLock<Vec<Geometry<f64>>> = LazyLock::new(|| {
+    (0..ROWS)
+        .map(|i| {
+            Geometry::Point(Point::new(
+                20.0 * ordinate(i) - 10.0,
+                20.0 * ordinate(i + 1) - 10.0,
+            ))
+        })
+        .collect()
+});
+
+/// Deterministic pseudo-random value in `[0, 1)`.
+fn ordinate(i: usize) -> f64 {
+    (i.wrapping_mul(2_654_435_761) % 10_000) as f64 / 10_000.0
+}
+
+fn exact_intersects(query: &Geometry<f64>, rows: &[Geometry<f64>]) -> usize {
+    rows.iter().filter(|row| query.intersects(*row)).count()
+}
+
+fn bbox_intersects(query: &Geometry<f64>, rows: &[Geometry<f64>]) -> usize {
+    let query_rect = query.bounding_rect();
+    rows.iter()
+        .filter(|row| {
+            let rejected = query_rect
+                .zip(row.bounding_rect())
+                .is_some_and(|(query, row)| !query.intersects(&row));
+            !rejected && query.intersects(*row)
+        })
+        .count()
+}
+
+fn exact_contains(query: &Geometry<f64>, rows: &[Geometry<f64>]) -> usize {
+    rows.iter().filter(|row| query.contains(*row)).count()
+}
+
+fn bbox_contains(query: &Geometry<f64>, rows: &[Geometry<f64>]) -> usize {
+    let query_rect = query.bounding_rect();
+    rows.iter()
+        .filter(|row| {
+            let rejected = query_rect
+                .zip(row.bounding_rect())
+                .is_some_and(|(query, row)| !query.contains(&row));
+            !rejected && query.contains(*row)
+        })
+        .count()
+}
+
+#[divan::bench]
+fn intersects_exact_disjoint(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*DISJOINT))
+        .bench_refs(|(query, rows)| exact_intersects(query, rows));
+}
+
+#[divan::bench]
+fn intersects_bbox_disjoint(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*DISJOINT))
+        .bench_refs(|(query, rows)| bbox_intersects(query, rows));
+}
+
+#[divan::bench]
+fn intersects_exact_candidate(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*CANDIDATES))
+        .bench_refs(|(query, rows)| exact_intersects(query, rows));
+}
+
+#[divan::bench]
+fn intersects_bbox_candidate(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*CANDIDATES))
+        .bench_refs(|(query, rows)| bbox_intersects(query, rows));
+}
+
+#[divan::bench]
+fn contains_exact_disjoint(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*DISJOINT))
+        .bench_refs(|(query, rows)| exact_contains(query, rows));
+}
+
+#[divan::bench]
+fn contains_bbox_disjoint(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*DISJOINT))
+        .bench_refs(|(query, rows)| bbox_contains(query, rows));
+}
+
+#[divan::bench]
+fn contains_exact_candidate(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*CANDIDATES))
+        .bench_refs(|(query, rows)| exact_contains(query, rows));
+}
+
+#[divan::bench]
+fn contains_bbox_candidate(bencher: Bencher) {
+    bencher
+        .counter(ItemsCount::new(ROWS))
+        .with_inputs(|| (&*QUERY, &*CANDIDATES))
+        .bench_refs(|(query, rows)| bbox_contains(query, rows));
+}

vortex-geo/src/scalar_fn/contains.rs

@@ -84,8 +84,16 @@ impl ScalarFnVTable for GeoContains {
     ) -> VortexResult<ArrayRef> {
         let a = args.get(0)?;
         let b = args.get(1)?;
-        // Containment is not symmetric: `a` is always the container and `b` the contained.
-        execute_null_propagating(&a, &b, |a, b| a.contains(b), ctx)
+        // Containment is not symmetric: `a` is always the container and `b` the contained. A
+        // container's rect must cover the contained's rect (`Rect::contains` is the closed
+        // test), so a contained rect poking outside proves the row false.
+        execute_null_propagating(
+            &a,
+            &b,
+            |a, b| a.contains(b),
+            Some(|ra, rb| (!ra.contains(rb)).then_some(false)),
+            ctx,
+        )
     }
 
     fn validity(
@@ -99,6 +107,10 @@ impl ScalarFnVTable for GeoContains {
     fn is_strict(&self, _: &Self::Options) -> bool {
         true
     }
+
+    fn is_fallible(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]
@@ -129,6 +141,7 @@ mod tests {
     use wkb::writer::WriteOptions;
 
     use super::GeoContains;
+    use crate::test_harness::linestring_column;
     use crate::test_harness::nullable_point_column;
     use crate::test_harness::point_column;
 
@@ -234,6 +247,20 @@ mod tests {
         assert_contains(away, point, [false; 2])
     }
 
+    /// Constant container vs a linestring column: a row whose bounding rect pokes outside the
+    /// container's rect is proven false by the rect pre-check alone; a fully inside row still
+    /// needs (and passes) the exact test.
+    #[test]
+    fn constant_container_vs_row_rect_poking_outside() -> VortexResult<()> {
+        let container = geometry_constant(&Geometry::Polygon(rect_polygon(0.0, 0.0, 4.0, 4.0)), 3)?;
+        let lines = linestring_column(vec![
+            vec![(1.0, 1.0), (3.0, 3.0)],
+            vec![(1.0, 1.0), (9.0, 1.0)],
+            vec![(5.0, 5.0), (9.0, 9.0)],
+        ])?;
+        assert_contains(container, lines, [true, false, false])
+    }
+
     /// Column vs column pairs rows: each polygon row is tested against the point row at the
     /// same position.
     #[test]

vortex-geo/src/scalar_fn/distance.rs

@@ -85,7 +85,8 @@ impl ScalarFnVTable for GeoDistance {
     ) -> VortexResult<ArrayRef> {
         let a = args.get(0)?;
         let b = args.get(1)?;
-        execute_null_propagating(&a, &b, |x, y| Euclidean.distance(x, y), ctx)
+        // Distance is a value, not a verdict: no bounding-rect test can decide it.
+        execute_null_propagating(&a, &b, |x, y| Euclidean.distance(x, y), None, ctx)
     }
 
     fn validity(
@@ -99,6 +100,10 @@ impl ScalarFnVTable for GeoDistance {
     fn is_strict(&self, _: &Self::Options) -> bool {
         true
     }
+
+    fn is_fallible(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]
@@ -123,6 +128,7 @@ mod tests {
     use super::GeoDistance;
     use crate::test_harness::nullable_point_column;
     use crate::test_harness::point_column;
+    use crate::test_harness::polygon_column;
 
     /// A constant `Point` column of length `len`, every row at `(x, y)`.
     fn point_constant(
@@ -173,6 +179,23 @@ mod tests {
         Ok(())
     }
 
+    /// Distance passes no bounding-rect rejection: a point far outside a constant polygon's
+    /// bounding rect still gets its true distance, alongside an inside point at distance zero.
+    #[test]
+    fn distance_to_constant_polygon_is_exact() -> VortexResult<()> {
+        let session = vortex_array::array_session();
+        let mut ctx = session.create_execution_ctx();
+
+        let ring = vec![(0.0, 0.0), (4.0, 0.0), (4.0, 4.0), (0.0, 4.0), (0.0, 0.0)];
+        let single = polygon_column(vec![vec![ring]])?.execute_scalar(0, &mut ctx)?;
+        let square = ConstantArray::new(single, 2).into_array();
+        let points = point_column(vec![7.0, 2.0], vec![2.0, 2.0])?;
+        let distance = GeoDistance::try_new_array(points, square)?.into_array();
+
+        assert_eq!(distances(distance, &mut ctx)?, vec![3.0, 0.0]);
+        Ok(())
+    }
+
     /// The constant query point may be either operand; distance is symmetric.
     #[test]
     fn distance_with_constant_first_operand() -> VortexResult<()> {

vortex-geo/src/scalar_fn/envelope.rs

@@ -216,6 +216,10 @@ impl ScalarFnVTable for GeoEnvelope {
     fn is_strict(&self, _: &Self::Options) -> bool {
         true
     }
+
+    fn is_fallible(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]