fix: yo dawg, I heard you like inlining (#5363)

i heard you like inlining.

so i `#[inlined]` your `Inlined` so you can get inlining for your
`Inlined`s

---------

Signed-off-by: Andrew Duffy <[email protected]>

Author: a10y

encodings/zstd/src/array.rs

@@ -454,10 +454,32 @@ impl ZstdArray {
 
         let decompressed = decompressed.freeze();
         // Last, we slice the exact values requested out of the decompressed data.
-        let slice_validity = self
+        let mut slice_validity = self
             .unsliced_validity
             .slice(self.slice_start..self.slice_stop);
 
+        // NOTE: this block handles setting the output type when the validity and DType disagree.
+        //
+        // ZSTD is a compact block compressor, meaning that null values are not stored inline in
+        // the data frames. A ZSTD Array that was initialized must always hold onto its full
+        // validity bitmap, even if sliced to only include non-null values.
+        //
+        // We ensure that the validity of the decompressed array ALWAYS matches the validity
+        // implied by the DType.
+        if !self.dtype().is_nullable() && slice_validity != Validity::NonNullable {
+            assert!(
+                slice_validity.all_valid(slice_n_rows),
+                "ZSTD array expects to be non-nullable but there are nulls after decompression"
+            );
+
+            slice_validity = Validity::NonNullable;
+        } else if self.dtype.is_nullable() && slice_validity == Validity::NonNullable {
+            slice_validity = Validity::AllValid;
+        }
+        //
+        // END OF IMPORTANT BLOCK
+        //
+
         match &self.dtype {
             DType::Primitive(..) => {
                 let slice_values_buffer = decompressed.slice(
@@ -531,6 +553,21 @@ impl ZstdArray {
     }
 
     pub(crate) fn _slice(&self, start: usize, stop: usize) -> ZstdArray {
+        let new_start = self.slice_start + start;
+        let new_stop = self.slice_start + stop;
+
+        assert!(
+            new_start <= self.slice_stop,
+            "new slice start {new_start} exceeds end {}",
+            self.slice_stop
+        );
+
+        assert!(
+            new_stop <= self.slice_stop,
+            "new slice stop {new_stop} exceeds end {}",
+            self.slice_stop
+        );
+
         ZstdArray {
             slice_start: self.slice_start + start,
             slice_stop: self.slice_start + stop,

encodings/zstd/src/compute/cast.rs

@@ -2,46 +2,67 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use vortex_array::compute::{CastKernel, CastKernelAdapter};
-use vortex_array::{ArrayRef, IntoArray, register_kernel};
-use vortex_dtype::DType;
+use vortex_array::{ArrayRef, register_kernel};
+use vortex_dtype::{DType, Nullability};
 use vortex_error::VortexResult;
 
 use crate::{ZstdArray, ZstdVTable};
 
 impl CastKernel for ZstdVTable {
     fn cast(&self, array: &ZstdArray, dtype: &DType) -> VortexResult<Option<ArrayRef>> {
-        if !dtype.is_nullable() || !array.all_valid() {
-            // We cannot cast to non-nullable since the validity containing nulls is used to decode
-            // the ZSTD array, this would require rewriting tables.
+        if !dtype.eq_ignore_nullability(array.dtype()) {
+            // Type changes can't be handled in ZSTD, need to decode and tweak.
+            // TODO(aduffy): handle trivial conversions like Binary -> UTF8, integer widening, etc.
             return Ok(None);
         }
-        // ZstdArray is a general-purpose compression encoding using Zstandard compression.
-        // It can handle nullability changes without decompression by updating the validity
-        // bitmap, but type changes require decompression since the compressed data is
-        // type-specific and Zstd operates on raw bytes.
-        if array.dtype().eq_ignore_nullability(dtype) {
-            // Create a new validity with the target nullability
-            let new_validity = array
-                .unsliced_validity
-                .clone()
-                .cast_nullability(dtype.nullability(), array.len())?;
-
-            return Ok(Some(
+
+        let src_nullability = array.dtype().nullability();
+        let target_nullability = dtype.nullability();
+
+        match (src_nullability, target_nullability) {
+            // Same type case. This should be handled in the layer above but for
+            // completeness of the match arms we also handle it here.
+            (Nullability::Nullable, Nullability::Nullable)
+            | (Nullability::NonNullable, Nullability::NonNullable) => Ok(Some(array.to_array())),
+            (Nullability::NonNullable, Nullability::Nullable) => Ok(Some(
+                // nonnull => null, trivial cast by altering the validity
                 ZstdArray::new(
                     array.dictionary.clone(),
                     array.frames.clone(),
                     dtype.clone(),
                     array.metadata.clone(),
                     array.unsliced_n_rows(),
-                    new_validity,
+                    array.unsliced_validity.clone(),
                 )
-                ._slice(array.slice_start(), array.slice_stop())
-                .into_array(),
-            ));
+                .slice(array.slice_start()..array.slice_stop()),
+            )),
+            (Nullability::Nullable, Nullability::NonNullable) => {
+                // null => non-null works if there are no nulls in the sliced range
+                let sliced_len = array.slice_stop() - array.slice_start();
+                let has_nulls = !array
+                    .unsliced_validity
+                    .slice(array.slice_start()..array.slice_stop())
+                    .all_valid(sliced_len);
+
+                // We don't attempt to handle casting when there are nulls.
+                if has_nulls {
+                    return Ok(None);
+                }
+
+                // If there are no nulls, the cast is trivial
+                Ok(Some(
+                    ZstdArray::new(
+                        array.dictionary.clone(),
+                        array.frames.clone(),
+                        dtype.clone(),
+                        array.metadata.clone(),
+                        array.unsliced_n_rows(),
+                        array.unsliced_validity.clone(),
+                    )
+                    .slice(array.slice_start()..array.slice_stop()),
+                ))
+            }
         }
-
-        // For other casts (e.g., type changes), decode to canonical and let the underlying array handle it
-        Ok(None)
     }
 }
 

vortex-vector/src/binaryview/view.rs

@@ -46,6 +46,7 @@ pub struct Inlined {
 
 impl Inlined {
     /// Creates a new inlined representation from the provided value of constant size.
+    #[inline]
     fn new<const N: usize>(value: &[u8]) -> Self {
         debug_assert_eq!(value.len(), N);
         let mut inlined = Self {