refactor(vortex-geo): address null-propagation review

- Add validity() + is_null_sensitive()=false vtable overrides to the three
  geo kernels so the planner can derive the output null mask symbolically
  (mask pushdown), matching vortex-tensor's binary functions. Shared via a
  binary_result_validity helper. is_fallible is left at the default since
  geometry decoding can genuinely fail.
- Cover the two-column paths for the asymmetric ST_Contains: nulls on either
  side, plus the all-null-column and empty-combined-mask early returns.
- Drop the unreachable AllOr::None arm in build_array; callers short-circuit
  the fully-null case to all_null_array.
- Document that operands may differ in geometry type, and that the nullable
  test column's validity mirrors its storage.

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

Author: HarukiMoriarty

vortex-geo/src/extension/mod.rs

@@ -73,8 +73,10 @@ 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. Nullable operands are allowed; the kernels propagate nulls (a null
-/// geometry input yields a null result) rather than decoding null rows.
+/// 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!(

vortex-geo/src/scalar_fn/contains.rs

@@ -9,6 +9,7 @@ use vortex_array::ExecutionCtx;
 use vortex_array::arrays::ScalarFnArray;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
+use vortex_array::expr::Expression;
 use vortex_array::scalar_fn::Arity;
 use vortex_array::scalar_fn::ChildName;
 use vortex_array::scalar_fn::EmptyOptions;
@@ -21,6 +22,7 @@ use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
 use crate::extension::validate_geometry_operands;
+use crate::scalar_fn::binary_result_validity;
 use crate::scalar_fn::execute_null_propagating;
 
 /// OGC `ST_Contains` between two native geometry operands, each a column or a constant
@@ -85,6 +87,18 @@ impl ScalarFnVTable for GeoContains {
         // Containment is not symmetric: `a` is always the container and `b` the contained.
         execute_null_propagating(&a, &b, |a, b| a.contains(b), ctx)
     }
+
+    fn validity(
+        &self,
+        _: &Self::Options,
+        expression: &Expression,
+    ) -> VortexResult<Option<Expression>> {
+        binary_result_validity(expression)
+    }
+
+    fn is_null_sensitive(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]
@@ -285,6 +299,71 @@ mod tests {
         Ok(())
     }
 
+    /// Both operands nullable columns: containment (asymmetric) is null wherever either the
+    /// container or the contained row is null, and computed on the rows valid in both.
+    #[test]
+    fn contains_propagates_column_pair_nulls() -> VortexResult<()> {
+        let session = vortex_array::array_session();
+        let mut ctx = session.create_execution_ctx();
+
+        // A point contains another point only when they are equal.
+        let container = nullable_point_column(vec![
+            Some((1.0, 1.0)),
+            None,
+            Some((2.0, 2.0)),
+            Some((3.0, 3.0)),
+        ])?;
+        let contained = nullable_point_column(vec![
+            Some((1.0, 1.0)),
+            Some((5.0, 5.0)),
+            None,
+            Some((4.0, 4.0)),
+        ])?;
+        let contains = GeoContains::try_new_array(container, contained)?.into_array();
+
+        let expected = BoolArray::new(
+            BitBuffer::from_iter([true, false, false, false]),
+            Validity::from_iter([true, false, false, true]),
+        )
+        .into_array();
+        assert_arrays_eq!(contains, expected, &mut ctx);
+        Ok(())
+    }
+
+    /// An entirely-null geometry column yields an all-null output (the fully-null fast path in
+    /// `eval_column`).
+    #[test]
+    fn contains_all_null_column_is_all_null() -> VortexResult<()> {
+        let session = vortex_array::array_session();
+        let mut ctx = session.create_execution_ctx();
+
+        let container = geometry_constant(&Geometry::Polygon(rect_polygon(0.0, 0.0, 4.0, 4.0)), 2)?;
+        let points = nullable_point_column(vec![None, None])?;
+        let contains = GeoContains::try_new_array(container, points)?.into_array();
+
+        let expected =
+            BoolArray::new(BitBuffer::from_iter([false, false]), Validity::AllInvalid).into_array();
+        assert_arrays_eq!(contains, expected, &mut ctx);
+        Ok(())
+    }
+
+    /// Two nullable columns whose nulls never line up: the combined mask is empty, so the output
+    /// is all null (the `valid.all_false()` early return in `eval_column_pair`).
+    #[test]
+    fn contains_column_pair_all_null() -> VortexResult<()> {
+        let session = vortex_array::array_session();
+        let mut ctx = session.create_execution_ctx();
+
+        let container = nullable_point_column(vec![Some((1.0, 1.0)), None])?;
+        let contained = nullable_point_column(vec![None, Some((2.0, 2.0))])?;
+        let contains = GeoContains::try_new_array(container, contained)?.into_array();
+
+        let expected =
+            BoolArray::new(BitBuffer::from_iter([false, false]), Validity::AllInvalid).into_array();
+        assert_arrays_eq!(contains, expected, &mut ctx);
+        Ok(())
+    }
+
     /// A non-geometry operand dtype is rejected up front, before execution.
     #[test]
     fn non_geometry_operand_is_rejected() -> VortexResult<()> {

vortex-geo/src/scalar_fn/distance.rs

@@ -11,6 +11,7 @@ use vortex_array::arrays::ScalarFnArray;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
+use vortex_array::expr::Expression;
 use vortex_array::scalar_fn::Arity;
 use vortex_array::scalar_fn::ChildName;
 use vortex_array::scalar_fn::EmptyOptions;
@@ -23,6 +24,7 @@ use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
 use crate::extension::validate_geometry_operands;
+use crate::scalar_fn::binary_result_validity;
 use crate::scalar_fn::execute_null_propagating;
 
 /// Planar (Euclidean) `ST_Distance` (no geodesic correction) between two native geometry
@@ -85,6 +87,18 @@ impl ScalarFnVTable for GeoDistance {
         let b = args.get(1)?;
         execute_null_propagating(&a, &b, |x, y| Euclidean.distance(x, y), ctx)
     }
+
+    fn validity(
+        &self,
+        _: &Self::Options,
+        expression: &Expression,
+    ) -> VortexResult<Option<Expression>> {
+        binary_result_validity(expression)
+    }
+
+    fn is_null_sensitive(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]

vortex-geo/src/scalar_fn/intersects.rs

@@ -9,6 +9,7 @@ use vortex_array::ExecutionCtx;
 use vortex_array::arrays::ScalarFnArray;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
+use vortex_array::expr::Expression;
 use vortex_array::scalar_fn::Arity;
 use vortex_array::scalar_fn::ChildName;
 use vortex_array::scalar_fn::EmptyOptions;
@@ -21,6 +22,7 @@ use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
 use crate::extension::validate_geometry_operands;
+use crate::scalar_fn::binary_result_validity;
 use crate::scalar_fn::execute_null_propagating;
 
 /// OGC `ST_Intersects` (not disjoint; boundary contact counts) between two native geometry
@@ -83,6 +85,18 @@ impl ScalarFnVTable for GeoIntersects {
         let b = args.get(1)?;
         execute_null_propagating(&a, &b, |x, y| x.intersects(y), ctx)
     }
+
+    fn validity(
+        &self,
+        _: &Self::Options,
+        expression: &Expression,
+    ) -> VortexResult<Option<Expression>> {
+        binary_result_validity(expression)
+    }
+
+    fn is_null_sensitive(&self, _: &Self::Options) -> bool {
+        false
+    }
 }
 
 #[cfg(test)]

vortex-geo/src/scalar_fn/mod.rs

@@ -18,6 +18,8 @@ use vortex_array::arrays::PrimitiveArray;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::PType;
+use vortex_array::expr::Expression;
+use vortex_array::expr::and;
 use vortex_array::scalar::Scalar;
 use vortex_array::validity::Validity;
 use vortex_buffer::BitBuffer;
@@ -71,8 +73,6 @@ impl GeoOutput for f64 {
         match valid.indices() {
             // No nulls: `values` already lines up one-to-one with the rows.
             AllOr::All => PrimitiveArray::new(values, validity).into_array(),
-            // No valid rows: the whole output is null.
-            AllOr::None => PrimitiveArray::new(vec![0.0f64; len], validity).into_array(),
             // Some nulls: scatter each computed value back to the row it came from.
             AllOr::Some(rows) => {
                 let mut data = vec![0.0f64; len];
@@ -81,6 +81,11 @@ impl GeoOutput for f64 {
                 }
                 PrimitiveArray::new(data, validity).into_array()
             }
+            // The all-null case never reaches here, `eval_column` / `eval_column_pair` return
+            // `all_null_array` when the combined mask is empty.
+            AllOr::None => {
+                unreachable!("empty masks are handled by all_null_array in eval_column(_pair)")
+            }
         }
     }
 }
