@@ -7,7 +7,6 @@ use itertools::Itertools;
 use vortex_buffer::BitBuffer;
 use vortex_buffer::Buffer;
 use vortex_buffer::BufferAllocatorRef;
-use vortex_buffer::BufferMut;
 use vortex_buffer::BufferString;
 use vortex_buffer::ByteBuffer;
 use vortex_buffer::buffer;
@@ -35,8 +34,6 @@ use crate::arrays::UnionArray;
 use crate::arrays::VarBinViewArray;
 use crate::arrays::VariantArray;
 use crate::arrays::varbinview::BinaryView;
-use crate::builders::ArrayBuilder;
-use crate::builders::VarBinViewBuilder;
 use crate::builders::builder_with_capacity_in;
 use crate::dtype::DType;
 use crate::dtype::DecimalType;
@@ -48,7 +45,6 @@ use crate::match_smallest_list_offset_type;
 use crate::scalar::DecimalValue;
 use crate::scalar::ListScalar;
 use crate::scalar::Scalar;
-use crate::scalar::ScalarValue;
 use crate::validity::Validity;
 
 /// Shared implementation for both `canonicalize` and `execute` methods.
@@ -310,65 +306,19 @@ fn constant_canonical_list_array(
 }
 
 /// The elements of a list scalar as an array, one row per element; empty for a null list.
