Outline nulls from ScalarValue and fix de/serialization with Proto

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().clone()),
         )?;
 
         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().value().is_some_and(|value| value.is_zero()))
 }
 
 #[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/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/vtable/mod.rs

@@ -2,13 +2,10 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::fmt::Debug;
-use std::fmt::Formatter;
 
 use vortex_array::ArrayRef;
-use vortex_array::DeserializeMetadata;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
-use vortex_array::SerializeMetadata;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::serde::ArrayChildren;
 use vortex_array::vtable;
@@ -41,7 +38,9 @@ vtable!(FoR);
 impl VTable for FoRVTable {
     type Array = FoRArray;
 
-    type Metadata = ScalarValueMetadata;
+    // TODO(connor): This should really be a `Scalar` but we need to deprecate `deserialize` for the
+    // `build` method.
+    type Metadata = Vec<u8>;
 
     type ArrayVTable = Self;
     type OperationsVTable = Self;
@@ -67,14 +66,16 @@ impl VTable for FoRVTable {
         Ok(())
     }
 
+    // TODO(connor): DON'T TOUCH THIS UNLESS YOU KNOW WHAT YOU ARE DOING!!!
     fn metadata(array: &FoRArray) -> VortexResult<Self::Metadata> {
-        Ok(ScalarValueMetadata(
-            array.reference_scalar().value().clone(),
+        Ok(ScalarValue::to_proto_bytes(
+            array.reference_scalar().value(),
         ))
     }
 
+    // TODO(connor): DON'T TOUCH THIS UNLESS YOU KNOW WHAT YOU ARE DOING!!!
     fn serialize(metadata: Self::Metadata) -> VortexResult<Option<Vec<u8>>> {
-        Ok(Some(metadata.serialize()))
+        Ok(Some(metadata))
     }
 
     fn deserialize(
@@ -83,7 +84,7 @@ impl VTable for FoRVTable {
         _len: usize,
         _session: &VortexSession,
     ) -> VortexResult<Self::Metadata> {
-        ScalarValueMetadata::deserialize(bytes)
+        Ok(bytes.to_vec())
     }
 
     fn build(
@@ -101,7 +102,8 @@ impl VTable for FoRVTable {
         }
 
         let encoded = children.get(0, dtype, len)?;
-        let reference = Scalar::new(dtype.clone(), metadata.0.clone());
+        let scalar_value = ScalarValue::from_proto_bytes(metadata, dtype)?;
+        let reference = Scalar::try_new(dtype.clone(), scalar_value)?;
 
         FoRArray::try_new(encoded, reference)
     }
@@ -134,27 +136,3 @@ pub struct FoRVTable;
 impl FoRVTable {
     pub const ID: ArrayId = ArrayId::new_ref("fastlanes.for");
 }
-
-#[derive(Clone)]
-pub struct ScalarValueMetadata(pub ScalarValue);
-
-impl SerializeMetadata for ScalarValueMetadata {
-    fn serialize(self) -> Vec<u8> {
-        self.0.to_protobytes()
-    }
-}
-
-impl DeserializeMetadata for ScalarValueMetadata {
-    type Output = ScalarValueMetadata;
-
-    fn deserialize(metadata: &[u8]) -> VortexResult<Self::Output> {
-        let scalar_value = ScalarValue::from_protobytes(metadata)?;
-        Ok(ScalarValueMetadata(scalar_value))
-    }
-}
-
-impl Debug for ScalarValueMetadata {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", &self.0)
-    }
-}

encodings/fastlanes/src/rle/vtable/operations.rs

@@ -27,7 +27,7 @@ impl OperationsVTable<RLEVTable> for RLEVTable {
             .values()
             .scalar_at(value_idx_offset + chunk_relative_idx)?;
 
-        Ok(Scalar::new(array.dtype().clone(), scalar.into_value()))
+        Scalar::try_new(array.dtype().clone(), scalar.into_value())
     }
 }
 

encodings/fsst/src/compute/compare.rs

@@ -104,16 +104,16 @@ fn compare_fsst_constant(
         DType::Binary(_) => {
             let value = right
                 .as_binary()
-                .value()
+                .to_value()
                 .vortex_expect("Expected non-null scalar");
             ByteBuffer::from(compressor.compress(value.as_slice()))
         }
         _ => unreachable!("FSSTArray can only have string or binary data type"),
     };
 
-    let encoded_scalar = Scalar::new(
-        DType::Binary(left.dtype().nullability() | right.dtype().nullability()),
-        encoded_buffer.into(),
+    let encoded_scalar = Scalar::binary(
+        encoded_buffer,
+        left.dtype().nullability() | right.dtype().nullability(),
     );
 
     let rhs = ConstantArray::new(encoded_scalar, left.len());

encodings/fsst/src/compute/mod.rs

@@ -16,7 +16,6 @@ use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 use vortex_scalar::Scalar;
-use vortex_scalar::ScalarValue;
 
 use crate::FSSTArray;
 use crate::FSSTVTable;
@@ -41,10 +40,7 @@ impl TakeExecute for FSSTVTable {
                     .map_err(|_| vortex_err!("take for codes must return varbin array"))?,
                 fill_null(
                     &array.uncompressed_lengths().take(indices.to_array())?,
-                    &Scalar::new(
-                        array.uncompressed_lengths_dtype().clone(),
-                        ScalarValue::from(0),
-                    ),
+                    &Scalar::zero_value(&array.uncompressed_lengths_dtype().clone()),
                 )?,
             )?
             .into_array(),

encodings/fsst/src/ops.rs

@@ -14,7 +14,7 @@ use crate::FSSTVTable;
 impl OperationsVTable<FSSTVTable> for FSSTVTable {
     fn scalar_at(array: &FSSTArray, index: usize) -> VortexResult<Scalar> {
         let compressed = array.codes().scalar_at(index)?;
-        let binary_datum = compressed.as_binary().value().vortex_expect("non-null");
+        let binary_datum = compressed.as_binary().to_value().vortex_expect("non-null");
 
         let decoded_buffer =
             ByteBuffer::from(array.decompressor().decompress(binary_datum.as_slice()));

encodings/runend/src/array.rs

@@ -222,13 +222,13 @@ impl RunEndArray {
 
         // Validate the offset and length are valid for the given ends and values
         if offset != 0 && length != 0 {
-            let first_run_end: usize = ends.scalar_at(0)?.as_ref().try_into()?;
+            let first_run_end: usize = usize::try_from(&ends.scalar_at(0)?)?;
             if first_run_end <= offset {
                 vortex_bail!("First run end {first_run_end} must be bigger than offset {offset}");
             }
         }
 
-        let last_run_end: usize = ends.scalar_at(ends.len() - 1)?.as_ref().try_into()?;
+        let last_run_end: usize = usize::try_from(&ends.scalar_at(ends.len() - 1)?)?;
         let min_required_end = offset + length;
         if last_run_end < min_required_end {
             vortex_bail!("Last run end {last_run_end} must be >= offset+length {min_required_end}");
@@ -302,7 +302,7 @@ impl RunEndArray {
         let length: usize = if ends.is_empty() {
             0
         } else {
-            ends.scalar_at(ends.len() - 1)?.as_ref().try_into()?
+            usize::try_from(&ends.scalar_at(ends.len() - 1)?)?
         };
 
         Self::try_new_offset_length(ends, values, 0, length)

encodings/runend/src/compute/cast.rs

@@ -76,27 +76,19 @@ mod tests {
         // RunEnd encoding should expand to [100, 100, 100, 200, 200, 100, 100, 100, 300, 300]
         assert_eq!(decoded.len(), 10);
         assert_eq!(
-            TryInto::<i64>::try_into(decoded.scalar_at(0).unwrap().as_ref())
-                .ok()
-                .unwrap(),
+            TryInto::<i64>::try_into(decoded.scalar_at(0).unwrap()).unwrap(),
             100i64
         );
         assert_eq!(
-            TryInto::<i64>::try_into(decoded.scalar_at(3).unwrap().as_ref())
-                .ok()
-                .unwrap(),
+            TryInto::<i64>::try_into(decoded.scalar_at(3).unwrap()).unwrap(),
             200i64
         );
         assert_eq!(
-            TryInto::<i64>::try_into(decoded.scalar_at(5).unwrap().as_ref())
-                .ok()
-                .unwrap(),
+            TryInto::<i64>::try_into(decoded.scalar_at(5).unwrap()).unwrap(),
             100i64
         );
         assert_eq!(
-            TryInto::<i64>::try_into(decoded.scalar_at(8).unwrap().as_ref())
-                .ok()
-                .unwrap(),
+            TryInto::<i64>::try_into(decoded.scalar_at(8).unwrap()).unwrap(),
             300i64
         );
     }