feat(vortex-geo): null-propagating scalar functions (#8803)

## What

Make the binary geo scalar functions (`ST_Distance`, `ST_Intersects`,
`ST_Contains`) null-propagating. A nullable geometry operand is now
allowed, and any row whose geometry input is null yields a `NULL`
result.

## Why

Spiral needs the geo predicates to accept nullable inputs to run
SpatialBench.

## How

A geo kernel decodes each operand into a `geo_types` geometry, and a
null row has no geometry to decode, so it can't compute over every row
and mask the nulls afterwards the way the numeric kernels do.

Instead, the shared `execute_null_propagating` dispatch drops the null
rows, computes only the rows valid in both operands, and scatters the
results back under the combined null mask. `return_dtype` is nullable
iff an operand is, and `validate_geometry_operands` no longer rejects
nullable operands.

---------

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

Author: HarukiMoriarty

Cargo.lock

@@ -24,7 +24,9 @@ geoarrow-cast = { workspace = true }
 prost = { workspace = true }
 vortex-array = { workspace = true }
 vortex-arrow = { workspace = true }
+vortex-buffer = { workspace = true }
 vortex-error = { workspace = true }
+vortex-mask = { workspace = true }
 vortex-session = { workspace = true }
 wkb = { workspace = true }
 

vortex-geo/Cargo.toml

@@ -202,8 +202,20 @@ impl AggregateFnVTable for GeometryAabb {
             Columnar::Canonical(canonical) => canonical.clone().into_array(),
             Columnar::Constant(constant) => constant.clone().into_array(),
         };
-        // Min/max the raw x/y buffers directly - cheap, and avoids `to_geometry`'s panic on empty
-        // points (which decoding each geometry would hit).
+        // Drop null rows before reading coordinates: a null geometry's storage holds placeholder
+        // coordinates (e.g. `(0, 0)`) that would otherwise widen the zone box and drag it toward
+        // the origin. `filter` collapses an all-true mask back to the input, so a null-free batch
+        // passes through unchanged.
+        //
+        // TODO(perf): on nullable data this `filter` compacts the whole column even for a single
+        // null before we min/max it. A validity-aware min/max straight over the raw x/y buffers
+        // would skip that copy. Left as-is for now: this is a write-time zone stat, and the common
+        // non-nullable case already costs nothing (the all-true mask makes `filter` a no-op).
+        let valid = array.validity()?.execute_mask(array.len(), ctx)?;
+        let array = array.filter(valid)?;
+        // Null rows are gone, so every coordinate below belongs to a present geometry — the
+        // `unmasked_field_by_name` reads are therefore safe. Min/max the raw x/y buffers directly:
+        // cheap, and avoids `to_geometry`'s panic on empty points (which decoding would hit).
         let coords = flatten_coordinates(&array, ctx)?;
         let xs = coords
             .unmasked_field_by_name("x")?
@@ -254,6 +266,7 @@ mod tests {
     use crate::test_harness::multilinestring_column;
     use crate::test_harness::multipoint_column;
     use crate::test_harness::multipolygon_column;
+    use crate::test_harness::nullable_point_column;
     use crate::test_harness::point_column;
     use crate::test_harness::polygon_column;
 
@@ -304,6 +317,23 @@ mod tests {
         Ok(())
     }
 
+    /// Null rows contribute no extent: their placeholder coordinates must not widen the box toward
+    /// the origin (matching min/max, which skip nulls).
+    #[test]
+    fn null_rows_do_not_widen_aabb() -> VortexResult<()> {
+        let session = vortex_array::array_session();
+        let mut ctx = session.create_execution_ctx();
+
+        // The valid points sit far from the origin; the null row's storage placeholder is (0, 0).
+        let column = nullable_point_column(vec![Some((5.0, 6.0)), None, Some((7.0, 8.0))])?;
+        let mut acc = Accumulator::try_new(GeometryAabb, EmptyOptions, column.dtype().clone())?;
+        acc.accumulate(&column, &mut ctx)?;
+
+        // The box covers only the two valid points, never (0, 0).
+        assert_eq!(aabb(&acc.finish()?)?, (5.0, 6.0, 7.0, 8.0));
+        Ok(())
+    }
+
     /// The AABB of a Polygon column unions every ring vertex - exercising the `List<List<Struct>>`
     /// unwrap, not just the bare Point struct.
     #[test]

vortex-geo/src/aggregate_fn/aabb.rs

@@ -72,18 +72,17 @@ pub(crate) fn is_native_geometry(dtype: &DType) -> bool {
     })
 }
 