-///
-/// Primitive and byte elements are written straight from their values rather than through a scalar
-/// each.
 pub(crate) fn list_scalar_elements(list: &ListScalar, allocator: &BufferAllocatorRef) -> ArrayRef {
     let element_dtype = list.element_dtype();
-    let Some(values) = list.element_values() else {
+    let Some(elements) = list.elements() else {
         return Canonical::empty(element_dtype).into_array();
     };
 
-    match element_dtype {
-        DType::Primitive(ptype, nullability) => match_each_native_ptype!(ptype, |T| {
-            let mut buffer = BufferMut::<T>::with_capacity_in(values.len(), allocator.clone());
-            buffer.extend(values.iter().map(|value| {
-                value.as_ref().map_or_else(T::default, |value| {
-                    value
-                        .as_primitive()
-                        .cast::<T>()
-                        .vortex_expect("list element of the list's element ptype")
-                })
-            }));
-            PrimitiveArray::new(buffer.freeze(), element_validity(values, *nullability))
-                .into_array()
-        }),
-        DType::Utf8(_) | DType::Binary(_) => {
-            let mut builder = VarBinViewBuilder::with_capacity_in(
-                element_dtype.clone(),
-                values.len(),
-                allocator.clone(),
-            );
-            for value in values {
-                match value {
-                    None => builder.append_null(),
-                    Some(ScalarValue::Utf8(value)) => builder.append_value(value.as_bytes()),
-                    Some(value) => builder.append_value(value.as_binary().as_slice()),
-                }
-            }
-            builder.finish()
-        }
-        _ => {
-            let mut builder = builder_with_capacity_in(element_dtype, values.len(), allocator);
-            for idx in 0..values.len() {
-                builder
-                    .append_scalar(&list.element(idx).vortex_expect("index within the list"))
-                    .vortex_expect("list element scalar was invalid");
-            }
-            builder.finish()
-        }
-    }
-}
-
-/// The validity of a list's elements, a null element being a `None` value.
-fn element_validity(values: &[Option<ScalarValue>], nullability: Nullability) -> Validity {
-    match nullability {
-        Nullability::NonNullable => Validity::NonNullable,
-        Nullability::Nullable if values.iter().all(Option::is_some) => Validity::AllValid,
-        Nullability::Nullable => {
-            Validity::from(BitBuffer::from_iter(values.iter().map(Option::is_some)))
-        }
+    let mut builder = builder_with_capacity_in(element_dtype, elements.len(), allocator);
+    for element in &elements {
+        builder
+            .append_scalar(element)
+            .vortex_expect("list element scalar was invalid");
     }
+    builder.finish()
 }
 
 /// Creates a [`FixedSizeListArray`] whose every row holds the same list.
 
@@ -41,10 +41,15 @@ pub trait ListContainsElementReduce: VTable {
 /// For a needle that is not canonical, execution prepares a constant list into a
 /// [`PreparedSetArray`] and runs the kernels again. Thus a kernel can get the prepared set with
 /// `list.as_opt::<PreparedSet>()`, and probe its own values with [`PreparedSetData::contains`].
-/// For example, a dictionary probes only its values.
+/// For example, a dictionary probes only its values. A single value, such as the fill value of a
+/// sparse needle, is probed with [`PreparedSetData::contains_scalar`]. A kernel that finds its
+/// matches in another way makes its result with [`PreparedSetData::result_from_bits`], which applies
+/// the same null semantics.
 ///
 /// [`PreparedSetArray`]: crate::scalar_fn::fns::list_contains::PreparedSetArray
 /// [`PreparedSetData::contains`]: crate::scalar_fn::fns::list_contains::PreparedSetData::contains
+/// [`PreparedSetData::contains_scalar`]: crate::scalar_fn::fns::list_contains::PreparedSetData::contains_scalar
+/// [`PreparedSetData::result_from_bits`]: crate::scalar_fn::fns::list_contains::PreparedSetData::result_from_bits
 pub trait ListContainsElementKernel: VTable {
     fn list_contains(
         list: &ArrayRef,
 
@@ -142,54 +142,74 @@ impl PreparedSetData {
     /// a null element gives `null` for a needle that matches no element.
     ///
     /// A constant needle is probed once, and gives a constant result.
+    ///
+    /// # Errors
+    ///
+    /// Fails when the needles do not have the dtype of the list's elements.
     pub fn contains(
         &self,
         needles: &ArrayRef,
         options: &ListContainsOptions,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<ArrayRef> {
-        let element_dtype = self.elements.dtype();
-        if !element_dtype.eq_ignore_nullability(needles.dtype()) {
-            vortex_bail!(
-                "Element type {} of list does not match search value {}",
-                element_dtype,
-                needles.dtype(),
-            );
-        }
-
         if let Some(needle) = needles.as_constant() {
             let result = self.contains_scalar(&needle, options, ctx)?;
             return Ok(ConstantArray::new(result, needles.len()).into_array());
         }
 
+        self.check_needle_dtype(needles.dtype())?;
         let (bits, needle_validity) = self.set.probe.contains(needles, ctx)?;
-        self.finish(bits, needle_validity, needles.dtype(), options)
+        self.result_from_bits(bits, needle_validity, needles.dtype(), options)
     }
 
     /// Whether the constant `needle` is an element of the list, under `options`.
     ///
-    /// The answer is the one [`Self::contains`] gives for each row of a needle with this value. The
-    /// probe gets one row, not a row per needle.
-    fn contains_scalar(
+    /// The answer is the one [`Self::contains`] gives for each row of a needle with this value, for
+    /// example for the fill value of a sparse needle. The probe gets one row.
+    ///
+    /// # Errors
+    ///
+    /// Fails when the needle does not have the dtype of the list's elements.
+    pub fn contains_scalar(
         &self,
         needle: &Scalar,
         options: &ListContainsOptions,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<Scalar> {
+        self.check_needle_dtype(needle.dtype())?;
         let needles = ConstantArray::new(needle.clone(), 1).into_array();
         let (bits, needle_validity) = self.set.probe.contains(&needles, ctx)?;
-        self.finish(bits, needle_validity, needle.dtype(), options)?
+        self.result_from_bits(bits, needle_validity, needle.dtype(), options)?
             .execute_scalar(0, ctx)
     }
 
-    /// Makes the result from one membership bit per needle and the validity of the needles.
-    fn finish(
+    /// Makes the result of [`Self::contains`] from one membership bit per needle.
+    ///
+    /// A kernel that finds the matches of its needles without this probe, for example from bounds
+    /// of its own, uses this to apply the null semantics of [`Self::contains`]. A set bit in `bits`
+    /// tells that the needle equals an element. `needle_validity` is the validity of the needles,
+    /// and `needle_dtype` is their dtype. The bit of a null needle has no effect.
+    ///
+    /// # Errors
+    ///
+    /// Fails when `needle_dtype` is not the dtype of the list's elements, or when
+    /// `needle_validity` does not have one row per bit.
+    pub fn result_from_bits(
         &self,
         bits: BitBuffer,
         needle_validity: Validity,
         needle_dtype: &DType,
         options: &ListContainsOptions,
     ) -> VortexResult<ArrayRef> {
+        self.check_needle_dtype(needle_dtype)?;
+        if let Some(len) = needle_validity.maybe_len() {
+            vortex_ensure!(
+                len == bits.len(),
+                "Needle validity has {len} rows, but there are {} membership bits",
+                bits.len()
+            );
+        }
+
         let nullability = options.result_nullability(&self.list_dtype(), needle_dtype);
 
         let validity = if self.set.is_empty && !options.sql_null_semantics {
@@ -203,6 +223,19 @@ impl PreparedSetData {
 
         Ok(BoolArray::new(bits, validity.union_nullability(nullability)).into_array())
     }
+
+    /// Fails when needles of `needle_dtype` cannot be elements of the list.
+    fn check_needle_dtype(&self, needle_dtype: &DType) -> VortexResult<()> {
+        let element_dtype = self.elements.dtype();
+        if !element_dtype.eq_ignore_nullability(needle_dtype) {
+            vortex_bail!(
+                "Element type {} of list does not match search value {}",
+                element_dtype,
+                needle_dtype,
+            );
+        }
+        Ok(())
+    }
 }
 
 impl Debug for PreparedSetData {
@@ -236,7 +269,17 @@ impl ArrayEq for PreparedSetData {
 impl Array<PreparedSet> {
     /// Prepares `elements`, the elements of a non-null list with the list nullability
     /// `nullability`, as a set repeated `len` times.
-    pub(crate) fn try_new(
+    ///
+    /// [`ListContains`] prepares a constant list itself. A kernel crate can use this to test its
+    /// [`ListContainsElementKernel`] against a prepared set.
+    ///
+    /// [`ListContains`]: crate::scalar_fn::fns::list_contains::ListContains
+    /// [`ListContainsElementKernel`]: crate::scalar_fn::fns::list_contains::ListContainsElementKernel
+    ///
+    /// # Errors
+    ///
+    /// Fails when the elements cannot be executed into the probe.
+    pub fn try_new(
         elements: ArrayRef,
         nullability: Nullability,
         len: usize,
 
@@ -371,3 +371,44 @@ fn test_constant_row_needle_probes_one_row() -> VortexResult<()> {
     );
     Ok(())
 }
+
+#[rstest]
+#[case::default(ListContainsOptions::default(), [Some(true), Some(false), None])]
+#[case::sql(ListContainsOptions { sql_null_semantics: true }, [Some(true), None, None])]
+fn test_result_from_bits_applies_null_semantics(
+    #[case] options: ListContainsOptions,
+    #[case] expected: [Option<bool>; 3],
+) -> VortexResult<()> {
+    // Against `{2, null}`: a match, a non-match, and a null needle whose bit has no effect.
+    let mut ctx = array_session().create_execution_ctx();
+    let set = prepare(&set_with_null(), &mut ctx)?;
+    let needle_dtype = DType::Primitive(PType::I32, Nullability::Nullable);
+    let bits = BitBuffer::from_iter([true, false, true]);
+    let validity = Validity::from_iter([true, true, false]);
+
+    let result = set.result_from_bits(bits, validity, &needle_dtype, &options)?;
+    assert_arrays_eq!(result, BoolArray::from_iter(expected), &mut ctx);
+    Ok(())
+}
+
+#[test]
+fn test_result_from_bits_rejects_mismatched_needles() -> VortexResult<()> {
+    let mut ctx = array_session().create_execution_ctx();
+    let set = prepare(&set_with_null(), &mut ctx)?;
+    let options = ListContainsOptions::default();
+    let bits = BitBuffer::from_iter([true, false]);
+
+    let short_validity = Validity::from_iter([true]);
+    let needle_dtype = DType::Primitive(PType::I32, Nullability::Nullable);
+    assert!(
+        set.result_from_bits(bits.clone(), short_validity, &needle_dtype, &options)
+            .is_err()
+    );
+
+    let wrong_dtype = DType::Primitive(PType::I64, Nullability::Nullable);
+    assert!(
+        set.result_from_bits(bits, Validity::AllValid, &wrong_dtype, &options)
+            .is_err()
+    );
+    Ok(())
+}