@@ -104,10 +109,6 @@ impl GeoOutput for bool {
         match valid.indices() {
             // No nulls: `values` already lines up one-to-one with the rows.
             AllOr::All => BoolArray::new(BitBuffer::from_iter(values), validity).into_array(),
-            // No valid rows: the whole output is null.
-            AllOr::None => {
-                BoolArray::new(BitBuffer::from_iter(vec![false; len]), validity).into_array()
-            }
             // Some nulls: scatter each computed value back to the row it came from.
             AllOr::Some(rows) => {
                 let mut data = vec![false; len];
@@ -116,6 +117,11 @@ impl GeoOutput for bool {
                 }
                 BoolArray::new(BitBuffer::from_iter(data), validity).into_array()
             }
+            // The all-null case never reaches here — `eval_column` / `eval_column_pair` return
+            // `all_null_array` when the combined mask is empty.
+            AllOr::None => {
+                unreachable!("empty masks are handled by all_null_array in eval_column(_pair)")
+            }
         }
     }
 }
@@ -125,6 +131,16 @@ fn all_null_array<T: GeoOutput>(len: usize) -> ArrayRef {
     ConstantArray::new(Scalar::null(T::null_dtype()), len).into_array()
 }
 
+/// The validity expression for a binary geo kernel: the result is null iff either operand is
+/// null. Returning this lets the planner derive the output's null mask symbolically (mask
+/// pushdown) instead of executing the whole kernel.
+pub(crate) fn binary_result_validity(expression: &Expression) -> VortexResult<Option<Expression>> {
+    Ok(Some(and(
+        expression.child(0).validity()?,
+        expression.child(1).validity()?,
+    )))
+}
+
 /// Run a binary geo kernel over operands `a` and `b`, each a column or a constant literal.
 ///
 /// The output is null wherever either operand is null, and its type is nullable if either operand