Correctly handle Nan/Inf comparison in ALP between reduce (#8126)

Found in #8105. ALP comparisons to NaN/Inf can be trivially folded since
those values must exist in patches

---------

Signed-off-by: Robert Kruszewski <github@robertk.io>

Author: robert3005

encodings/alp/src/alp/compute/between.rs

@@ -3,6 +3,7 @@
 
 use std::fmt::Debug;
 
+use num_traits::Bounded;
 use vortex_array::ArrayRef;
 use vortex_array::ArrayView;
 use vortex_array::IntoArray;
@@ -19,6 +20,7 @@ use vortex_error::VortexResult;
 
 use crate::ALP;
 use crate::ALPFloat;
+use crate::Exponents;
 use crate::alp::array::ALPArrayExt;
 use crate::alp::array::ALPArraySlotsExt;
 use crate::match_each_alp_float_ptype;
@@ -72,14 +74,18 @@ where
     // and we encode into value below enc_below(value) < value < x, in which case the comparison
     // becomes enc(value) < x. See `alp_scalar_compare` for more details.
     // note that if the value doesn't encode than value != x, so must use strict comparison.
-    let (lower_enc, lower_strict) = T::encode_single(lower, exponents)
-        .map(|x| (x, options.lower_strict))
-        .unwrap_or_else(|| (T::encode_below(lower, exponents), StrictComparison::Strict));
+    let Some((lower_enc, lower_strict)) =
+        encode_lower_bound::<T>(lower, exponents, options.lower_strict)
+    else {
+        return Ok(ConstantArray::new(Scalar::bool(false, nullability), array.len()).into_array());
+    };
 
     // the upper value `x { < | <= } value` similarly encodes or `x < value < enc_above(value())`
-    let (upper_enc, upper_strict) = T::encode_single(upper, exponents)
-        .map(|x| (x, options.upper_strict))
-        .unwrap_or_else(|| (T::encode_above(upper, exponents), StrictComparison::Strict));
+    let Some((upper_enc, upper_strict)) =
+        encode_upper_bound::<T>(upper, exponents, options.upper_strict)
+    else {
+        return Ok(ConstantArray::new(Scalar::bool(false, nullability), array.len()).into_array());
+    };
 
     let options = BetweenOptions {
         lower_strict,
@@ -93,14 +99,50 @@ where
     )
 }
 
+fn encode_lower_bound<T: ALPFloat>(
+    lower: T,
+    exponents: Exponents,
+    strict: StrictComparison,
+) -> Option<(T::ALPInt, StrictComparison)> {
+    if NativePType::is_nan(lower) || NativePType::is_infinite(lower) {
+        return NativePType::is_lt(lower, T::zero())
+            .then_some((T::ALPInt::min_value(), StrictComparison::NonStrict));
+    }
+
+    Some(
+        T::encode_single(lower, exponents)
+            .map(|x| (x, strict))
+            .unwrap_or_else(|| (T::encode_below(lower, exponents), StrictComparison::Strict)),
+    )
+}
+
+fn encode_upper_bound<T: ALPFloat>(
+    upper: T,
+    exponents: Exponents,
+    strict: StrictComparison,
+) -> Option<(T::ALPInt, StrictComparison)> {
+    if NativePType::is_nan(upper) || NativePType::is_infinite(upper) {
+        return NativePType::is_gt(upper, T::zero())
+            .then_some((T::ALPInt::max_value(), StrictComparison::NonStrict));
+    }
+
+    Some(
+        T::encode_single(upper, exponents)
+            .map(|x| (x, strict))
+            .unwrap_or_else(|| (T::encode_above(upper, exponents), StrictComparison::Strict)),
+    )
+}
+
 #[cfg(test)]
 mod tests {
+    use vortex_array::IntoArray;
     use vortex_array::LEGACY_SESSION;
     use vortex_array::VortexSessionExecute;
-    use vortex_array::arrays::BoolArray;
+    use vortex_array::arrays::ConstantArray;
     use vortex_array::arrays::PrimitiveArray;
     use vortex_array::assert_arrays_eq;
     use vortex_array::dtype::Nullability;
+    use vortex_array::scalar::Scalar;
     use vortex_array::scalar_fn::fns::between::BetweenOptions;
     use vortex_array::scalar_fn::fns::between::StrictComparison;
 
@@ -118,7 +160,11 @@ mod tests {
     ) {
         let res =
             between_impl(arr.as_view(), lower, upper, Nullability::Nullable, options).unwrap();
-        assert_arrays_eq!(res, BoolArray::from_iter([Some(expected)]));
+        assert_arrays_eq!(
+            res,
+            ConstantArray::new(Scalar::bool(expected, res.dtype().nullability()), arr.len())
+                .into_array()
+        );
     }
 
     #[test]
@@ -188,4 +234,27 @@ mod tests {
             true,
         );
     }
+
+    #[test]
+    fn non_finite_bounds_use_total_order() {
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        let array = PrimitiveArray::from_iter([1.234f32; 10]);
+        let encoded = alp_encode(array.as_view(), None, &mut ctx).unwrap();
+        assert!(encoded.patches().is_none());
+
+        let options = BetweenOptions {
+            lower_strict: StrictComparison::NonStrict,
+            upper_strict: StrictComparison::Strict,
+        };
+
+        assert_between(&encoded, f32::from_bits(0xffffff5e), 2.0, &options, true);
+        assert_between(&encoded, f32::NAN, 2.0, &options, false);
+        assert_between(&encoded, f32::NEG_INFINITY, 2.0, &options, true);
+        assert_between(&encoded, f32::INFINITY, 2.0, &options, false);
+
+        assert_between(&encoded, 0.0, f32::NAN, &options, true);
+        assert_between(&encoded, 0.0, f32::from_bits(0xffffff5e), &options, false);
+        assert_between(&encoded, 0.0, f32::INFINITY, &options, true);
+        assert_between(&encoded, 0.0, f32::NEG_INFINITY, &options, false);
+    }
 }