Read integer validity a word per chunk, and fill nulls for runs (#10342)

## Summary

Stacked on #10341, which adds the benchmarks.

Integer statistics with nulls sliced the validity `BitBuffer` for every
chunk of 64 values to count its set bits, then tested each value's bit
with a branch. With nulls spread out, almost no chunk is fully valid
(0.9⁶⁴ ≈ 0.1% at 10% nulls), so nearly every chunk took that slow path.

## Changes

In `vortex-compressor/src/stats/integer.rs`:

- Read the validity one `u64` word per chunk instead of slicing a bit
buffer. A word of zeros is skipped, and a full word takes the non-null
loop as before.
- Without distinct values, fill each null with the valid value before
it. A filled null adds no run, so the chunk takes the same branch-free
run count as a non-null chunk.
- With distinct values, push only the set bits, since a filled null
would count as an occurrence.
- The trailing values are padded into a last chunk whose padding is
null. This replaces the separate per-bit remainder loop.

Min/max is unchanged: it is still computed by the `min_max` aggregate
before this loop.

`test_matches_naive_reference` now also covers statistics without
distinct values and chunks that are entirely null. Filling the nulls
with their own values instead makes 38 of its cases fail.

---------

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: joseph-isaacs

vortex-compressor/src/stats/integer.rs

@@ -15,7 +15,6 @@ use vortex_array::expr::stats::Stat;
 use vortex_array::match_each_integer_ptype;
 use vortex_array::scalar::PValue;
 use vortex_array::scalar::Scalar;
-use vortex_buffer::BitBuffer;
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -394,48 +393,42 @@ where
 
     let sliced = buffer.slice(head_idx..array.len());
     let (chunks, remainder) = sliced.as_slice().as_chunks::<64>();
-    match validity.bit_buffer() {
+    let remainder_valid = match validity.bit_buffer() {
         AllOr::All => {
             for chunk in chunks {
                 inner_loop_nonnull(chunk, count_distinct_values, &mut loop_state)
             }
-            inner_loop_naive(
-                remainder,
-                count_distinct_values,
-                &BitBuffer::new_set(remainder.len()),
-                &mut loop_state,
-            );
+            (1u64 << remainder.len()) - 1
         }
         AllOr::None => unreachable!("All invalid arrays have been handled before"),
         AllOr::Some(v) => {
             let mask = v.slice(head_idx..array.len());
-            let mut offset = 0;
-            for chunk in chunks {
-                let validity = mask.slice(offset..(offset + 64));
-                offset += 64;
-
-                match validity.true_count() {
+            let words = mask.chunks();
+            // One validity word per chunk of 64 values.
+            for (chunk, valid) in chunks.iter().zip(words.iter()) {
+                match valid {
                     // All nulls -> no stats to update.
-                    0 => continue,
+                    0 => {}
                     // Inner loop for when validity check can be elided.
-                    64 => inner_loop_nonnull(chunk, count_distinct_values, &mut loop_state),
+                    u64::MAX => inner_loop_nonnull(chunk, count_distinct_values, &mut loop_state),
                     // Inner loop for when we need to check validity.
-                    _ => inner_loop_nullable(
-                        chunk,
-                        count_distinct_values,
-                        &validity,
-                        &mut loop_state,
-                    ),
+                    _ => inner_loop_nullable(chunk, count_distinct_values, valid, &mut loop_state),
                 }
             }
-            // Final iteration, run naive loop.
-            inner_loop_naive(
-                remainder,
-                count_distinct_values,
-                &mask.slice(offset..(offset + remainder.len())),
-                &mut loop_state,
-            );
+            words.remainder_bits()
         }
+    };
+
+    // Pad the trailing values into a last chunk, whose padding is null.
+    if remainder_valid != 0 {
+        let mut last = [loop_state.prev; 64];
+        last[..remainder.len()].copy_from_slice(remainder);
+        inner_loop_nullable(
+            &last,
+            count_distinct_values,
+            remainder_valid,
+            &mut loop_state,
+        );
     }
 
     if count_distinct_values {
@@ -653,57 +646,45 @@ fn inner_loop_nonnull_impl<T: IntegerPType, const COUNT_DISTINCT_VALUES: bool>(
     state.prev = values[63];
 }
 
-/// Inner loop for nullable chunks of 64 values.
+/// Inner loop for chunks of 64 values with nulls, whose bits are unset in `valid`.
 #[allow(clippy::inline_always)]
 #[inline(always)]
 fn inner_loop_nullable<T: IntegerPType>(
     values: &[T; 64],
     count_distinct_values: bool,
-    is_valid: &BitBuffer,
+    valid: u64,
     state: &mut LoopState<T>,
 ) where
     NativeValue<T>: Eq + Hash,
 {
     if count_distinct_values {
-        inner_loop_masked::<T, true>(values, is_valid, state);
-    } else {
-        inner_loop_masked::<T, false>(values, is_valid, state);
-    }
-}
-
-/// Fallback inner loop for remainder values.
-#[allow(clippy::inline_always)]
-#[inline(always)]
-fn inner_loop_naive<T: IntegerPType>(
-    values: &[T],
-    count_distinct_values: bool,
-    is_valid: &BitBuffer,
-    state: &mut LoopState<T>,
-) where
-    NativeValue<T>: Eq + Hash,
-{
-    if count_distinct_values {
-        inner_loop_masked::<T, true>(values, is_valid, state);
+        // A filled null would count as an occurrence, so only the valid values are pushed.
+        let mut valid = valid;
+        while valid != 0 {
+            state.push::<true>(values[valid.trailing_zeros() as usize]);
+            valid &= valid - 1;
+        }
     } else {
-        inner_loop_masked::<T, false>(values, is_valid, state);
+        // A null filled with the valid value before it adds no run, so the chunk takes the
+        // branch-free loop of non-null chunks.
+        inner_loop_nonnull_impl::<T, false>(&forward_fill(values, valid, state.prev), state);
     }
 }
 
-/// Processes values with a validity mask, skipping nulls without breaking runs.
+/// Returns `values` with each null, whose bit is unset in `valid`, replaced by the closest valid
+/// value before it, or by `prev`, the last valid value before the chunk. The cost is one store per
+/// null.
 #[allow(clippy::inline_always)]
 #[inline(always)]
-fn inner_loop_masked<T: IntegerPType, const COUNT_DISTINCT_VALUES: bool>(
-    values: &[T],
-    is_valid: &BitBuffer,
-    state: &mut LoopState<T>,
-) where
-    NativeValue<T>: Eq + Hash,
-{
-    for (idx, &value) in values.iter().enumerate() {
-        if is_valid.value(idx) {
-            state.push::<COUNT_DISTINCT_VALUES>(value);
-        }
-    }
+fn forward_fill<T: Copy>(values: &[T; 64], valid: u64, prev: T) -> [T; 64] {
+    let mut filled = *values;
+    let mut nulls = !valid;
+    while nulls != 0 {
+        let i = nulls.trailing_zeros() as usize;
+        filled[i] = if i == 0 { prev } else { filled[i - 1] };
+        nulls &= nulls - 1;
+    }
+    filled
 }
 
 #[cfg(test)]
@@ -853,16 +834,20 @@ mod tests {
 
     /// Checks the chunked loops against a naive reference. The run lengths cover aligned
     /// constant chunks (64), runs that cross chunk boundaries (3, 100), and chunks where every
-    /// value changes (1).
+    /// value changes (1). With nulls, `null_chunks` also nulls two whole chunks.
     #[rstest]
     fn test_matches_naive_reference(
         #[values(1, 63, 64, 65, 64 * 20 + 17)] len: u32,
         #[values(1, 3, 64, 100)] run_len: u32,
         #[values(None, Some(97), Some(3))] null_every: Option<u32>,
+        #[values(false, true)] null_chunks: bool,
+        #[values(false, true)] count_distinct_values: bool,
     ) -> VortexResult<()> {
         let values: Vec<u32> = (0..len).map(|i| (i / run_len) % 16).collect();
         let valid: Vec<bool> = (0..len)
-            .map(|i| null_every.is_none_or(|n| i % n != 0))
+            .map(|i| {
+                null_every.is_none_or(|n| i % n != 0 && !(null_chunks && (64..192).contains(&i)))
+            })
             .collect();
         let (expected, runs) = naive_stats(&values, &valid);
 
@@ -872,16 +857,18 @@ mod tests {
         };
         let array = PrimitiveArray::new(Buffer::from(values), validity);
         let mut ctx = array_session().create_execution_ctx();
-        let stats = typed_int_stats::<u32>(&array, true, &mut ctx)?;
+        let stats = typed_int_stats::<u32>(&array, count_distinct_values, &mut ctx)?;
 
         let ErasedStats::U32(typed) = stats.erased() else {
             unreachable!()
         };
-        let actual: HashMap<u32, u32> = typed
-            .distinct()
-            .map(|d| d.distinct_values().iter().map(|(k, &c)| (k.0, c)).collect())
-            .unwrap_or_default();
-        assert_eq!(actual, expected);
+        if count_distinct_values {
+            let actual: HashMap<u32, u32> = typed
+                .distinct()
+                .map(|d| d.distinct_values().iter().map(|(k, &c)| (k.0, c)).collect())
+                .unwrap_or_default();
+            assert_eq!(actual, expected);
+        }
         let value_count: u32 = expected.values().sum();
         assert_eq!(stats.value_count, value_count);
         if value_count > 0 {