@@ -25,7 +25,10 @@ use crate::arrays::ConstantArray;
 use crate::arrays::ExtensionArray;
 use crate::arrays::FixedSizeListArray;
 use crate::arrays::PrimitiveArray;
+use crate::arrays::VarBinViewArray;
+use crate::arrays::varbinview::BinaryView;
 use crate::assert_arrays_eq;
+use crate::buffer::BufferHandle;
 use crate::dtype::DType;
 use crate::dtype::NativePType;
 use crate::dtype::Nullability;
@@ -43,6 +46,8 @@ use crate::scalar_fn::unstable::row::OutputElement;
 use crate::scalar_fn::unstable::row::OutputSink;
 use crate::scalar_fn::unstable::row::RowFn;
 use crate::scalar_fn::unstable::row::RowVisitor;
+use crate::scalar_fn::unstable::row::Utf8Column;
+use crate::scalar_fn::unstable::row::execute::execute_bool_dense_attempt;
 use crate::scalar_fn::unstable::row::execute_rows;
 use crate::scalar_fn::unstable::row::row_fn_return_dtype;
 use crate::validity::Validity;
@@ -83,6 +88,20 @@ struct PackedGreaterThan<const MULTIVERSIONED: bool>;
 #[derive(Clone)]
 struct DeferredGreaterThan<const MULTIVERSIONED: bool>;
 
+#[derive(Clone, Copy)]
+enum BooleanInputShape {
+    Columns,
+    ConstantLhs,
+    ConstantRhs,
+}
+
+#[derive(Clone, Copy)]
+enum BooleanValidity {
+    AllValid,
+    AllNull,
+    Partial,
+}
+
 #[derive(Clone)]
 struct ValidOnlyPositive;
 
@@ -866,6 +885,100 @@ fn test_deferred_bool_output_handles_partial_constants() -> VortexResult<()> {
     Ok(())
 }
 