-/// Validate the operands of a geo scalar function: each must be a native geometry type (so the
-/// kernel can decode it) and non-nullable (geometry arrays never carry nulls).
+/// Validate the operands of a geo scalar function: each must be a native geometry type so the
+/// kernel can decode it. The two operands need not share a geometry type — e.g. a `Point` against
+/// a `Polygon` is valid, since distance/containment/intersection across types is meaningful.
+/// Nullable operands are allowed; the kernels propagate nulls (a null geometry input yields a null
+/// result) rather than decoding null rows.
 pub(crate) fn validate_geometry_operands(dtypes: &[DType]) -> VortexResult<()> {
     for dtype in dtypes {
         vortex_ensure!(
             is_native_geometry(dtype),
             "geo: operand {dtype} is not a native geometry type"
         );
-        vortex_ensure!(
-            !dtype.is_nullable(),
-            "geo: nullable operand {dtype} is unsupported"
-        );
     }
     Ok(())
 }

vortex-geo/src/extension/mod.rs

@@ -175,6 +175,22 @@ mod tests {
         GeoDistancePrune.falsify(&predicate, &StatsRewriteCtx::new(&session, &scope))
     }
 
+    /// A null geometry literal (`ST_Distance(geom, NULL) <= r`) declines cleanly instead of
+    /// erroring in the stats rewrite: the all-null predicate can never prune.
+    #[test]
+    fn null_literal_is_not_pruned() -> VortexResult<()> {
+        let session = geo_session();
+
+        let scope = point_column(vec![0.0], vec![0.0])?.dtype().clone();
+        let null_query = Scalar::null(scope.as_nullable());
+        let distance = GeoDistance.new_expr(EmptyOptions, [root(), lit(null_query)]);
+        let predicate = Binary.new_expr(Operator::Lte, [distance, lit(0.5f64)]);
+
+        let ctx = StatsRewriteCtx::new(&session, &scope);
+        assert!(GeoDistancePrune.falsify(&predicate, &ctx)?.is_none());
+        Ok(())
+    }
+
     /// All four distance comparisons prune (`<=`/`<` via min-distance, `>=`/`>` via max-distance);
     /// `==`/`!=` are left to the scan.
     #[rstest]

vortex-geo/src/prune/distance.rs

@@ -59,6 +59,7 @@ mod tests {
     use vortex_array::expr::Expression;
     use vortex_array::expr::lit;
     use vortex_array::expr::root;
+    use vortex_array::scalar::Scalar;
     use vortex_array::scalar_fn::EmptyOptions;
     use vortex_array::scalar_fn::ScalarFnVTableExt;
     use vortex_array::stats::rewrite::StatsRewriteCtx;
@@ -114,6 +115,21 @@ mod tests {
         Ok(())
     }
 
+    /// A null geometry literal (`ST_Intersects(geom, NULL)`) declines cleanly instead of erroring
+    /// in the stats rewrite: the all-null predicate can never prune.
+    #[test]
+    fn null_literal_is_not_pruned() -> VortexResult<()> {
+        let session = geo_session();
+
+        let scope = point_column(vec![0.0], vec![0.0])?.dtype().clone();
+        let null_query = Scalar::null(scope.as_nullable());
+        let predicate = GeoIntersects.new_expr(EmptyOptions, [root(), lit(null_query)]);
+
+        let ctx = StatsRewriteCtx::new(&session, &scope);
+        assert!(GeoIntersectsPrune.falsify(&predicate, &ctx)?.is_none());
+        Ok(())
+    }
+
     /// End-to-end: a zone strictly separated from the query is skipped; zones containing or merely
     /// touching the query must scan, touching geometries intersect under OGC semantics.
     #[test]

vortex-geo/src/prune/intersects.rs

@@ -82,6 +82,10 @@ fn geometry_and_constant<'a>(
 /// Prove claims against this box rather than the constant itself: the constant lies inside it,
 /// so whatever holds for the box holds for the geometry.
 fn query_aabb(constant: &Scalar, ctx: &StatsRewriteCtx<'_>) -> VortexResult<Option<GeoRect<f64>>> {
+    // A null geometry literal has no extent to prove against, so it can never prune.
+    if constant.is_null() {
+        return Ok(None);
+    }
     // Decoding the constant into a concrete geometry runs through the compute stack, which needs
     // an execution context.
     let mut exec = ctx.session().create_execution_ctx();