Fix Scalar de/serialization with proto and outline nulls from ScalarValue (#6309)

This PR is generally just a refactoring of the `vortex-scalar` crate and
all of its dependents.

Currently, we store an opaque `InnerScalarValue` inside a `ScalarValue`,
and the `InnerScalarValue` is allowed to be `Null`.

This PR both removes outlines the `InnerScalarValue::Null` variant,
where a nullable scalar is now an `Option<ScalarValue>` (instead of just
a `ScalarValue`. This also completely removes `InnerScalarValue` in
favor of a public `ScalarValue` enum:

```rust
// Before:
pub struct ScalarValue(pub(crate) InnerScalarValue);

// After:
pub enum ScalarValue {
    // No `Null` variant!

    /// A boolean value.
    Bool(bool),
    /// A primitive numeric value.
    Primitive(PValue),
    /// A decimal value.
    Decimal(DecimalValue),
    /// A UTF-8 encoded string value.
    Utf8(BufferString),
    /// A binary (byte array) value.
    Binary(ByteBuffer),
    /// A list of potentially null scalar values.
    List(Vec<Option<ScalarValue>>),
    /// TODO?
    // Extension(ExtScalarRef), // ?
}
```

(**IMPORTANT CHANGE**) Additionally, all `Scalar`s are verified to be
sound on construction by checking that the `DType` of the `Scalar`
`is_compatible` with the `Option<&ScalarValue>` that is passed.

The stricter construction changes then mean that we need to change how
deserialization of scalars work. The protobuf format is not exactly 1-1
with our `Scalar`s (notably, it only supports 64-bit integers). This
means that we might write 8-bit integers and the round trip returns a
64-bit integer. So this PR also changes some APIs to allow us to pass a
`DType` to the statistics deserializers. TBD on if this needs to happen
in more places (`FileStatistics`?).

For reviewers: try to look over all of the diffs since a large majority
is **not** just renaming variables, they are semantic changes that I am
not super confidant is right.

## Breaks

Breaks the old `file_stats` method on `VortexFile` to return
`FileStatistics` instead of the array of `StatsSet`. We needed to do
this in order to correctly read statistics from DataFusion, specifically
in the case of schema evolution.

## TODO

Some benchmarks are failing, and also I still need to review everything
myself to make sure everything is correct. I also want to add more tests
in certain places where I'm very scared things are wrong.

---------

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

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

@@ -45,8 +45,8 @@ impl BetweenKernel for ALPVTable {
         match_each_alp_float_ptype!(array.ptype(), |F| {
             between_impl::<F>(
                 array,
-                F::try_from(lower)?,
-                F::try_from(upper)?,
+                F::try_from(&lower)?,
+                F::try_from(&upper)?,
                 nullability,
                 options,
             )

encodings/alp/src/alp/compute/compare.rs

@@ -15,7 +15,7 @@ use vortex_array::register_kernel;
 use vortex_dtype::NativePType;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
-use vortex_scalar::PrimitiveScalar;
+use vortex_error::vortex_err;
 use vortex_scalar::Scalar;
 
 use crate::ALPArray;
@@ -42,7 +42,13 @@ impl CompareKernel for ALPVTable {
         }
 
         if let Some(const_scalar) = rhs.as_constant() {
-            let pscalar = PrimitiveScalar::try_from(&const_scalar)?;
+            let pscalar = const_scalar.as_primitive_opt().ok_or_else(|| {
+                vortex_err!(
+                    "ALP Compare RHS had the wrong type {}, expected {}",
+                    const_scalar,
+                    const_scalar.dtype()
+                )
+            })?;
 
             match_each_alp_float_ptype!(pscalar.ptype(), |T| {
                 match pscalar.typed_value::<T>() {

encodings/alp/src/alp/ops.rs

@@ -22,10 +22,8 @@ impl OperationsVTable<ALPVTable> for ALPVTable {
         let encoded_val = array.encoded().scalar_at(index)?;
 
         Ok(match_each_alp_float_ptype!(array.ptype(), |T| {
-            let encoded_val: <T as ALPFloat>::ALPInt = encoded_val
-                .as_ref()
-                .try_into()
-                .vortex_expect("invalid ALPInt");
+            let encoded_val: <T as ALPFloat>::ALPInt =
+                (&encoded_val).try_into().vortex_expect("invalid ALPInt");
             Scalar::primitive(
                 <T as ALPFloat>::decode_single(encoded_val, array.exponents()),
                 array.dtype().nullability(),

encodings/alp/src/alp_rd/compute/take.rs

@@ -9,7 +9,6 @@ use vortex_array::arrays::TakeExecute;
 use vortex_array::compute::fill_null;
 use vortex_error::VortexResult;
 use vortex_scalar::Scalar;
-use vortex_scalar::ScalarValue;
 
 use crate::ALPRDArray;
 use crate::ALPRDVTable;
@@ -36,7 +35,7 @@ impl TakeExecute for ALPRDVTable {
             .transpose()?;
         let right_parts = fill_null(
             &array.right_parts().take(indices.to_array())?,
-            &Scalar::new(array.right_parts().dtype().clone(), ScalarValue::from(0)),
+            &Scalar::zero_value(array.right_parts().dtype()),
         )?;
 
         Ok(Some(

encodings/datetime-parts/src/compute/rules.rs

@@ -171,7 +171,7 @@ fn try_extract_days_constant(array: &ArrayRef) -> Option<i64> {
 fn is_constant_zero(array: &ArrayRef) -> bool {
     array
         .as_opt::<ConstantVTable>()
-        .is_some_and(|c| c.scalar().is_zero())
+        .is_some_and(|c| c.scalar().is_zero() == Some(true))
 }
 
 #[cfg(test)]

encodings/decimal-byte-parts/src/decimal_byte_parts/compute/compare.rs

@@ -46,9 +46,9 @@ impl CompareKernel for DecimalBytePartsVTable {
             .vortex_expect("checked for null in entry func");
 
         match decimal_value_wrapper_to_primitive(rhs_decimal, lhs.msp.as_primitive_typed().ptype())
-            .map(|value| Scalar::new(scalar_type.clone(), value))
         {
-            Ok(encoded_scalar) => {
+            Ok(value) => {
+                let encoded_scalar = Scalar::try_new(scalar_type, Some(value))?;
                 let encoded_const = ConstantArray::new(encoded_scalar, rhs.len());
                 compare(&lhs.msp, &encoded_const.to_array(), operator).map(Some)
             }
@@ -165,7 +165,10 @@ mod tests {
         )
         .unwrap()
         .to_array();
-        let rhs = ConstantArray::new(Scalar::new(dtype, DecimalValue::I64(400).into()), lhs.len());
+        let rhs = ConstantArray::new(
+            Scalar::try_new(dtype, Some(DecimalValue::I64(400).into())).unwrap(),
+            lhs.len(),
+        );
 
         let res = compare(lhs.as_ref(), rhs.as_ref(), Operator::Eq).unwrap();
 
@@ -215,10 +218,11 @@ mod tests {
         .to_array();
         // This cannot be converted to a i32.
         let rhs = ConstantArray::new(
-            Scalar::new(
+            Scalar::try_new(
                 dtype.clone(),
-                DecimalValue::I128(-9999999999999965304).into(),
-            ),
+                Some(DecimalValue::I128(-9999999999999965304).into()),
+            )
+            .unwrap(),
             lhs.len(),
         );
 
@@ -236,7 +240,7 @@ mod tests {
 
         // This cannot be converted to a i32.
         let rhs = ConstantArray::new(
-            Scalar::new(dtype, DecimalValue::I128(9999999999999965304).into()),
+            Scalar::try_new(dtype, Some(DecimalValue::I128(9999999999999965304).into())).unwrap(),
             lhs.len(),
         );
 

encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs

@@ -44,6 +44,7 @@ use vortex_error::vortex_bail;
 use vortex_error::vortex_ensure;
 use vortex_scalar::DecimalValue;
 use vortex_scalar::Scalar;
+use vortex_scalar::ScalarValue;
 use vortex_session::VortexSession;
 
 use crate::decimal_byte_parts::compute::kernel::PARENT_KERNELS;
@@ -285,10 +286,10 @@ impl OperationsVTable<DecimalBytePartsVTable> for DecimalBytePartsVTable {
         let primitive_scalar = scalar.as_primitive();
         // TODO(joe): extend this to support multiple parts.
         let value = primitive_scalar.as_::<i64>().vortex_expect("non-null");
-        Ok(Scalar::new(
+        Scalar::try_new(
             array.dtype.clone(),
-            DecimalValue::I64(value).into(),
-        ))
+            Some(ScalarValue::Decimal(DecimalValue::I64(value))),
+        )
     }
 }
 
@@ -319,6 +320,7 @@ mod tests {
     use vortex_dtype::Nullability;
     use vortex_scalar::DecimalValue;
     use vortex_scalar::Scalar;
+    use vortex_scalar::ScalarValue;
 
     use crate::DecimalBytePartsArray;
 
@@ -339,11 +341,15 @@ mod tests {
 
         assert_eq!(Scalar::null(dtype.clone()), array.scalar_at(0).unwrap());
         assert_eq!(
-            Scalar::new(dtype.clone(), DecimalValue::I64(200).into()),
+            Scalar::try_new(
+                dtype.clone(),
+                Some(ScalarValue::Decimal(DecimalValue::I64(200)))
+            )
+            .unwrap(),
             array.scalar_at(1).unwrap()
         );
         assert_eq!(
-            Scalar::new(dtype, DecimalValue::I64(400).into()),
+            Scalar::try_new(dtype, Some(ScalarValue::Decimal(DecimalValue::I64(400)))).unwrap(),
             array.scalar_at(2).unwrap()
         );
     }

encodings/fastlanes/src/bitpacking/array/bitpack_decompress.rs

@@ -262,7 +262,7 @@ mod tests {
             .iter()
             .enumerate()
             .for_each(|(i, v)| {
-                let scalar: u16 = unpack_single(&compressed, i).try_into().unwrap();
+                let scalar: u16 = (&unpack_single(&compressed, i)).try_into().unwrap();
                 assert_eq!(scalar, *v);
             });
     }

encodings/fastlanes/src/for/array/for_compress.rs

@@ -175,10 +175,7 @@ mod test {
             .iter()
             .enumerate()
             .for_each(|(i, v)| {
-                assert_eq!(
-                    *v,
-                    i8::try_from(compressed.scalar_at(i).unwrap().as_ref()).unwrap()
-                );
+                assert_eq!(*v, i8::try_from(&compressed.scalar_at(i).unwrap()).unwrap());
             });
         assert_arrays_eq!(decompressed, array);
         Ok(())

encodings/fastlanes/src/for/compute/compare.rs

@@ -19,7 +19,6 @@ use vortex_error::VortexError;
 use vortex_error::VortexExpect as _;
 use vortex_error::VortexResult;
 use vortex_scalar::PValue;
-use vortex_scalar::PrimitiveScalar;
 use vortex_scalar::Scalar;
 
 use crate::FoRArray;
@@ -33,7 +32,7 @@ impl CompareKernel for FoRVTable {
         operator: Operator,
     ) -> VortexResult<Option<ArrayRef>> {
         if let Some(constant) = rhs.as_constant()
-            && let Ok(constant) = PrimitiveScalar::try_from(&constant)
+            && let Some(constant) = constant.as_primitive_opt()
         {
             match_each_integer_ptype!(constant.ptype(), |T| {
                 return compare_constant(