+#[rstest]
+#[case::empty(0)]
+#[case::one(1)]
+#[case::word_tail(63)]
+#[case::word(64)]
+#[case::word_remainder(65)]
+#[case::multiword(130)]
+fn test_deferred_bool_sliced_validity_and_constants(
+    #[case] len: usize,
+    #[values(false, true)] multiversioned: bool,
+    #[values(
+        BooleanInputShape::Columns,
+        BooleanInputShape::ConstantLhs,
+        BooleanInputShape::ConstantRhs
+    )]
+    input_shape: BooleanInputShape,
+    #[values(
+        BooleanValidity::AllValid,
+        BooleanValidity::AllNull,
+        BooleanValidity::Partial
+    )]
+    validity_pattern: BooleanValidity,
+) -> VortexResult<()> {
+    let valid = |index: usize| match validity_pattern {
+        BooleanValidity::AllValid => true,
+        BooleanValidity::AllNull => false,
+        BooleanValidity::Partial => !index.is_multiple_of(3),
+    };
+    let validity = Validity::Array(BoolArray::from_iter((0..len + 2).map(valid)).into_array());
+    let varying = PrimitiveArray::new(
+        (0..len + 2)
+            .map(|index| if valid(index) { index as i64 } else { i64::MIN })
+            .collect::<Vec<_>>(),
+        validity,
+    )
+    .into_array()
+    .slice(1..len + 1)?;
+    let constant = ConstantArray::new(32_i64, len).into_array();
+    let other = PrimitiveArray::from_iter(std::iter::repeat_n(32_i64, len)).into_array();
+    let (lhs, rhs) = match input_shape {
+        BooleanInputShape::Columns => (varying, other),
+        BooleanInputShape::ConstantLhs => (constant, varying),
+        BooleanInputShape::ConstantRhs => (varying, constant),
+    };
+    let args = VecExecutionArgs::new(vec![lhs, rhs], len);
+    let mut ctx = array_session().create_execution_ctx();
+    let actual = if multiversioned {
+        execute_rows(&DeferredGreaterThan::<true>, &EmptyOptions, &args, &mut ctx)?
+    } else {
+        execute_rows(
+            &DeferredGreaterThan::<false>,
+            &EmptyOptions,
+            &args,
+            &mut ctx,
+        )?
+    };
+    let expected = BoolArray::from_iter((1..len + 1).map(|index| {
+        valid(index).then_some(if matches!(input_shape, BooleanInputShape::ConstantLhs) {
+            32 > index as i64
+        } else {
+            index as i64 > 32
+        })
+    }))
+    .into_array();
+    assert_arrays_eq!(&actual, &expected, &mut ctx);
+    Ok(())
+}
+
+#[test]
+fn test_deferred_bool_decode_error_is_terminal() {
+    let views = Buffer::from(vec![BinaryView::make_view(&[0xff], 0, 0)]);
+    let input = VarBinViewArray::new_handle(
+        BufferHandle::new_host(views.into_byte_buffer()),
+        Default::default(),
+        DType::Utf8(Nullability::NonNullable),
+        Validity::NonNullable,
+    )
+    .into_array();
+    let args = VecExecutionArgs::new(vec![input], 1);
+    let mut ctx = array_session().create_execution_ctx();
+
+    let result = execute_bool_dense_attempt::<(Utf8Column,), (), bool, true>(
+        &args,
+        &mut ctx,
+        |_| (),
+        |&(), (value,)| (!value.is_empty(), false),
+        |_| Ok(()),
+    );
+    assert!(
+        result.is_err(),
+        "a decode error must not become a deferred row error"
+    );
+}
+
 #[test]
 fn test_deferred_bool_output_reports_valid_row_failure() -> VortexResult<()> {
     let lhs = PrimitiveArray::from_iter([1_i64, i64::MIN, -1]).into_array();
 
@@ -15,6 +15,7 @@ pub(super) use owned::execute_owned_infallible_valid_rows;
 pub(super) use owned::execute_owned_valid_rows;
 
 mod packed_bool;
+pub(super) use packed_bool::execute_bool_dense_attempt;
 pub(super) use packed_bool::execute_owned_bool;
 pub(super) use packed_bool::execute_owned_infallible_bool;
 
 
@@ -5,6 +5,10 @@
 //!
 //! These executors share the indexed input decoding used by owned outputs, but construct a
 //! canonical [`BoolArray`] without first collecting one byte per row.
+//!
+//! [`execute_owned_bool`] reports rejected failure evidence as a terminal error.
+//! [`execute_bool_dense_attempt`] hands it back as a [`DenseAttempt::DeferredError`] so that batch
+//! execution can retry over valid rows. The two keep separate copies of the row loop on purpose.
 
 use std::mem::size_of;
 
@@ -13,6 +17,7 @@ use vortex_compute::lane_kernels::IndexedSource;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 
+use super::DenseAttempt;
 use crate::ArrayRef;
 use crate::ExecutionCtx;
 use crate::IntoArray;
@@ -82,14 +87,15 @@ where
 
     let mut failure = Fail::default();
     let collect = |index| {
-        // SAFETY: Both collectors only invoke this closure with `index < row_count`, and the decoded
-        // source was constructed with exactly `row_count` rows.
+        // SAFETY: Both collectors only invoke this closure with `index < row_count`, and the
+        // decoded source was constructed with exactly `row_count` rows.
         let elements = unsafe { source.get_unchecked(index) };
         let (value, row_failure) = apply(&prepared, elements);
         failure |= row_failure;
 
         value
     };
+
     let values = if MULTIVERSIONED {
         BitBuffer::collect_bool_multiversioned(row_count, collect)
     } else {
@@ -100,3 +106,89 @@ where
 
     Ok(BoolArray::new(values, Validity::NonNullable).into_array())
 }
+
+/// Pack a dense Boolean attempt and report rejected failure evidence to the batch executor.
+///
+/// Returns the same terminal-versus-deferred split as
+/// [`execute_owned_dense_attempt`](super::execute_owned_dense_attempt), which owns that contract.
+pub(crate) fn execute_bool_dense_attempt<Args, Prepared, Fail, const MULTIVERSIONED: bool>(
+    args: &dyn ExecutionArgs,
+    ctx: &mut ExecutionCtx,
+    prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
+    apply: impl Fn(&Prepared, Args::Elems<'_>) -> (bool, Fail),
+    finish_failure: impl FnOnce(Fail) -> VortexResult<()>,
+) -> VortexResult<DenseAttempt>
+where
+    Args: IndexedElementTuple,
+    Fail: FailureEvidence,
+{
+    const {
+        assert!(
+            size_of::<Fail>() <= size_of::<bool>(),
+            "failure evidence must be no wider than the value, or it bounds the vector width"
+        )
+    };
+
+    // Keep this dense row loop separate from `execute_owned_bool`. Factoring their shared state
+    // into a helper changes LLVM's optimized dense kernel even when the helper is inlined.
+    let columns = Args::decode(args, ctx)?;
+    let prepared = prepare(Args::const_values(&columns));
+    let row_count = args.row_count();
+
+    let Some(source) = decoded_source::<Args>(&columns, row_count) else {
+        vortex_bail!("a decoded row input does not address exactly {row_count} rows");
+    };
+
+    // NB: The collector must capture one borrow of the whole state. That shape is load-bearing for
+    // code generation, not a borrow checker workaround. See the `DeferredBoolState` documentation.
+    let mut state = DeferredBoolState {
+        source,
+        prepared,
+        apply,
+        failure: Fail::default(),
+    };
+    let state_ref = &mut state;
+
+    let collect = move |index| {
+        let state = &mut *state_ref;
+
+        // SAFETY: Both collectors only invoke this closure with `index < row_count`, and the
+        // decoded source was constructed with exactly `row_count` rows.
+        let elements = unsafe { state.source.get_unchecked(index) };
+        let (value, row_failure) = (state.apply)(&state.prepared, elements);
+        state.failure |= row_failure;
+
+        value
+    };
+
+    let values = if MULTIVERSIONED {
+        BitBuffer::collect_bool_multiversioned(row_count, collect)
+    } else {
+        BitBuffer::collect_bool(row_count, collect)
+    };
+
+    match finish_failure(state.failure) {
+        Ok(()) => Ok(DenseAttempt::Values(
+            BoolArray::new(values, Validity::NonNullable).into_array(),
+        )),
+        Err(error) => Ok(DenseAttempt::DeferredError(error)),
+    }
+}
+
+/// Row-loop state for a deferred Boolean computation: the decoded input, the prepared batch
+/// constants, the per-row callback, and the accumulated failure evidence.
+///
+/// The collector closure captures this struct through one mutable borrow. It does not capture the
+/// four parts separately. The single borrow gives the out-of-line collector one exclusive pointer
+/// to its state, and the vectorizer needs that to keep the multiversioned word loop packed.
+/// Separate captures lose the same aliasing information, and the loop stays scalar.
+///
+/// The capture shape is therefore load-bearing for code generation. The measurements come from
+/// the `row_fn_bool_retry` benchmark, which builds in the `bench` profile with 16 codegen units
+/// and no LTO. Recheck that benchmark and its optimized IR before you change the captures.
+struct DeferredBoolState<Source, Prepared, Apply, Fail> {
+    source: Source,
+    prepared: Prepared,
+    apply: Apply,
+    failure: Fail,
+}
@@ -31,6 +31,7 @@ use crate::scalar_fn::unstable::row::RowFn;
 use crate::scalar_fn::unstable::row::SinkResult;
 use crate::scalar_fn::unstable::row::batch::BorrowedRowFnArgs;
 use crate::scalar_fn::unstable::row::execute::DenseAttempt;
+use crate::scalar_fn::unstable::row::execute::execute_bool_dense_attempt;
 use crate::scalar_fn::unstable::row::execute::execute_owned_dense_attempt;
 use crate::scalar_fn::unstable::row::execute::execute_owned_infallible;
 use crate::scalar_fn::unstable::row::execute::execute_sink;
@@ -152,4 +153,31 @@ impl<F: RowFn> RowVisitor for ExecuteDenseWithRetry<'_, '_, '_, F> {
             finish_failure,
         )
     }
+
+    fn visit_prepared_deferred_bool<Args, Prepared, Fail, const MULTIVERSIONED: bool>(
+        self,
+        prepare: impl FnOnce(Args::ConstElems<'_>) -> Prepared,
+        apply: impl Fn(&Prepared, Args::Elems<'_>) -> (bool, Fail),
+        finish_failure: impl FnOnce(Fail) -> VortexResult<()>,
+    ) -> VortexResult<Self::VisitResult>
+    where
+        Args: IndexedElementTuple,
+        Fail: FailureEvidence,
+    {
+        const { assert_deferred_visit_contract::<F, Args, bool, Fail>() };
+        let visited = BatchPlan::new(
+            validate_owned_visit::<Args, bool>(self.args.dtypes())?,
+            self.output_dtype,
+            RowPolicy::for_deferred_output::<Args>(),
+        )?;
+        self.args.plan().ensure_reproduced_by(&visited)?;
+
+        execute_bool_dense_attempt::<Args, Prepared, Fail, MULTIVERSIONED>(
+            self.args,
+            self.ctx,
+            prepare,
+            apply,
+            finish_failure,
+        )
+    }
 }
@@ -21,8 +21,9 @@
 //! simple enough that the per-level duplication and its `#[target_feature]` call boundary pay
 //! off.
 //!
-//! The bit-at-a-time loop lives on as [`collect_bool_word_scalar`], used for tail chunks and as
-//! the reference implementation for tests and benchmarks.
+//! The bit-at-a-time loop lives on as [`collect_bool_word_scalar`], the reference implementation
+//! for tests and benchmarks. The word loop packs its tail chunk with [`collect_bool_word_tail`],
+//! its own copy of that loop.
 
 /// Packs up to 64 boolean values into a little-endian `u64` word one bit at a time.
 ///
@@ -172,10 +173,32 @@ where
 
     if remainder != 0 {
         let offset = full * 64;
-        words[full] = collect_bool_word_scalar(remainder, |bit_idx| f(offset + bit_idx));
+        words[full] = collect_bool_word_tail(remainder, |bit_idx| f(offset + bit_idx));
     }
 }
 
+/// Tail chunk of [`collect_bool_words_with`]: a copy of [`collect_bool_word_scalar`] that always
+/// inlines. `len` **must** be at most 64.
+///
+/// The word loop must keep the borrowed callback inline. An out-of-line tail lets the callback
+/// state escape, and the full words packed by the same loop then fail to vectorize.
+/// [`collect_bool_word_scalar`] stays an ordinary `#[inline]` function so that this constraint does
+/// not reach its public callers, which include the scalar benchmark baselines.
+#[expect(clippy::inline_always)]
+#[inline(always)]
+fn collect_bool_word_tail<F>(len: usize, mut f: F) -> u64
+where
+    F: FnMut(usize) -> bool,
+{
+    debug_assert!(len <= 64, "cannot pack {len} bits into a u64 word");
+
+    let mut packed = 0;
+    for bit_idx in 0..len {
+        packed |= (f(bit_idx) as u64) << bit_idx;
+    }
+    packed
+}
+
 /// SSE2 copy of the [`collect_bool_words`](crate::bit::collect_bool_words) word loop.
 ///
 /// `words` must hold at least `len.div_ceil(64)` entries and `f` is invoked with `0..len`.