arrow_select/

filter.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Defines filter kernels

use std::ops::AddAssign;
use std::sync::Arc;

use arrow_array::builder::BooleanBufferBuilder;
use arrow_array::cast::AsArray;
use arrow_array::types::{
    ArrowDictionaryKeyType, ArrowPrimitiveType, ByteArrayType, ByteViewType, RunEndIndexType,
};
use arrow_array::*;
use arrow_buffer::bit_chunk_iterator::BitChunks;
use arrow_buffer::{
    ArrowNativeType, BooleanBuffer, NullBuffer, OffsetBuffer, RunEndBuffer, ScalarBuffer, bit_util,
};
use arrow_buffer::{Buffer, MutableBuffer};
use arrow_data::bit_iterator::{BitIndexIterator, BitSliceIterator};
use arrow_data::transform::MutableArrayData;
use arrow_schema::*;

/// If the filter selects more than this fraction of rows, use
/// [`SlicesIterator`] to copy ranges of values. Otherwise iterate
/// over individual rows using [`IndexIterator`]
///
/// Threshold of 0.8 chosen based on <https://dl.acm.org/doi/abs/10.1145/3465998.3466009>
///
const FILTER_SLICES_SELECTIVITY_THRESHOLD: f64 = 0.8;

/// An iterator of `(usize, usize)` each representing an interval
/// `[start, end)` whose slots of a bitmap [Buffer] are true.
///
/// Each interval corresponds to a contiguous region of memory to be
/// "taken" from an array to be filtered.
///
/// ## Notes:
///
/// 1. Ignores the validity bitmap (ignores nulls)
///
/// 2. Only performant for filters that copy across long contiguous runs
#[derive(Debug)]
pub struct SlicesIterator<'a>(BitSliceIterator<'a>);

impl<'a> SlicesIterator<'a> {
    /// Creates a new iterator from a [BooleanArray]
    pub fn new(filter: &'a BooleanArray) -> Self {
        filter.values().into()
    }
}

impl<'a> From<&'a BooleanBuffer> for SlicesIterator<'a> {
    fn from(filter: &'a BooleanBuffer) -> Self {
        Self(filter.set_slices())
    }
}

impl Iterator for SlicesIterator<'_> {
    type Item = (usize, usize);

    #[inline]
    fn next(&mut self) -> Option<Self::Item> {
        self.0.next()
    }
}

/// An iterator of `usize` whose index in [`BooleanArray`] is true
///
/// This provides the best performance on most predicates, apart from those which keep
/// large runs and therefore favour [`SlicesIterator`]
pub(crate) struct IndexIterator<'a> {
    remaining: usize,
    iter: BitIndexIterator<'a>,
}

impl<'a> IndexIterator<'a> {
    pub(crate) fn new(filter: &'a BooleanArray, remaining: usize) -> Self {
        assert_eq!(filter.null_count(), 0);
        let iter = filter.values().set_indices();
        Self { remaining, iter }
    }

    /// Collect this iterator as a [`Vec`]
    /// This is more efficient than the standard `collect` as we can
    /// pre-allocate the entire uninitialized buffer and then fill it (roughly 1.6x faster)
    pub fn collect(mut self) -> Vec<usize> {
        let len = self.remaining;
        let mut result = Vec::with_capacity(len);
        let ptr: *mut usize = result.as_mut_ptr();
        for i in 0..len {
            // SAFETY: we have allocated enough space in `result` and remaining
            // correctly tracks the number of elements
            let next = self.iter.next();
            debug_assert!(next.is_some(), "IndexIterator exhausted early");
            unsafe {
                *ptr.add(i) = next.unwrap_unchecked();
            }
        }
        // SAFETY: we have initialized `len` elements
        unsafe {
            result.set_len(len);
        }
        result
    }
}

impl Iterator for IndexIterator<'_> {
    type Item = usize;

    fn next(&mut self) -> Option<Self::Item> {
        if self.remaining != 0 {
            // Fascinatingly swapping these two lines around results in a 50%
            // performance regression for some benchmarks
            let next = self.iter.next().expect("IndexIterator exhausted early");
            self.remaining -= 1;
            // Must panic if exhausted early as trusted length iterator
            return Some(next);
        }
        None
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (self.remaining, Some(self.remaining))
    }
}

/// Convert all null values in `BooleanArray` to `false`
///
/// This is useful for filter-like operations which select only `true`
/// values, but not `false` or `NULL` values
///
/// Internally this is implemented as a bitwise `AND` operation with null bits
/// and the boolean bits.
///
/// # Example
/// ```
/// # use arrow_array::{Array, BooleanArray};
/// # use arrow_select::filter::prep_null_mask_filter;
/// let filter = BooleanArray::from(vec![
///   Some(true),
///   Some(false),
///   None
/// ]);
/// // convert Boolean array to a filter mask
/// let null_mask = prep_null_mask_filter(&filter);
/// // there are no nulls in the output mask
/// assert!(null_mask.nulls().is_none());
/// assert_eq!(null_mask, BooleanArray::from(vec![
///  true,
///  false,
///  false, // Null is converted to false
/// ]));
/// ```
pub fn prep_null_mask_filter(filter: &BooleanArray) -> BooleanArray {
    let nulls = filter.nulls().unwrap();
    let mask = filter.values() & nulls.inner();
    BooleanArray::new(mask, None)
}

/// Returns a filtered `values` [`Array`] where the corresponding elements of
/// `predicate` are `true`.
///
/// If multiple arrays (or record batches) need to be filtered using the same predicate array,
/// consider using [FilterBuilder] to create a single [FilterPredicate] and then
/// calling [FilterPredicate::filter_record_batch].
///
/// In contrast to this function, it is then the responsibility of the caller
/// to use [FilterBuilder::optimize] if appropriate.
///
/// If the number of rows that `predicate` selects is already known,
/// [`FilterBuilder::with_count`] avoids counting them again.
///
/// # See also
/// * [`FilterBuilder`] for more control over the filtering process.
/// * [`filter_record_batch`] to filter a [`RecordBatch`]
/// * [`BatchCoalescer`]: to filter multiple [`RecordBatch`] and coalesce
///   the results into a single array.
///
/// [`BatchCoalescer`]: crate::coalesce::BatchCoalescer
///
/// # Example
/// ```rust
/// # use arrow_array::{Int32Array, BooleanArray};
/// # use arrow_select::filter::filter;
/// let array = Int32Array::from(vec![5, 6, 7, 8, 9]);
/// let filter_array = BooleanArray::from(vec![true, false, false, true, false]);
/// let c = filter(&array, &filter_array).unwrap();
/// let c = c.as_any().downcast_ref::<Int32Array>().unwrap();
/// assert_eq!(c, &Int32Array::from(vec![5, 8]));
/// ```
pub fn filter(values: &dyn Array, predicate: &BooleanArray) -> Result<ArrayRef, ArrowError> {
    let mut filter_builder = FilterBuilder::new(predicate);

    if FilterBuilder::is_optimize_beneficial(values.data_type()) {
        // Only optimize if filtering more than one array
        // Otherwise, the overhead of optimization can be more than the benefit
        filter_builder = filter_builder.optimize();
    }

    let predicate = filter_builder.build();

    filter_array(values, &predicate)
}

/// Returns a filtered [RecordBatch] where the corresponding elements of
/// `predicate` are true.
///
/// This is the equivalent of calling [filter] on each column of the [RecordBatch].
///
/// If multiple record batches (or arrays) need to be filtered using the same predicate array,
/// consider using [FilterBuilder] to create a single [FilterPredicate] and then
/// calling [FilterPredicate::filter_record_batch].
/// In contrast to this function, it is then the responsibility of the caller
/// to use [FilterBuilder::optimize] if appropriate.
///
/// If the number of rows that `predicate` selects is already known,
/// [`FilterBuilder::with_count`] avoids counting them again.
pub fn filter_record_batch(
    record_batch: &RecordBatch,
    predicate: &BooleanArray,
) -> Result<RecordBatch, ArrowError> {
    let mut filter_builder = FilterBuilder::new(predicate);
    let num_cols = record_batch.num_columns();
    if num_cols > 1
        || (num_cols > 0
            && FilterBuilder::is_optimize_beneficial(
                record_batch.schema_ref().field(0).data_type(),
            ))
    {
        // Only optimize if filtering more than one column or if the column contains multiple internal arrays
        // Otherwise, the overhead of optimization can be more than the benefit
        filter_builder = filter_builder.optimize();
    }
    let filter = filter_builder.build();

    filter.filter_record_batch(record_batch)
}

/// A builder to construct [`FilterPredicate`]
#[derive(Debug)]
pub struct FilterBuilder {
    filter: BooleanArray,
    /// The number of rows `filter` selects, if provided by [`Self::with_count`]
    count: Option<usize>,
    optimize: bool,
}

impl FilterBuilder {
    /// Create a new [`FilterBuilder`] that can be used to construct a [`FilterPredicate`]
    pub fn new(filter: &BooleanArray) -> Self {
        let filter = match filter.null_count() {
            0 => filter.clone(),
            _ => prep_null_mask_filter(filter),
        };

        Self {
            filter,
            count: None,
            optimize: false,
        }
    }

    /// Set the number of rows that the filter selects, so that [`Self::build`]
    /// does not have to count them.
    ///
    /// Callers that build a mask row by row, or derive it from a validity
    /// buffer with a cached null count, often already hold this number.
    ///
    /// # Safety
    ///
    /// `count` must equal [`BooleanArray::true_count`] of the filter passed to
    /// [`Self::new`]: the number of `true` values that are not null.
    ///
    /// # Example
    /// ```
    /// # use arrow_array::{BooleanArray, Int32Array};
    /// # use arrow_select::filter::FilterBuilder;
    /// let values = Int32Array::from(vec![1, 2, 3, 4]);
    /// let mask = BooleanArray::from(vec![Some(true), None, Some(true), Some(false)]);
    /// // The null is not selected, so the mask selects two rows.
    /// // SAFETY: the count matches the mask.
    /// let predicate = unsafe { FilterBuilder::new(&mask).with_count(2) }.build();
    /// assert_eq!(predicate.count(), 2);
    /// let filtered = predicate.filter(&values).unwrap();
    /// assert_eq!(filtered.as_ref(), &Int32Array::from(vec![1, 3]));
    /// ```
    pub unsafe fn with_count(mut self, count: usize) -> Self {
        debug_assert_eq!(
            count,
            self.filter.true_count(),
            "count must match the number of rows the filter selects"
        );
        self.count = Some(count);
        self
    }

    /// Compute an optimized representation of the provided `filter` mask in
    /// [`Self::build`], so that it can be applied to an array more quickly.
    ///
    /// When filtering multiple arrays (e.g. a [`RecordBatch`] or a
    /// [`StructArray`] with multiple fields), optimizing the filter can provide
    /// significant performance benefits.
    ///
    /// However, optimization takes time and can have a larger memory footprint
    /// than the original mask, so it is often faster to filter a single array,
    /// without filter optimization.
    pub fn optimize(mut self) -> Self {
        self.optimize = true;
        self
    }

    /// Determines if calling [FilterBuilder::optimize] is beneficial for the
    /// given type even when filtering just a single array.
    ///
    /// See [`FilterBuilder::optimize`] for more details.
    pub fn is_optimize_beneficial(data_type: &DataType) -> bool {
        match data_type {
            DataType::Struct(fields) => {
                fields.len() > 1
                    || fields.len() == 1
                        && FilterBuilder::is_optimize_beneficial(fields[0].data_type())
            }
            DataType::Union(fields, UnionMode::Sparse) => !fields.is_empty(),
            _ => false,
        }
    }

    /// Construct the final `FilterPredicate`
    pub fn build(self) -> FilterPredicate {
        let count = self.count.unwrap_or_else(|| self.filter.true_count());
        let mut strategy = IterationStrategy::default_strategy(self.filter.len(), count);
        if self.optimize {
            match strategy {
                IterationStrategy::SlicesIterator => {
                    let slices = SlicesIterator::new(&self.filter).collect();
                    strategy = IterationStrategy::Slices(slices)
                }
                IterationStrategy::IndexIterator => {
                    let indices = IndexIterator::new(&self.filter, count).collect();
                    strategy = IterationStrategy::Indices(indices)
                }
                _ => {}
            }
        }

        FilterPredicate {
            filter: self.filter,
            count,
            strategy,
        }
    }
}

/// The iteration strategy used to evaluate [`FilterPredicate`]
#[derive(Debug)]
enum IterationStrategy {
    /// A lazily evaluated iterator of ranges
    SlicesIterator,
    /// A lazily evaluated iterator of indices
    IndexIterator,
    /// A precomputed list of indices
    Indices(Vec<usize>),
    /// A precomputed array of ranges
    Slices(Vec<(usize, usize)>),
    /// Select all rows
    All,
    /// Select no rows
    None,
}

impl IterationStrategy {
    /// The default [`IterationStrategy`] for a filter of length `filter_length`
    /// and selecting `filter_count` rows
    fn default_strategy(filter_length: usize, filter_count: usize) -> Self {
        if filter_length == 0 || filter_count == 0 {
            return IterationStrategy::None;
        }

        if filter_count == filter_length {
            return IterationStrategy::All;
        }

        // Compute the selectivity of the predicate by dividing the number of true
        // bits in the predicate by the predicate's total length
        //
        // This can then be used as a heuristic for the optimal iteration strategy
        let selectivity_frac = filter_count as f64 / filter_length as f64;
        if selectivity_frac > FILTER_SLICES_SELECTIVITY_THRESHOLD {
            return IterationStrategy::SlicesIterator;
        }
        IterationStrategy::IndexIterator
    }
}

/// Borrowed description of which rows a [`FilterPredicate`] selects.
///
/// This is used for filtering multiple arrays with the same predicate without
/// having to clone the predicate's internal data structures (e.g. the list of
/// indices or slices).
pub(crate) enum FilterSelection<'a> {
    /// No rows are selected
    None,
    /// All `len` rows are selected
    All { len: usize },
    /// Iterator of `(start, end)` slices, each a run of contiguous selected rows
    Slices(FilterSlices<'a>),
    /// Iterator of the indices of the selected rows
    Indices(FilterIndices<'a>),
}

pub(crate) type FilterSlices<'a> =
    FilterIterator<std::iter::Copied<std::slice::Iter<'a, (usize, usize)>>, SlicesIterator<'a>>;

pub(crate) type FilterIndices<'a> =
    FilterIterator<std::iter::Copied<std::slice::Iter<'a, usize>>, IndexIterator<'a>>;

/// Internal implementation of [`FilterSelection`] that holds either an iterator
/// over a precomputed (materialized) list of rows, or a lazy iterator that
/// derives the selected rows from the predicate on the fly.
///
/// This does not implement [`Iterator`] on purpose. Callers use
/// [`Self::for_each`] or [`Self::try_for_each`] so the enum is matched once
/// before the loop, not once per row in `next`.
pub(crate) enum FilterIterator<M, I> {
    Materialized(M),
    Lazy(I),
}

impl<M, I> FilterIterator<M, I>
where
    M: Iterator,
    I: Iterator<Item = M::Item>,
{
    /// Call the infallible function `f` for each item in this [`FilterIterator`]
    pub(crate) fn for_each<F>(self, f: F)
    where
        F: FnMut(M::Item),
    {
        match self {
            Self::Materialized(iter) => iter.for_each(f),
            Self::Lazy(iter) => iter.for_each(f),
        }
    }

    /// Call the fallible function `f` for each item in this [`FilterIterator`],
    /// stopping and returning the error if `f` returns `Err`.
    pub(crate) fn try_for_each<F, E>(self, mut f: F) -> Result<(), E>
    where
        F: FnMut(M::Item) -> Result<(), E>,
    {
        match self {
            Self::Materialized(iter) => {
                for item in iter {
                    f(item)?;
                }
            }
            Self::Lazy(iter) => {
                for item in iter {
                    f(item)?;
                }
            }
        }

        Ok(())
    }
}

/// A filtering predicate that can be applied to an [`Array`]
#[derive(Debug)]
pub struct FilterPredicate {
    filter: BooleanArray,
    count: usize,
    /// Precomputed strategy for iterating over the selected rows of this predicate
    strategy: IterationStrategy,
}

impl FilterPredicate {
    /// Selects rows from `values` based on this [`FilterPredicate`]
    pub fn filter(&self, values: &dyn Array) -> Result<ArrayRef, ArrowError> {
        filter_array(values, self)
    }

    /// Returns a filtered [`RecordBatch`] containing only the rows that are selected by this
    /// [`FilterPredicate`].
    ///
    /// This is the equivalent of calling [filter] on each column of the [`RecordBatch`].
    pub fn filter_record_batch(
        &self,
        record_batch: &RecordBatch,
    ) -> Result<RecordBatch, ArrowError> {
        let filtered_arrays = record_batch
            .columns()
            .iter()
            .map(|a| filter_array(a, self))
            .collect::<Result<Vec<_>, _>>()?;

        // SAFETY: we know that the set of filtered arrays will match the schema of the original
        // record batch
        unsafe {
            Ok(RecordBatch::new_unchecked(
                record_batch.schema(),
                filtered_arrays,
                self.count,
            ))
        }
    }

    /// Number of rows being selected based on this [`FilterPredicate`]
    pub fn count(&self) -> usize {
        self.count
    }

    /// Length of the filter mask, including rows that are not selected
    pub(crate) fn filter_len(&self) -> usize {
        self.filter.len()
    }

    /// Return a [`FilterSelection`] for iterating over the rows selected by
    /// this [`FilterPredicate`].
    pub(crate) fn selection(&self) -> FilterSelection<'_> {
        match &self.strategy {
            IterationStrategy::None => FilterSelection::None,
            IterationStrategy::All => FilterSelection::All { len: self.count },
            IterationStrategy::Slices(slices) => {
                FilterSelection::Slices(FilterIterator::Materialized(slices.iter().copied()))
            }
            IterationStrategy::SlicesIterator => {
                FilterSelection::Slices(FilterIterator::Lazy(SlicesIterator::new(&self.filter)))
            }
            IterationStrategy::Indices(indices) => {
                FilterSelection::Indices(FilterIterator::Materialized(indices.iter().copied()))
            }
            IterationStrategy::IndexIterator => FilterSelection::Indices(FilterIterator::Lazy(
                IndexIterator::new(&self.filter, self.count),
            )),
        }
    }

    /// Filters the given `nulls` buffer using this predicate.
    ///
    /// Returns `None` when there is nothing to track in the output, either
    /// because the input `nulls` was `None`, the input had no nulls, or the
    /// filtered result has no nulls. Otherwise returns the filtered
    /// [`NullBuffer`] with its precomputed null count.
    ///
    /// # Panics
    ///
    /// Panics if the input contains nulls and is shorter than this predicate.
    pub fn filter_nulls(&self, nulls: Option<&NullBuffer>) -> Option<NullBuffer> {
        let nulls = nulls?;
        if nulls.null_count() == 0 {
            return None;
        }

        assert!(nulls.len() >= self.filter.len());
        match self.strategy {
            IterationStrategy::None => return None,
            IterationStrategy::All => {
                let nulls = nulls.slice(0, self.count);
                return (nulls.null_count() != 0).then_some(nulls);
            }
            _ => {}
        }

        let nulls = filter_bits(nulls.inner(), self);
        // The filtered `nulls` has a length of `self.count` bits and therefore
        // the null count is this minus the number of valid bits
        let null_count = self.count - nulls.count_set_bits_offset(0, self.count);

        if null_count == 0 {
            return None;
        }

        let buffer = BooleanBuffer::new(nulls, 0, self.count);
        debug_assert_eq!(null_count, buffer.len() - buffer.count_set_bits());
        // SAFETY: `null_count` was derived from `buffer` above, so it matches
        // the number of unset bits as required by `new_unchecked`.
        Some(unsafe { NullBuffer::new_unchecked(buffer, null_count) })
    }
}

fn filter_array(values: &dyn Array, predicate: &FilterPredicate) -> Result<ArrayRef, ArrowError> {
    if predicate.filter.len() > values.len() {
        return Err(ArrowError::InvalidArgumentError(format!(
            "Filter predicate of length {} is larger than target array of length {}",
            predicate.filter.len(),
            values.len()
        )));
    }

    match predicate.strategy {
        IterationStrategy::None => Ok(new_empty_array(values.data_type())),
        IterationStrategy::All => Ok(values.slice(0, predicate.count)),
        // actually filter
        _ => downcast_primitive_array! {
            values => Ok(Arc::new(filter_primitive(values, predicate))),
            DataType::Boolean => {
                let values = values.as_any().downcast_ref::<BooleanArray>().unwrap();
                Ok(Arc::new(filter_boolean(values, predicate)))
            }
            DataType::Utf8 => {
                Ok(Arc::new(filter_bytes(values.as_string::<i32>(), predicate)))
            }
            DataType::LargeUtf8 => {
                Ok(Arc::new(filter_bytes(values.as_string::<i64>(), predicate)))
            }
            DataType::Utf8View => {
                Ok(Arc::new(filter_byte_view(values.as_string_view(), predicate)))
            }
            DataType::Binary => {
                Ok(Arc::new(filter_bytes(values.as_binary::<i32>(), predicate)))
            }
            DataType::LargeBinary => {
                Ok(Arc::new(filter_bytes(values.as_binary::<i64>(), predicate)))
            }
            DataType::BinaryView => {
                Ok(Arc::new(filter_byte_view(values.as_binary_view(), predicate)))
            }
            DataType::FixedSizeBinary(_) => {
                Ok(Arc::new(filter_fixed_size_binary(values.as_fixed_size_binary(), predicate)))
            }
            DataType::ListView(_) => {
                Ok(Arc::new(filter_list_view::<i32>(values.as_list_view(), predicate)))
            }
            DataType::LargeListView(_) => {
                Ok(Arc::new(filter_list_view::<i64>(values.as_list_view(), predicate)))
            }
            DataType::RunEndEncoded(_, _) => {
                downcast_run_array!{
                    values => Ok(Arc::new(filter_run_end_array(values, predicate)?)),
                    t => unimplemented!("Filter not supported for RunEndEncoded type {:?}", t)
                }
            }
            DataType::Dictionary(_, _) => downcast_dictionary_array! {
                values => Ok(Arc::new(filter_dict(values, predicate))),
                t => unimplemented!("Filter not supported for dictionary type {:?}", t)
            }
            DataType::Struct(_) => {
                Ok(Arc::new(filter_struct(values.as_struct(), predicate)?))
            }
            DataType::Union(_, UnionMode::Sparse) => {
                Ok(Arc::new(filter_sparse_union(values.as_union(), predicate)?))
            }
            _ => {
                let data = values.to_data();
                // fallback to using MutableArrayData
                let mut mutable = MutableArrayData::new(
                    vec![&data],
                    false,
                    predicate.count,
                );

                match &predicate.strategy {
                    IterationStrategy::Slices(slices) => {
                        for (start, end) in slices {
                            mutable.try_extend(0, *start, *end)?;
                        }
                    }
                    _ => {
                        let iter = SlicesIterator::new(&predicate.filter);
                        for (start, end) in iter {
                            mutable.try_extend(0, start, end)?;
                        }
                    }
                }

                let data = mutable.freeze();
                Ok(make_array(data))
            }
        },
    }
}

/// Filter any supported [`RunArray`] based on a [`FilterPredicate`]
fn filter_run_end_array<R: RunEndIndexType>(
    array: &RunArray<R>,
    predicate: &FilterPredicate,
) -> Result<RunArray<R>, ArrowError>
where
    R::Native: Into<i64> + From<bool>,
    R::Native: AddAssign,
{
    let run_ends: &RunEndBuffer<R::Native> = array.run_ends();
    let start_physical = run_ends.get_start_physical_index();
    let end_physical = run_ends.get_end_physical_index();
    let physical_len = end_physical - start_physical + 1;

    let mut new_run_ends = vec![R::default_value(); physical_len];
    let offset = run_ends.offset() as u64;

    let mut start = 0u64;
    let mut j = 0;
    let mut count = R::default_value();
    let filter_values = predicate.filter.values();
    let run_ends = run_ends.inner();

    let pred: BooleanArray = BooleanBuffer::collect_bool(physical_len, |i| {
        let mut keep = false;
        let mut end = (run_ends[i + start_physical].into() as u64).saturating_sub(offset);
        let difference = end.saturating_sub(filter_values.len() as u64);
        end -= difference;

        // Safety: we subtract the difference off `end` so we are always within bounds
        for pred in (start..end).map(|i| unsafe { filter_values.value_unchecked(i as usize) }) {
            count += R::Native::from(pred);
            keep |= pred
        }
        // this is to avoid branching
        new_run_ends[j] = count;
        j += keep as usize;

        start = end;
        keep
    })
    .into();

    new_run_ends.truncate(j);

    let values = array.values_slice();
    let values = filter(values.as_ref(), &pred)?;

    let run_ends = PrimitiveArray::<R>::try_new(new_run_ends.into(), None)?;
    RunArray::try_new(&run_ends, &values)
}

/// Filter the packed bitmask `buffer` with `predicate`, choosing between the
/// strategy-based and compress-based kernels by filter density
fn filter_bits(buffer: &BooleanBuffer, predicate: &FilterPredicate) -> Buffer {
    // Compressing scans the whole mask a word at a time, so it loses to the
    // slices strategies once fewer than one bit per word is dropped, and to
    // precomputed `Indices` once fewer than one bit per word is kept. The lazy
    // `IndexIterator` scans the mask anyway, so it never beats compressing
    let len = predicate.filter.len();
    let count = predicate.count;
    let dense = count >= len - len / 64;
    let sparse_indices =
        count <= len / 64 && matches!(predicate.strategy, IterationStrategy::Indices(_));
    if !dense && !sparse_indices {
        return filter_bits_compress(buffer, predicate);
    }
    filter_bits_strategy(buffer, predicate)
}

/// Filter the packed bitmask `buffer` with `predicate` using its
/// [`IterationStrategy`]
fn filter_bits_strategy(buffer: &BooleanBuffer, predicate: &FilterPredicate) -> Buffer {
    let src = buffer.values();
    let offset = buffer.offset();
    assert!(buffer.len() >= predicate.filter.len());

    match &predicate.strategy {
        IterationStrategy::IndexIterator => {
            let bits =
                // SAFETY: IndexIterator uses the filter predicate to derive indices
                IndexIterator::new(&predicate.filter, predicate.count).map(|src_idx| unsafe {
                    bit_util::get_bit_raw(buffer.values().as_ptr(), src_idx + offset)
                });

            // SAFETY: `IndexIterator` reports its size correctly
            unsafe { MutableBuffer::from_trusted_len_iter_bool(bits).into() }
        }
        IterationStrategy::Indices(indices) => {
            // SAFETY: indices were derived from the filter predicate
            let bits = indices.iter().map(|src_idx| unsafe {
                bit_util::get_bit_raw(buffer.values().as_ptr(), *src_idx + offset)
            });
            // SAFETY: `Vec::iter()` reports its size correctly
            unsafe { MutableBuffer::from_trusted_len_iter_bool(bits).into() }
        }
        IterationStrategy::SlicesIterator => {
            let mut builder = BooleanBufferBuilder::new(predicate.count);
            for (start, end) in SlicesIterator::new(&predicate.filter) {
                builder.append_packed_range(start + offset..end + offset, src)
            }
            builder.into()
        }
        IterationStrategy::Slices(slices) => {
            let mut builder = BooleanBufferBuilder::new(predicate.count);
            for (start, end) in slices {
                builder.append_packed_range(*start + offset..*end + offset, src)
            }
            builder.into()
        }
        IterationStrategy::All | IterationStrategy::None => unreachable!(),
    }
}

/// Filter the packed bitmask `buffer` with `predicate` by extracting the kept
/// bits of each 64-bit word with [`bit_util::compress`] (`pext`)
///
/// Not inlined: within `filter_array` the packing state spills to the stack
#[inline(never)]
fn filter_bits_compress(buffer: &BooleanBuffer, predicate: &FilterPredicate) -> Buffer {
    /// Packs the bits extracted from successive words into the low `filled`
    /// bits of `current`; once complete it is written at `idx` and restarts
    /// from the bits that did not fit
    struct Packer {
        ptr: *mut u64,
        idx: usize,
        current: u64,
        filled: u32,
    }

    impl Packer {
        #[inline(always)]
        fn push(&mut self, values: u64, mask: u64) {
            let bits = bit_util::compress(values, mask);
            self.current |= bits << self.filled;
            let total = self.filled + mask.count_ones();
            if total < 64 {
                self.filled = total;
            } else {
                // SAFETY: `count` is the number of set bits in the filter, so
                // at most `count / 64` words are ever completed and the
                // buffer holds `count / 64 + 1`
                unsafe { self.ptr.add(self.idx).write(self.current) };
                self.idx += 1;
                // `bits >> (64 - filled)`, written so that `filled == 0`
                // shifts everything out
                self.current = (bits >> 1) >> (63 - self.filled);
                self.filled = total - 64;
            }
        }
    }

    assert!(buffer.len() >= predicate.filter.len());
    let mask_chunks = predicate.filter.values().bit_chunks();
    let value_chunks = BitChunks::new(buffer.values(), buffer.offset(), predicate.filter.len());
    // `count` is the filter's set bit count, which the buffer size and the
    // raw writes below rely on, and both chunk views cover
    // `predicate.filter.len()` bits, so indexing `value_chunks` by the
    // position in `mask_chunks` stays in bounds
    debug_assert_eq!(predicate.count, predicate.filter.true_count());
    debug_assert_eq!(mask_chunks.chunk_len(), value_chunks.chunk_len());

    // One word beyond the complete ones for the trailing partial word
    let mut out: Vec<u64> = Vec::with_capacity(predicate.count / 64 + 1);
    let mut packer = Packer {
        ptr: out.as_mut_ptr(),
        idx: 0,
        current: 0,
        filled: 0,
    };

    for (index, mask) in mask_chunks.iter().enumerate() {
        // Words with no kept bits are skipped before the corresponding values
        // are read, so only the mask is touched for them
        if mask == 0 {
            continue;
        }
        packer.push(value_chunks.chunk(index), mask);
    }
    packer.push(value_chunks.remainder_bits(), mask_chunks.remainder_bits());

    // The trailing partial word; its bits above `filled` are zero
    // SAFETY: `idx <= count / 64`, so this and every word below it is
    // within the buffer and written
    debug_assert!(packer.idx < out.capacity());
    unsafe {
        packer.ptr.add(packer.idx).write(packer.current);
        out.set_len(packer.idx + 1);
    }
    let mut out = MutableBuffer::from(out);
    out.truncate(bit_util::ceil(predicate.count, 8));
    out.into()
}

/// `filter` implementation for boolean buffers
fn filter_boolean(array: &BooleanArray, predicate: &FilterPredicate) -> BooleanArray {
    let buffer = filter_bits(array.values(), predicate);
    let values = BooleanBuffer::new(buffer, 0, predicate.count);
    let nulls = predicate.filter_nulls(array.nulls());

    BooleanArray::new(values, nulls)
}

#[inline(never)]
pub(crate) fn filter_native<T: ArrowNativeType>(
    values: &[T],
    predicate: &FilterPredicate,
) -> Buffer {
    assert!(values.len() >= predicate.filter.len());

    match &predicate.strategy {
        IterationStrategy::SlicesIterator => {
            let mut buffer = Vec::with_capacity(predicate.count);
            for (start, end) in SlicesIterator::new(&predicate.filter) {
                // SAFETY: indices were derived from the filter predicate
                buffer.extend_from_slice(unsafe { values.get_unchecked(start..end) });
            }
            buffer.into()
        }
        IterationStrategy::Slices(slices) => {
            let mut buffer = Vec::with_capacity(predicate.count);
            for (start, end) in slices {
                // SAFETY: indices were derived from the filter predicate
                buffer.extend_from_slice(unsafe { values.get_unchecked(*start..*end) });
            }
            buffer.into()
        }
        IterationStrategy::IndexIterator => {
            // SAFETY: indices were derived from the filter predicate
            let iter = IndexIterator::new(&predicate.filter, predicate.count)
                .map(|x| unsafe { *values.get_unchecked(x) });

            // SAFETY: IndexIterator is trusted length
            unsafe { MutableBuffer::from_trusted_len_iter(iter) }.into()
        }
        IterationStrategy::Indices(indices) => {
            // SAFETY: indices were derived from the filter predicate
            let iter = indices.iter().map(|x| unsafe { *values.get_unchecked(*x) });
            iter.collect::<Vec<_>>().into()
        }
        IterationStrategy::All | IterationStrategy::None => unreachable!(),
    }
}

/// `filter` implementation for primitive arrays
fn filter_primitive<T>(array: &PrimitiveArray<T>, predicate: &FilterPredicate) -> PrimitiveArray<T>
where
    T: ArrowPrimitiveType,
{
    let buffer = filter_native(array.values(), predicate);
    let values = ScalarBuffer::new(buffer, 0, predicate.count);
    let nulls = predicate.filter_nulls(array.nulls());
    let filtered = PrimitiveArray::new(values, nulls);

    // Avoid the compatibility check when the physical type already matches.
    if array.data_type() == &T::DATA_TYPE {
        filtered
    } else {
        filtered.with_data_type(array.data_type().clone())
    }
}

/// [`FilterBytes`] is created from a source [`GenericByteArray`] and can be
/// used to build a new [`GenericByteArray`] by copying values from the source
///
/// TODO(raphael): Could this be used for the take kernel as well?
struct FilterBytes<'a, OffsetSize> {
    src_offsets: &'a [OffsetSize],
    src_values: &'a [u8],
    dst_offsets: Vec<OffsetSize>,
    dst_values: Vec<u8>,
    cur_offset: OffsetSize,
}

impl<'a, OffsetSize> FilterBytes<'a, OffsetSize>
where
    OffsetSize: OffsetSizeTrait,
{
    fn new<T>(capacity: usize, array: &'a GenericByteArray<T>) -> Self
    where
        T: ByteArrayType<Offset = OffsetSize>,
    {
        let dst_values = Vec::new();
        let mut dst_offsets: Vec<OffsetSize> = Vec::with_capacity(capacity + 1);
        let cur_offset = OffsetSize::from_usize(0).unwrap();

        dst_offsets.push(cur_offset);

        Self {
            src_offsets: array.value_offsets(),
            src_values: array.value_data(),
            dst_offsets,
            dst_values,
            cur_offset,
        }
    }

    /// Returns the byte offset at `idx`
    #[inline]
    fn get_value_offset(&self, idx: usize) -> usize {
        self.src_offsets[idx].as_usize()
    }

    /// Returns the start and end of the value at index `idx` along with its length
    #[inline]
    fn get_value_range(&self, idx: usize) -> (usize, usize, OffsetSize) {
        // These can only fail if `array` contains invalid data
        let start = self.get_value_offset(idx);
        let end = self.get_value_offset(idx + 1);
        let len = OffsetSize::from_usize(end - start).expect("illegal offset range");
        (start, end, len)
    }

    fn extend_offsets_idx(&mut self, iter: impl Iterator<Item = usize>) {
        self.dst_offsets.extend(iter.map(|idx| {
            let start = self.src_offsets[idx].as_usize();
            let end = self.src_offsets[idx + 1].as_usize();
            let len = OffsetSize::from_usize(end - start).expect("illegal offset range");
            self.cur_offset += len;

            self.cur_offset
        }));
    }

    /// Extends the in-progress array by the indexes in the provided iterator
    fn extend_idx(&mut self, iter: impl Iterator<Item = usize>) {
        self.dst_values.reserve_exact(self.cur_offset.as_usize());

        for idx in iter {
            let start = self.src_offsets[idx].as_usize();
            let end = self.src_offsets[idx + 1].as_usize();
            self.dst_values
                .extend_from_slice(&self.src_values[start..end]);
        }
    }

    fn extend_offsets_slices(&mut self, iter: impl Iterator<Item = (usize, usize)>, count: usize) {
        self.dst_offsets.reserve_exact(count);
        for (start, end) in iter {
            // These can only fail if `array` contains invalid data
            for idx in start..end {
                let (_, _, len) = self.get_value_range(idx);
                self.cur_offset += len;
                self.dst_offsets.push(self.cur_offset);
            }
        }
    }

    /// Extends the in-progress array by the ranges in the provided iterator
    fn extend_slices(&mut self, iter: impl Iterator<Item = (usize, usize)>) {
        self.dst_values.reserve_exact(self.cur_offset.as_usize());

        for (start, end) in iter {
            let value_start = self.get_value_offset(start);
            let value_end = self.get_value_offset(end);
            self.dst_values
                .extend_from_slice(&self.src_values[value_start..value_end]);
        }
    }
}

/// `filter` implementation for byte arrays
///
/// Note: NULLs with a non-zero slot length in `array` will have the corresponding
/// data copied across. This allows handling the null mask separately from the data
fn filter_bytes<T>(array: &GenericByteArray<T>, predicate: &FilterPredicate) -> GenericByteArray<T>
where
    T: ByteArrayType,
{
    let mut filter = FilterBytes::new(predicate.count, array);

    match &predicate.strategy {
        IterationStrategy::SlicesIterator => {
            filter.extend_offsets_slices(SlicesIterator::new(&predicate.filter), predicate.count);
            filter.extend_slices(SlicesIterator::new(&predicate.filter))
        }
        IterationStrategy::Slices(slices) => {
            filter.extend_offsets_slices(slices.iter().copied(), predicate.count);
            filter.extend_slices(slices.iter().copied())
        }
        IterationStrategy::IndexIterator => {
            filter.extend_offsets_idx(IndexIterator::new(&predicate.filter, predicate.count));
            filter.extend_idx(IndexIterator::new(&predicate.filter, predicate.count))
        }
        IterationStrategy::Indices(indices) => {
            filter.extend_offsets_idx(indices.iter().copied());
            filter.extend_idx(indices.iter().copied())
        }
        IterationStrategy::All | IterationStrategy::None => unreachable!(),
    }

    // SAFETY: `dst_offsets` starts at `[0]` and only grows by the running
    // `cur_offset`, so it is monotonically non-decreasing.
    let offsets = unsafe { OffsetBuffer::new_unchecked(filter.dst_offsets.into()) };
    let nulls = predicate.filter_nulls(array.nulls());

    // SAFETY: `offsets` index into `dst_values` by construction, and each slot
    // is a byte-for-byte copy from `array`, so UTF-8 validity (if any) is preserved.
    // Length invariant: `offsets.len() - 1 == predicate.count == nulls.len()`.
    unsafe { GenericByteArray::new_unchecked(offsets, filter.dst_values.into(), nulls) }
}

/// `filter` implementation for byte view arrays.
fn filter_byte_view<T: ByteViewType>(
    array: &GenericByteViewArray<T>,
    predicate: &FilterPredicate,
) -> GenericByteViewArray<T> {
    let new_view_buffer = filter_native(array.views(), predicate);
    let views = ScalarBuffer::new(new_view_buffer, 0, predicate.count);
    let buffers = Arc::clone(array.data_buffers());
    let nulls = predicate.filter_nulls(array.nulls());

    // SAFETY: each view is copied unchanged from `array.views()` and `buffers`
    // is the same buffer list, so every view still points to an in-bounds
    // (and, for strings, UTF-8 valid) range.
    unsafe { GenericByteViewArray::new_unchecked(views, buffers, nulls) }
}

/// Copies fixed-size binary elements at `indices` from `values` into a new `MutableBuffer`.
/// Uses raw pointer writes and `with_capacity` to avoid zero-initialization and per-call overhead.
#[inline(always)]
fn copy_fsb_indices(
    values: &[u8],
    value_length: usize,
    indices: impl Iterator<Item = usize>,
    count: usize,
) -> MutableBuffer {
    let total = count * value_length;
    let mut buffer = MutableBuffer::with_capacity(total);
    let dst_base = buffer.as_mut_ptr();
    let mut write_offset = 0usize;
    for idx in indices {
        let src_start = idx * value_length;
        // SAFETY: `idx` is derived from the filter predicate so it is a valid array index;
        // we allocated `count * value_length` bytes and advance by `value_length` per step.
        unsafe {
            std::ptr::copy_nonoverlapping(
                values.as_ptr().add(src_start),
                dst_base.add(write_offset),
                value_length,
            );
        }
        write_offset += value_length;
    }
    // SAFETY: we wrote exactly `count * value_length` bytes into the buffer.
    unsafe { buffer.set_len(total) };
    buffer
}

fn filter_fixed_size_binary(
    array: &FixedSizeBinaryArray,
    predicate: &FilterPredicate,
) -> FixedSizeBinaryArray {
    let values: &[u8] = array.values();
    let value_length = array.value_length() as usize;
    let calculate_offset_from_index = |index: usize| index * value_length;
    let buffer = match &predicate.strategy {
        IterationStrategy::SlicesIterator => {
            let mut buffer = MutableBuffer::with_capacity(predicate.count * value_length);
            for (start, end) in SlicesIterator::new(&predicate.filter) {
                buffer.extend_from_slice(
                    &values[calculate_offset_from_index(start)..calculate_offset_from_index(end)],
                );
            }
            buffer
        }
        IterationStrategy::Slices(slices) => {
            let mut buffer = MutableBuffer::with_capacity(predicate.count * value_length);
            for (start, end) in slices {
                buffer.extend_from_slice(
                    &values[calculate_offset_from_index(*start)..calculate_offset_from_index(*end)],
                );
            }
            buffer
        }
        IterationStrategy::IndexIterator => copy_fsb_indices(
            values,
            value_length,
            IndexIterator::new(&predicate.filter, predicate.count),
            predicate.count,
        ),
        IterationStrategy::Indices(indices) => copy_fsb_indices(
            values,
            value_length,
            indices.iter().copied(),
            predicate.count,
        ),
        IterationStrategy::All | IterationStrategy::None => unreachable!(),
    };

    let nulls = predicate.filter_nulls(array.nulls());

    FixedSizeBinaryArray::try_new_with_len(
        array.value_length(),
        buffer.into(),
        nulls,
        predicate.count,
    )
    .unwrap()
}

/// `filter` implementation for dictionaries
fn filter_dict<K: ArrowDictionaryKeyType>(
    array: &DictionaryArray<K>,
    predicate: &FilterPredicate,
) -> DictionaryArray<K> {
    // SAFETY:
    // Keys were valid before, filtered subset is therefore still valid
    let new_keys = filter_primitive(array.keys(), predicate);
    unsafe { DictionaryArray::new_unchecked(new_keys, array.values().clone()) }
}

/// `filter` implementation for structs
fn filter_struct(
    array: &StructArray,
    predicate: &FilterPredicate,
) -> Result<StructArray, ArrowError> {
    let columns = array
        .columns()
        .iter()
        .map(|column| filter_array(column, predicate))
        .collect::<Result<_, _>>()?;

    let nulls = predicate.filter_nulls(array.nulls());

    Ok(unsafe {
        StructArray::new_unchecked_with_length(
            array.fields().clone(),
            columns,
            nulls,
            predicate.count(),
        )
    })
}

/// `filter` implementation for sparse unions
fn filter_sparse_union(
    array: &UnionArray,
    predicate: &FilterPredicate,
) -> Result<UnionArray, ArrowError> {
    let DataType::Union(fields, UnionMode::Sparse) = array.data_type() else {
        unreachable!()
    };

    let type_ids = filter_primitive(
        &Int8Array::try_new(array.type_ids().clone(), None)?,
        predicate,
    );

    let children = fields
        .iter()
        .map(|(child_type_id, _)| filter_array(array.child(child_type_id), predicate))
        .collect::<Result<_, _>>()?;

    Ok(unsafe {
        UnionArray::new_unchecked(fields.clone(), type_ids.into_parts().1, None, children)
    })
}

/// `filter` implementation for list views
fn filter_list_view<OffsetType: OffsetSizeTrait>(
    array: &GenericListViewArray<OffsetType>,
    predicate: &FilterPredicate,
) -> GenericListViewArray<OffsetType> {
    let filtered_offsets = filter_native::<OffsetType>(array.offsets(), predicate);
    let filtered_sizes = filter_native::<OffsetType>(array.sizes(), predicate);

    let field = match array.data_type() {
        DataType::ListView(field) | DataType::LargeListView(field) => field.clone(),
        _ => unreachable!(),
    };
    let offsets = ScalarBuffer::new(filtered_offsets, 0, predicate.count);
    let sizes = ScalarBuffer::new(filtered_sizes, 0, predicate.count);
    let values = array.values().clone();
    let nulls = predicate.filter_nulls(array.nulls());

    // SAFETY: each `(offset, size)` pair is copied unchanged from `array` and
    // indexes into the same `values` child, so every range stays in-bounds.
    // `field` and `values`' data type are unchanged from `array`.
    unsafe { GenericListViewArray::new_unchecked(field, offsets, sizes, values, nulls) }
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow_array::builder::*;
    use arrow_array::cast::as_run_array;
    use arrow_array::types::*;
    use rand::distr::uniform::{UniformSampler, UniformUsize};
    use rand::distr::{Alphanumeric, StandardUniform};
    use rand::prelude::*;
    use rand::rng;

    #[test]
    fn test_filter_nulls_all() {
        let predicate = FilterBuilder::new(&BooleanArray::from(vec![true, true])).build();
        let nulls = NullBuffer::from(vec![true, false]);
        assert_eq!(predicate.filter_nulls(Some(&nulls)), Some(nulls));
    }

    #[test]
    fn test_filter_nulls_none() {
        let predicate = FilterBuilder::new(&BooleanArray::from(vec![false, false])).build();
        let nulls = NullBuffer::from(vec![true, false]);
        assert_eq!(predicate.filter_nulls(Some(&nulls)), None);
    }

    #[test]
    fn test_filter_nulls_selection() {
        // Cover full and partial prefixes, empty selections, and both general
        // iteration strategies, with and without materializing the selection.
        let filters = [
            vec![],
            vec![true],
            vec![true, true],
            vec![true; 6],
            vec![false; 6],
            vec![true, false, false, false, false, false],
            vec![true, true, true, true, true, false],
        ];
        for offset in [0, 3, 9] {
            let mut validity = vec![false; offset];
            validity.extend([true, false, true, false, true, false]);
            let nulls = NullBuffer::from(validity).slice(offset, 6);
            for filter in &filters {
                let expected: NullBuffer = filter
                    .iter()
                    .enumerate()
                    .filter(|(_, selected)| **selected)
                    .map(|(i, _)| nulls.is_valid(i))
                    .collect();
                let expected = (expected.null_count() != 0).then_some(expected);
                let filter = BooleanArray::from(filter.clone());
                for optimize in [false, true] {
                    let builder = FilterBuilder::new(&filter);
                    let predicate = if optimize {
                        builder.optimize()
                    } else {
                        builder
                    }
                    .build();
                    assert_eq!(predicate.filter_nulls(Some(&nulls)), expected);
                    assert_eq!(predicate.filter_nulls(None), None);
                    assert_eq!(
                        predicate.filter_nulls(Some(&NullBuffer::new_valid(6))),
                        None
                    );
                }
            }
        }
    }

    #[test]
    fn test_filter_nulls_length_check() {
        for filter in [vec![true, true], vec![false, false], vec![true, false]] {
            let predicate = FilterBuilder::new(&BooleanArray::from(filter)).build();
            let nulls = NullBuffer::new_null(1);
            assert!(std::panic::catch_unwind(|| predicate.filter_nulls(Some(&nulls))).is_err());
            // Inputs without nulls continue to short-circuit before checking length.
            assert_eq!(
                predicate.filter_nulls(Some(&NullBuffer::new_valid(1))),
                None
            );
            assert_eq!(predicate.filter_nulls(None), None);
        }
    }

    macro_rules! def_temporal_test {
        ($test:ident, $array_type: ident, $data: expr) => {
            #[test]
            fn $test() {
                let a = $data;
                let b = BooleanArray::from(vec![true, false, true, false]);
                let c = filter(&a, &b).unwrap();
                let d = c.as_ref().as_any().downcast_ref::<$array_type>().unwrap();
                assert_eq!(2, d.len());
                assert_eq!(1, d.value(0));
                assert_eq!(3, d.value(1));
            }
        };
    }

    def_temporal_test!(
        test_filter_date32,
        Date32Array,
        Date32Array::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_date64,
        Date64Array,
        Date64Array::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_time32_second,
        Time32SecondArray,
        Time32SecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_time32_millisecond,
        Time32MillisecondArray,
        Time32MillisecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_time64_microsecond,
        Time64MicrosecondArray,
        Time64MicrosecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_time64_nanosecond,
        Time64NanosecondArray,
        Time64NanosecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_duration_second,
        DurationSecondArray,
        DurationSecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_duration_millisecond,
        DurationMillisecondArray,
        DurationMillisecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_duration_microsecond,
        DurationMicrosecondArray,
        DurationMicrosecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_duration_nanosecond,
        DurationNanosecondArray,
        DurationNanosecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_timestamp_second,
        TimestampSecondArray,
        TimestampSecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_timestamp_millisecond,
        TimestampMillisecondArray,
        TimestampMillisecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_timestamp_microsecond,
        TimestampMicrosecondArray,
        TimestampMicrosecondArray::from(vec![1, 2, 3, 4])
    );
    def_temporal_test!(
        test_filter_timestamp_nanosecond,
        TimestampNanosecondArray,
        TimestampNanosecondArray::from(vec![1, 2, 3, 4])
    );

    #[test]
    fn test_filter_array_slice() {
        let a = Int32Array::from(vec![5, 6, 7, 8, 9]).slice(1, 4);
        let b = BooleanArray::from(vec![true, false, false, true]);
        // filtering with sliced filter array is not currently supported
        // let b_slice = BooleanArray::from(vec![true, false, false, true, false]).slice(1, 4);
        // let b = b_slice.as_any().downcast_ref().unwrap();
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(2, d.len());
        assert_eq!(6, d.value(0));
        assert_eq!(9, d.value(1));
    }

    #[test]
    fn test_filter_array_low_density() {
        // this test exercises the all 0's branch of the filter algorithm
        let mut data_values = (1..=65).collect::<Vec<i32>>();
        let mut filter_values = (1..=65).map(|i| matches!(i % 65, 0)).collect::<Vec<bool>>();
        // set up two more values after the batch
        data_values.extend_from_slice(&[66, 67]);
        filter_values.extend_from_slice(&[false, true]);
        let a = Int32Array::from(data_values);
        let b = BooleanArray::from(filter_values);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(2, d.len());
        assert_eq!(65, d.value(0));
        assert_eq!(67, d.value(1));
    }

    #[test]
    fn test_filter_array_high_density() {
        // this test exercises the all 1's branch of the filter algorithm
        let mut data_values = (1..=65).map(Some).collect::<Vec<_>>();
        let mut filter_values = (1..=65)
            .map(|i| !matches!(i % 65, 0))
            .collect::<Vec<bool>>();
        // set second data value to null
        data_values[1] = None;
        // set up two more values after the batch
        data_values.extend_from_slice(&[Some(66), None, Some(67), None]);
        filter_values.extend_from_slice(&[false, true, true, true]);
        let a = Int32Array::from(data_values);
        let b = BooleanArray::from(filter_values);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(67, d.len());
        assert_eq!(3, d.null_count());
        assert_eq!(1, d.value(0));
        assert!(d.is_null(1));
        assert_eq!(64, d.value(63));
        assert!(d.is_null(64));
        assert_eq!(67, d.value(65));
    }

    #[test]
    fn test_filter_string_array_simple() {
        let a = StringArray::from(vec!["hello", " ", "world", "!"]);
        let b = BooleanArray::from(vec![true, false, true, false]);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(2, d.len());
        assert_eq!("hello", d.value(0));
        assert_eq!("world", d.value(1));
    }

    #[test]
    fn test_filter_primitive_array_with_null() {
        let a = Int32Array::from(vec![Some(5), None]);
        let b = BooleanArray::from(vec![false, true]);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(1, d.len());
        assert!(d.is_null(0));
    }

    #[test]
    fn test_filter_string_array_with_null() {
        let a = StringArray::from(vec![Some("hello"), None, Some("world"), None]);
        let b = BooleanArray::from(vec![true, false, false, true]);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(2, d.len());
        assert_eq!("hello", d.value(0));
        assert!(!d.is_null(0));
        assert!(d.is_null(1));
    }

    #[test]
    fn test_filter_binary_array_with_null() {
        let data: Vec<Option<&[u8]>> = vec![Some(b"hello"), None, Some(b"world"), None];
        let a = BinaryArray::from(data);
        let b = BooleanArray::from(vec![true, false, false, true]);
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<BinaryArray>().unwrap();
        assert_eq!(2, d.len());
        assert_eq!(b"hello", d.value(0));
        assert!(!d.is_null(0));
        assert!(d.is_null(1));
    }

    fn _test_filter_byte_view<T>()
    where
        T: ByteViewType,
        str: AsRef<T::Native>,
        T::Native: PartialEq,
    {
        let array = {
            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
            let mut builder = GenericByteViewBuilder::<T>::new();
            builder.append_value("hello");
            builder.append_value("world");
            builder.append_null();
            builder.append_value("large payload over 12 bytes");
            builder.append_value("lulu");
            builder.finish()
        };

        {
            let predicate = BooleanArray::from(vec![true, false, true, true, false]);
            let actual = filter(&array, &predicate).unwrap();

            assert_eq!(actual.len(), 3);
            let actual_buffers = actual.as_byte_view::<T>().data_buffers();
            let input_buffers = array.data_buffers();
            assert!(Arc::ptr_eq(actual_buffers, input_buffers));

            let expected = {
                // ["hello", null, "large payload over 12 bytes"]
                let mut builder = GenericByteViewBuilder::<T>::new();
                builder.append_value("hello");
                builder.append_null();
                builder.append_value("large payload over 12 bytes");
                builder.finish()
            };

            assert_eq!(actual.as_ref(), &expected);
        }

        {
            let predicate = BooleanArray::from(vec![true, false, false, false, true]);
            let actual = filter(&array, &predicate).unwrap();

            assert_eq!(actual.len(), 2);

            let expected = {
                // ["hello", "lulu"]
                let mut builder = GenericByteViewBuilder::<T>::new();
                builder.append_value("hello");
                builder.append_value("lulu");
                builder.finish()
            };

            assert_eq!(actual.as_ref(), &expected);
        }
    }

    #[test]
    fn test_filter_string_view() {
        _test_filter_byte_view::<StringViewType>()
    }

    #[test]
    fn test_filter_binary_view() {
        _test_filter_byte_view::<BinaryViewType>()
    }

    #[test]
    fn test_filter_fixed_binary() {
        let v1 = [1_u8, 2];
        let v2 = [3_u8, 4];
        let v3 = [5_u8, 6];
        let v = vec![&v1, &v2, &v3];
        let a = FixedSizeBinaryArray::try_from(v).unwrap();
        let b = BooleanArray::from(vec![true, false, true]);
        let c = filter(&a, &b).unwrap();
        let d = c
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d.len(), 2);
        assert_eq!(d.value(0), &v1);
        assert_eq!(d.value(1), &v3);
        let c2 = FilterBuilder::new(&b)
            .optimize()
            .build()
            .filter(&a)
            .unwrap();
        let d2 = c2
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d, d2);

        let b = BooleanArray::from(vec![false, false, false]);
        let c = filter(&a, &b).unwrap();
        let d = c
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d.len(), 0);

        let b = BooleanArray::from(vec![true, true, true]);
        let c = filter(&a, &b).unwrap();
        let d = c
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d.len(), 3);
        assert_eq!(d.value(0), &v1);
        assert_eq!(d.value(1), &v2);
        assert_eq!(d.value(2), &v3);

        let b = BooleanArray::from(vec![false, false, true]);
        let c = filter(&a, &b).unwrap();
        let d = c
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d.len(), 1);
        assert_eq!(d.value(0), &v3);
        let c2 = FilterBuilder::new(&b)
            .optimize()
            .build()
            .filter(&a)
            .unwrap();
        let d2 = c2
            .as_ref()
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .unwrap();
        assert_eq!(d, d2);
    }

    #[test]
    fn test_filter_array_slice_with_null() {
        let a = Int32Array::from(vec![Some(5), None, Some(7), Some(8), Some(9)]).slice(1, 4);
        let b = BooleanArray::from(vec![true, false, false, true]);
        // filtering with sliced filter array is not currently supported
        // let b_slice = BooleanArray::from(vec![true, false, false, true, false]).slice(1, 4);
        // let b = b_slice.as_any().downcast_ref().unwrap();
        let c = filter(&a, &b).unwrap();
        let d = c.as_ref().as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(2, d.len());
        assert!(d.is_null(0));
        assert!(!d.is_null(1));
        assert_eq!(9, d.value(1));
    }

    #[test]
    fn test_filter_run_end_encoding_array() {
        let run_ends = Int64Array::from(vec![2, 3, 8]);
        let values = Int64Array::from(vec![7, -2, 9]);
        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
        let b = BooleanArray::from(vec![true, false, true, false, true, false, true, false]);
        let c = filter(&a, &b).unwrap();
        let actual: &RunArray<Int64Type> = as_run_array(&c);
        assert_eq!(4, actual.len());

        let expected = RunArray::try_new(
            &Int64Array::from(vec![1, 2, 4]),
            &Int64Array::from(vec![7, -2, 9]),
        )
        .expect("Failed to make expected RunArray test is broken");

        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
        assert_eq!(actual.values(), expected.values())
    }

    #[test]
    fn test_filter_run_end_encoding_array_sliced() {
        let run_ends = Int64Array::from(vec![2, 3, 8]);
        let values = Int64Array::from(vec![7, -2, 9]);
        let a = RunArray::try_new(&run_ends, &values).unwrap(); // [7, 7, -2, 9, 9, 9, 9, 9]
        let a = a.slice(2, 3); // [-2, 9, 9]
        let b = BooleanArray::from(vec![true, false, true]);
        let result = filter(&a, &b).unwrap();

        let result = result.as_run::<Int64Type>();
        let result = result.downcast::<Int64Array>().unwrap();

        let expected = vec![-2, 9];
        let actual = result.into_iter().flatten().collect::<Vec<_>>();
        assert_eq!(expected, actual);
    }

    #[test]
    fn test_filter_run_end_encoding_array_remove_value() {
        let run_ends = Int32Array::from(vec![2, 3, 8, 10]);
        let values = Int32Array::from(vec![7, -2, 9, -8]);
        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
        let b = BooleanArray::from(vec![
            false, true, false, false, true, false, true, false, false, false,
        ]);
        let c = filter(&a, &b).unwrap();
        let actual: &RunArray<Int32Type> = as_run_array(&c);
        assert_eq!(3, actual.len());

        let expected =
            RunArray::try_new(&Int32Array::from(vec![1, 3]), &Int32Array::from(vec![7, 9]))
                .expect("Failed to make expected RunArray test is broken");

        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
        assert_eq!(actual.values(), expected.values())
    }

    #[test]
    fn test_filter_run_end_encoding_array_remove_all_but_one() {
        let run_ends = Int16Array::from(vec![2, 3, 8, 10]);
        let values = Int16Array::from(vec![7, -2, 9, -8]);
        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
        let b = BooleanArray::from(vec![
            false, false, false, false, false, false, true, false, false, false,
        ]);
        let c = filter(&a, &b).unwrap();
        let actual: &RunArray<Int16Type> = as_run_array(&c);
        assert_eq!(1, actual.len());

        let expected = RunArray::try_new(&Int16Array::from(vec![1]), &Int16Array::from(vec![9]))
            .expect("Failed to make expected RunArray test is broken");

        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
        assert_eq!(actual.values(), expected.values())
    }

    #[test]
    fn test_filter_run_end_encoding_array_empty() {
        let run_ends = Int64Array::from(vec![2, 3, 8, 10]);
        let values = Int64Array::from(vec![7, -2, 9, -8]);
        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
        let b = BooleanArray::from(vec![
            false, false, false, false, false, false, false, false, false, false,
        ]);
        let c = filter(&a, &b).unwrap();
        let actual: &RunArray<Int64Type> = as_run_array(&c);
        assert_eq!(0, actual.len());
    }

    #[test]
    fn test_filter_run_end_encoding_array_max_value_gt_predicate_len() {
        let run_ends = Int64Array::from(vec![2, 3, 8, 10]);
        let values = Int64Array::from(vec![7, -2, 9, -8]);
        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
        let b = BooleanArray::from(vec![false, true, true]);
        let c = filter(&a, &b).unwrap();
        let actual: &RunArray<Int64Type> = as_run_array(&c);
        assert_eq!(2, actual.len());

        let expected = RunArray::try_new(
            &Int64Array::from(vec![1, 2]),
            &Int64Array::from(vec![7, -2]),
        )
        .expect("Failed to make expected RunArray test is broken");

        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
        assert_eq!(actual.values(), expected.values())
    }

    #[test]
    fn test_filter_dictionary_array() {
        let values = [Some("hello"), None, Some("world"), Some("!")];
        let a: Int8DictionaryArray = values.iter().copied().collect();
        let b = BooleanArray::from(vec![false, true, true, false]);
        let c = filter(&a, &b).unwrap();
        let d = c
            .as_ref()
            .as_any()
            .downcast_ref::<Int8DictionaryArray>()
            .unwrap();
        let value_array = d.values();
        let values = value_array.as_any().downcast_ref::<StringArray>().unwrap();
        // values are cloned in the filtered dictionary array
        assert_eq!(3, values.len());
        // but keys are filtered
        assert_eq!(2, d.len());
        assert!(d.is_null(0));
        assert_eq!("world", values.value(d.keys().value(1) as usize));
    }

    #[test]
    fn test_filter_list_array() {
        let field = Arc::new(Field::new_list_field(DataType::Int32, false));
        let offsets = OffsetBuffer::new(vec![0i64, 3, 6, 8, 8].into());
        let value_array = Arc::new(Int32Array::from_iter_values(0..8));
        let nulls = Some(NullBuffer::from(vec![true, true, true, false]));
        //  a = [[0, 1, 2], [3, 4, 5], [6, 7], null]
        let a = LargeListArray::new(field.clone(), offsets, value_array, nulls);
        let b = BooleanArray::from(vec![false, true, false, true]);
        let result = filter(&a, &b).unwrap();

        // expected: [[3, 4, 5], null]
        let offsets = OffsetBuffer::new(vec![0i64, 3, 3].into());
        let value_array = Arc::new(Int32Array::from_iter_values([3, 4, 5]));
        let nulls = Some(NullBuffer::from(vec![true, false]));
        let expected: ArrayRef = Arc::new(LargeListArray::new(field, offsets, value_array, nulls));

        assert_eq!(&expected, &result);
    }

    fn test_case_filter_list_view<T: OffsetSizeTrait>() {
        // [[1, 2], null, [], [3,4]]
        let mut list_array = GenericListViewBuilder::<T, _>::new(Int32Builder::new());
        list_array.append_value([Some(1), Some(2)]);
        list_array.append_null();
        list_array.append_value([]);
        list_array.append_value([Some(3), Some(4)]);

        let list_array = list_array.finish();
        let predicate = BooleanArray::from_iter([true, false, true, false]);

        // Filter result: [[1, 2], []]
        let filtered = filter(&list_array, &predicate)
            .unwrap()
            .as_list_view::<T>()
            .clone();

        let mut expected =
            GenericListViewBuilder::<T, _>::with_capacity(Int32Builder::with_capacity(5), 3);
        expected.append_value([Some(1), Some(2)]);
        expected.append_value([]);
        let expected = expected.finish();

        assert_eq!(&filtered, &expected);
    }

    fn test_case_filter_sliced_list_view<T: OffsetSizeTrait>() {
        // [[1, 2], null, [], [3,4]]
        let mut list_array =
            GenericListViewBuilder::<T, _>::with_capacity(Int32Builder::with_capacity(6), 4);
        list_array.append_value([Some(1), Some(2)]);
        list_array.append_null();
        list_array.append_value([]);
        list_array.append_value([Some(3), Some(4)]);

        let list_array = list_array.finish();

        // Sliced: [null, [], [3, 4]]
        let sliced = list_array.slice(1, 3);
        let predicate = BooleanArray::from_iter([false, false, true]);

        // Filter result: [[1, 2], []]
        let filtered = filter(&sliced, &predicate)
            .unwrap()
            .as_list_view::<T>()
            .clone();

        let mut expected = GenericListViewBuilder::<T, _>::new(Int32Builder::new());
        expected.append_value([Some(3), Some(4)]);
        let expected = expected.finish();

        assert_eq!(&filtered, &expected);
    }

    #[test]
    fn test_filter_list_view_array() {
        test_case_filter_list_view::<i32>();
        test_case_filter_list_view::<i64>();

        test_case_filter_sliced_list_view::<i32>();
        test_case_filter_sliced_list_view::<i64>();
    }

    /// Tests [`filter_bits_compress`] and [`filter_bits_strategy`] on the
    /// same inputs against a naive bit-by-bit filter, verifying both pathways
    /// produce the same output. Both are called directly rather than through
    /// [`filter_bits`], whose dispatch depends on the filter density, so both
    /// get coverage on every input
    #[test]
    fn test_filter_bits() {
        let mut rng = StdRng::seed_from_u64(42);

        // Lengths exercising partial words, exact word multiples, and the
        // carry logic across flushed words
        let lens = [0, 1, 7, 63, 64, 65, 127, 128, 200, 1024, 4099];
        // Densities covering empty, sparse, balanced, dense and full masks
        let densities = [0.0, 0.01, 0.5, 0.9, 1.0];
        // Bit offsets of the value buffer, including non byte-aligned ones
        let offsets = [0, 3, 8, 67];
        // Bit offsets of the filter, so the mask words are read unaligned too
        let filter_offsets = [0, 5];

        for len in lens {
            for density in densities {
                for offset in offsets {
                    for filter_offset in filter_offsets {
                        let values: BooleanBuffer =
                            (0..len + offset).map(|_| rng.random_bool(0.5)).collect();
                        let values = values.slice(offset, len);
                        let filter: BooleanArray = (0..len + filter_offset)
                            .map(|_| Some(rng.random_bool(density)))
                            .collect();
                        let filter = filter.slice(filter_offset, len);

                        let expected: BooleanBuffer = values
                            .iter()
                            .zip(filter.values().iter())
                            .filter_map(|(value, keep)| keep.then_some(value))
                            .collect();

                        // Lazy and precomputed strategies dispatch differently
                        let predicates = [
                            FilterBuilder::new(&filter).build(),
                            FilterBuilder::new(&filter).optimize().build(),
                        ];
                        for predicate in &predicates {
                            let case = format!(
                                "{:?}: len={len} density={density} offset={offset} filter_offset={filter_offset}",
                                predicate.strategy
                            );

                            let compressed = filter_bits_compress(&values, predicate);
                            let compressed = BooleanBuffer::new(compressed, 0, predicate.count);
                            assert_eq!(compressed, expected, "compress {case}");

                            // `filter_bits` is never reached with the `All` /
                            // `None` strategies, they are short-circuited by
                            // the callers
                            if matches!(
                                predicate.strategy,
                                IterationStrategy::All | IterationStrategy::None
                            ) {
                                continue;
                            }

                            let strategy = filter_bits_strategy(&values, predicate);
                            let strategy = BooleanBuffer::new(strategy, 0, predicate.count);
                            assert_eq!(strategy, expected, "strategy {case}");

                            // Also cover the dispatch between the two pathways
                            let dispatched = filter_bits(&values, predicate);
                            let dispatched = BooleanBuffer::new(dispatched, 0, predicate.count);
                            assert_eq!(dispatched, expected, "dispatch {case}");
                        }
                    }
                }
            }
        }
    }

    #[test]
    fn test_slice_iterator_bits() {
        let filter_values = (0..64).map(|i| i == 1).collect::<Vec<bool>>();
        let filter = BooleanArray::from(filter_values);
        let filter_count = filter.true_count();

        let iter = SlicesIterator::new(&filter);
        let chunks = iter.collect::<Vec<_>>();

        assert_eq!(chunks, vec![(1, 2)]);
        assert_eq!(filter_count, 1);
    }

    #[test]
    fn test_slice_iterator_bits1() {
        let filter_values = (0..64).map(|i| i != 1).collect::<Vec<bool>>();
        let filter = BooleanArray::from(filter_values);
        let filter_count = filter.true_count();

        let iter = SlicesIterator::new(&filter);
        let chunks = iter.collect::<Vec<_>>();

        assert_eq!(chunks, vec![(0, 1), (2, 64)]);
        assert_eq!(filter_count, 64 - 1);
    }

    #[test]
    fn test_slice_iterator_chunk_and_bits() {
        let filter_values = (0..130).map(|i| i % 62 != 0).collect::<Vec<bool>>();
        let filter = BooleanArray::from(filter_values);
        let filter_count = filter.true_count();

        let iter = SlicesIterator::new(&filter);
        let chunks = iter.collect::<Vec<_>>();

        assert_eq!(chunks, vec![(1, 62), (63, 124), (125, 130)]);
        assert_eq!(filter_count, 61 + 61 + 5);
    }

    #[test]
    fn test_filter_selection_iterators() {
        let slices = [(0, 2), (4, 5)];
        let mut ranges = Vec::new();
        let selection: FilterSlices<'_> = FilterIterator::Materialized(slices.iter().copied());
        selection.for_each(|range| ranges.push(range));
        assert_eq!(ranges, slices);

        let filter = BooleanArray::from(vec![true, true, false, false, true]);
        let mut ranges = Vec::new();
        let selection: FilterSlices<'_> = FilterIterator::Lazy(SlicesIterator::new(&filter));
        selection
            .try_for_each(|range| {
                ranges.push(range);
                Ok::<(), ArrowError>(())
            })
            .unwrap();
        assert_eq!(ranges, vec![(0, 2), (4, 5)]);

        let indices = [1, 3, 5];
        let mut selected = Vec::new();
        let selection: FilterIndices<'_> = FilterIterator::Materialized(indices.iter().copied());
        selection.for_each(|idx| selected.push(idx));
        assert_eq!(selected, indices);

        let filter = BooleanArray::from(vec![false, true, false, true]);
        let mut selected = Vec::new();
        let selection: FilterIndices<'_> = FilterIterator::Lazy(IndexIterator::new(&filter, 2));
        selection
            .try_for_each(|idx| {
                selected.push(idx);
                Ok::<(), ArrowError>(())
            })
            .unwrap();
        assert_eq!(selected, vec![1, 3]);
    }

    #[test]
    fn test_null_mask() {
        let a = Int64Array::from(vec![Some(1), Some(2), None]);

        let mask1 = BooleanArray::from(vec![Some(true), Some(true), None]);
        let out = filter(&a, &mask1).unwrap();
        assert_eq!(out.as_ref(), &a.slice(0, 2));
    }

    #[test]
    fn test_filter_record_batch_no_columns() {
        let pred = BooleanArray::from(vec![Some(true), Some(true), None]);
        let options = RecordBatchOptions::default().with_row_count(Some(100));
        let record_batch =
            RecordBatch::try_new_with_options(Arc::new(Schema::empty()), vec![], &options).unwrap();
        let out = filter_record_batch(&record_batch, &pred).unwrap();

        assert_eq!(out.num_rows(), 2);
    }

    #[test]
    fn test_fast_path() {
        let a: PrimitiveArray<Int64Type> = PrimitiveArray::from(vec![Some(1), Some(2), None]);

        // all true
        let mask = BooleanArray::from(vec![true, true, true]);
        let out = filter(&a, &mask).unwrap();
        let b = out
            .as_any()
            .downcast_ref::<PrimitiveArray<Int64Type>>()
            .unwrap();
        assert_eq!(&a, b);

        // all false
        let mask = BooleanArray::from(vec![false, false, false]);
        let out = filter(&a, &mask).unwrap();
        assert_eq!(out.len(), 0);
        assert_eq!(out.data_type(), &DataType::Int64);
    }

    #[test]
    fn test_slices() {
        // takes up 2 u64s
        let bools = std::iter::repeat_n(true, 10)
            .chain(std::iter::repeat_n(false, 30))
            .chain(std::iter::repeat_n(true, 20))
            .chain(std::iter::repeat_n(false, 17))
            .chain(std::iter::repeat_n(true, 4));

        let bool_array: BooleanArray = bools.map(Some).collect();

        let slices: Vec<_> = SlicesIterator::new(&bool_array).collect();
        let expected = vec![(0, 10), (40, 60), (77, 81)];
        assert_eq!(slices, expected);

        // slice with offset and truncated len
        let len = bool_array.len();
        let sliced_array = bool_array.slice(7, len - 10);
        let sliced_array = sliced_array
            .as_any()
            .downcast_ref::<BooleanArray>()
            .unwrap();
        let slices: Vec<_> = SlicesIterator::new(sliced_array).collect();
        let expected = vec![(0, 3), (33, 53), (70, 71)];
        assert_eq!(slices, expected);
    }

    fn test_slices_fuzz(mask_len: usize, offset: usize, truncate: usize) {
        let mut rng = rng();

        let bools: Vec<bool> = std::iter::from_fn(|| Some(rng.random()))
            .take(mask_len)
            .collect();

        let buffer = Buffer::from_iter(bools.iter().copied());

        let truncated_length = mask_len - offset - truncate;

        let filter = BooleanArray::new(BooleanBuffer::new(buffer, offset, truncated_length), None);

        let slice_bits: Vec<_> = SlicesIterator::new(&filter)
            .flat_map(|(start, end)| start..end)
            .collect();

        let count = filter.true_count();
        let index_bits: Vec<_> = IndexIterator::new(&filter, count).collect();

        let expected_bits: Vec<_> = bools
            .iter()
            .skip(offset)
            .take(truncated_length)
            .enumerate()
            .filter_map(|(idx, v)| v.then_some(idx))
            .collect();

        assert_eq!(slice_bits, expected_bits);
        assert_eq!(index_bits, expected_bits);
    }

    #[test]
    #[cfg_attr(miri, ignore)] // Takes too long
    fn fuzz_test_slices_iterator() {
        let mut rng = rng();

        let uusize = UniformUsize::new(usize::MIN, usize::MAX).unwrap();
        for _ in 0..100 {
            let mask_len = rng.random_range(0..1024);
            let max_offset = 64.min(mask_len);
            let offset = uusize.sample(&mut rng).checked_rem(max_offset).unwrap_or(0);

            let max_truncate = 128.min(mask_len - offset);
            let truncate = uusize
                .sample(&mut rng)
                .checked_rem(max_truncate)
                .unwrap_or(0);

            test_slices_fuzz(mask_len, offset, truncate);
        }

        test_slices_fuzz(64, 0, 0);
        test_slices_fuzz(64, 8, 0);
        test_slices_fuzz(64, 8, 8);
        test_slices_fuzz(32, 8, 8);
        test_slices_fuzz(32, 5, 9);
    }

    /// Filters `values` by `predicate` using standard rust iterators
    fn filter_rust<T>(values: impl IntoIterator<Item = T>, predicate: &[bool]) -> Vec<T> {
        values
            .into_iter()
            .zip(predicate)
            .filter(|(_, x)| **x)
            .map(|(a, _)| a)
            .collect()
    }

    /// Generates an array of length `len` with `valid_percent` non-null values
    fn gen_primitive<T>(len: usize, valid_percent: f64) -> Vec<Option<T>>
    where
        StandardUniform: Distribution<T>,
    {
        let mut rng = rng();
        (0..len)
            .map(|_| rng.random_bool(valid_percent).then(|| rng.random()))
            .collect()
    }

    /// Generates an array of length `len` with `valid_percent` non-null values
    fn gen_strings(
        len: usize,
        valid_percent: f64,
        str_len_range: std::ops::Range<usize>,
    ) -> Vec<Option<String>> {
        let mut rng = rng();
        (0..len)
            .map(|_| {
                rng.random_bool(valid_percent).then(|| {
                    let len = rng.random_range(str_len_range.clone());
                    (0..len)
                        .map(|_| char::from(rng.sample(Alphanumeric)))
                        .collect()
                })
            })
            .collect()
    }

    /// Returns an iterator that calls `Option::as_deref` on each item
    fn as_deref<T: std::ops::Deref>(src: &[Option<T>]) -> impl Iterator<Item = Option<&T::Target>> {
        src.iter().map(|x| x.as_deref())
    }

    #[test]
    #[cfg_attr(miri, ignore)] // Takes too long
    fn fuzz_filter() {
        let mut rng = rng();

        for i in 0..100 {
            let filter_percent = match i {
                0..=4 => 1.,
                5..=10 => 0.,
                _ => rng.random_range(0.0..1.0),
            };

            let valid_percent = rng.random_range(0.0..1.0);

            let array_len = rng.random_range(32..256);
            let array_offset = rng.random_range(0..10);

            // Construct a predicate
            let filter_offset = rng.random_range(0..10);
            let filter_truncate = rng.random_range(0..10);
            let bools: Vec<_> = std::iter::from_fn(|| Some(rng.random_bool(filter_percent)))
                .take(array_len + filter_offset - filter_truncate)
                .collect();

            let predicate = BooleanArray::from_iter(bools.iter().copied().map(Some));

            // Offset predicate
            let predicate = predicate.slice(filter_offset, array_len - filter_truncate);
            let predicate = predicate.as_any().downcast_ref::<BooleanArray>().unwrap();
            let bools = &bools[filter_offset..];

            // Test i32
            let values = gen_primitive(array_len + array_offset, valid_percent);
            let src = Int32Array::from_iter(values.iter().copied());

            let src = src.slice(array_offset, array_len);
            let src = src.as_any().downcast_ref::<Int32Array>().unwrap();
            let values = &values[array_offset..];

            let filtered = filter(src, predicate).unwrap();
            let array = filtered.as_any().downcast_ref::<Int32Array>().unwrap();
            let actual: Vec<_> = array.iter().collect();

            assert_eq!(actual, filter_rust(values.iter().copied(), bools));

            // Test string
            let strings = gen_strings(array_len + array_offset, valid_percent, 0..20);
            let src = StringArray::from_iter(as_deref(&strings));

            let src = src.slice(array_offset, array_len);
            let src = src.as_any().downcast_ref::<StringArray>().unwrap();

            let filtered = filter(src, predicate).unwrap();
            let array = filtered.as_any().downcast_ref::<StringArray>().unwrap();
            let actual: Vec<_> = array.iter().collect();

            let expected_strings = filter_rust(as_deref(&strings[array_offset..]), bools);
            assert_eq!(actual, expected_strings);

            // Test string dictionary
            let src = DictionaryArray::<Int32Type>::from_iter(as_deref(&strings));

            let src = src.slice(array_offset, array_len);
            let src = src
                .as_any()
                .downcast_ref::<DictionaryArray<Int32Type>>()
                .unwrap();

            let filtered = filter(src, predicate).unwrap();

            let array = filtered
                .as_any()
                .downcast_ref::<DictionaryArray<Int32Type>>()
                .unwrap();

            let values = array
                .values()
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();

            let actual: Vec<_> = array
                .keys()
                .iter()
                .map(|key| key.map(|key| values.value(key as usize)))
                .collect();

            assert_eq!(actual, expected_strings);
        }
    }

    #[test]
    fn test_filter_map() {
        let mut builder =
            MapBuilder::new(None, StringBuilder::new(), Int64Builder::with_capacity(4));
        // [{"key1": 1}, {"key2": 2, "key3": 3}, null, {"key1": 1}
        builder.keys().append_value("key1");
        builder.values().append_value(1);
        builder.append(true).unwrap();
        builder.keys().append_value("key2");
        builder.keys().append_value("key3");
        builder.values().append_value(2);
        builder.values().append_value(3);
        builder.append(true).unwrap();
        builder.append(false).unwrap();
        builder.keys().append_value("key1");
        builder.values().append_value(1);
        builder.append(true).unwrap();
        let maparray = Arc::new(builder.finish()) as ArrayRef;

        let indices = vec![Some(true), Some(false), Some(false), Some(true)]
            .into_iter()
            .collect::<BooleanArray>();
        let got = filter(&maparray, &indices).unwrap();

        let mut builder =
            MapBuilder::new(None, StringBuilder::new(), Int64Builder::with_capacity(2));
        builder.keys().append_value("key1");
        builder.values().append_value(1);
        builder.append(true).unwrap();
        builder.keys().append_value("key1");
        builder.values().append_value(1);
        builder.append(true).unwrap();
        let expected = Arc::new(builder.finish()) as ArrayRef;

        assert_eq!(&expected, &got);
    }

    #[test]
    fn test_filter_fixed_size_list_arrays() {
        let field = Arc::new(Field::new_list_field(DataType::Int32, false));
        let value_array = Arc::new(Int32Array::from_iter_values(0..9));
        let array = FixedSizeListArray::new(field, 3, value_array, None);

        let filter_array = BooleanArray::from(vec![true, false, false]);

        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<FixedSizeListArray>().unwrap();

        assert_eq!(filtered.len(), 1);

        let list = filtered.value(0);
        assert_eq!(
            &[0, 1, 2],
            list.as_any().downcast_ref::<Int32Array>().unwrap().values()
        );

        let filter_array = BooleanArray::from(vec![true, false, true]);

        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<FixedSizeListArray>().unwrap();

        assert_eq!(filtered.len(), 2);

        let list = filtered.value(0);
        assert_eq!(
            &[0, 1, 2],
            list.as_any().downcast_ref::<Int32Array>().unwrap().values()
        );
        let list = filtered.value(1);
        assert_eq!(
            &[6, 7, 8],
            list.as_any().downcast_ref::<Int32Array>().unwrap().values()
        );
    }

    #[test]
    fn test_filter_fixed_size_list_arrays_with_null() {
        let field = Arc::new(Field::new_list_field(DataType::Int32, false));
        let value_array = Arc::new(Int32Array::from_iter_values(0..10));
        let nulls = Some(NullBuffer::from(vec![true, false, false, true, true]));
        let array = FixedSizeListArray::new(field, 2, value_array, nulls);

        let filter_array = BooleanArray::from(vec![true, true, false, true, false]);

        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<FixedSizeListArray>().unwrap();

        assert_eq!(filtered.len(), 3);

        let list = filtered.value(0);
        assert_eq!(
            &[0, 1],
            list.as_any().downcast_ref::<Int32Array>().unwrap().values()
        );
        assert!(filtered.is_null(1));
        let list = filtered.value(2);
        assert_eq!(
            &[6, 7],
            list.as_any().downcast_ref::<Int32Array>().unwrap().values()
        );
    }

    #[test]
    fn test_filter_zero_width_fixed_size_binary() {
        // value_length=0 with no nulls: row count cannot be inferred from the empty
        // buffer, so filter must preserve it explicitly.
        let array = FixedSizeBinaryArray::try_new_with_len(
            0,
            Buffer::from_slice_ref(&[] as &[u8]),
            None,
            3,
        )
        .unwrap();
        let filter_array = BooleanArray::from(vec![true, false, true]);
        let result = filter(&array, &filter_array).unwrap();
        assert_eq!(result.len(), 2);
    }

    fn test_filter_union_array(array: UnionArray) {
        let filter_array = BooleanArray::from(vec![true, false, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);

        let filter_array = BooleanArray::from(vec![true, false, true]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);

        let filter_array = BooleanArray::from(vec![true, true, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);
    }

    #[test]
    fn test_filter_union_array_dense() {
        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let array = builder.build().unwrap();

        test_filter_union_array(array);
    }

    #[test]
    fn test_filter_run_union_array_dense() {
        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Int32Type>("A", 3).unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let array = builder.build().unwrap();

        let filter_array = BooleanArray::from(vec![true, true, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Int32Type>("A", 3).unwrap();
        let expected = builder.build().unwrap();

        assert_eq!(filtered.to_data(), expected.to_data());
    }

    #[test]
    fn test_filter_union_array_dense_with_nulls() {
        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        builder.append_null::<Float64Type>("B").unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let array = builder.build().unwrap();

        let filter_array = BooleanArray::from(vec![true, true, false, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);

        let filter_array = BooleanArray::from(vec![true, false, true, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_dense();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append_null::<Float64Type>("B").unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);
    }

    #[test]
    fn test_filter_union_array_sparse() {
        let mut builder = UnionBuilder::new_sparse();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let array = builder.build().unwrap();

        test_filter_union_array(array);
    }

    #[test]
    fn test_filter_union_array_sparse_with_nulls() {
        let mut builder = UnionBuilder::new_sparse();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append::<Float64Type>("B", 3.2).unwrap();
        builder.append_null::<Float64Type>("B").unwrap();
        builder.append::<Int32Type>("A", 34).unwrap();
        let array = builder.build().unwrap();

        let filter_array = BooleanArray::from(vec![true, false, true, false]);
        let c = filter(&array, &filter_array).unwrap();
        let filtered = c.as_any().downcast_ref::<UnionArray>().unwrap();

        let mut builder = UnionBuilder::new_sparse();
        builder.append::<Int32Type>("A", 1).unwrap();
        builder.append_null::<Float64Type>("B").unwrap();
        let expected_array = builder.build().unwrap();

        compare_union_arrays(filtered, &expected_array);
    }

    fn compare_union_arrays(union1: &UnionArray, union2: &UnionArray) {
        assert_eq!(union1.len(), union2.len());

        for i in 0..union1.len() {
            let type_id = union1.type_id(i);

            let slot1 = union1.value(i);
            let slot2 = union2.value(i);

            assert_eq!(slot1.is_null(0), slot2.is_null(0));

            if !slot1.is_null(0) && !slot2.is_null(0) {
                match type_id {
                    0 => {
                        let slot1 = slot1.as_any().downcast_ref::<Int32Array>().unwrap();
                        assert_eq!(slot1.len(), 1);
                        let value1 = slot1.value(0);

                        let slot2 = slot2.as_any().downcast_ref::<Int32Array>().unwrap();
                        assert_eq!(slot2.len(), 1);
                        let value2 = slot2.value(0);
                        assert_eq!(value1, value2);
                    }
                    1 => {
                        let slot1 = slot1.as_any().downcast_ref::<Float64Array>().unwrap();
                        assert_eq!(slot1.len(), 1);
                        let value1 = slot1.value(0);

                        let slot2 = slot2.as_any().downcast_ref::<Float64Array>().unwrap();
                        assert_eq!(slot2.len(), 1);
                        let value2 = slot2.value(0);
                        assert_eq!(value1, value2);
                    }
                    _ => unreachable!(),
                }
            }
        }
    }

    #[test]
    fn test_filter_struct() {
        let predicate = BooleanArray::from(vec![true, false, true, false]);

        let a = Arc::new(StringArray::from(vec!["hello", " ", "world", "!"]));
        let a_filtered = Arc::new(StringArray::from(vec!["hello", "world"]));

        let b = Arc::new(Int32Array::from(vec![5, 6, 7, 8]));
        let b_filtered = Arc::new(Int32Array::from(vec![5, 7]));

        let null_mask = NullBuffer::from(vec![true, false, false, true]);
        let null_mask_filtered = NullBuffer::from(vec![true, false]);

        let a_field = Field::new("a", DataType::Utf8, false);
        let b_field = Field::new("b", DataType::Int32, false);

        let array = StructArray::new(vec![a_field.clone()].into(), vec![a.clone()], None);
        let expected =
            StructArray::new(vec![a_field.clone()].into(), vec![a_filtered.clone()], None);

        let result = filter(&array, &predicate).unwrap();

        assert_eq!(result.to_data(), expected.to_data());

        let array = StructArray::new(
            vec![a_field.clone()].into(),
            vec![a.clone()],
            Some(null_mask.clone()),
        );
        let expected = StructArray::new(
            vec![a_field.clone()].into(),
            vec![a_filtered.clone()],
            Some(null_mask_filtered.clone()),
        );

        let result = filter(&array, &predicate).unwrap();

        assert_eq!(result.to_data(), expected.to_data());

        let array = StructArray::new(
            vec![a_field.clone(), b_field.clone()].into(),
            vec![a.clone(), b.clone()],
            None,
        );
        let expected = StructArray::new(
            vec![a_field.clone(), b_field.clone()].into(),
            vec![a_filtered.clone(), b_filtered.clone()],
            None,
        );

        let result = filter(&array, &predicate).unwrap();

        assert_eq!(result.to_data(), expected.to_data());

        let array = StructArray::new(
            vec![a_field.clone(), b_field.clone()].into(),
            vec![a.clone(), b.clone()],
            Some(null_mask.clone()),
        );

        let expected = StructArray::new(
            vec![a_field.clone(), b_field.clone()].into(),
            vec![a_filtered.clone(), b_filtered.clone()],
            Some(null_mask_filtered.clone()),
        );

        let result = filter(&array, &predicate).unwrap();

        assert_eq!(result.to_data(), expected.to_data());
    }

    #[test]
    fn test_filter_empty_struct() {
        /*
            "a": {
                "b": int64,
                "c": {}
            },
        */
        let fields = arrow_schema::Field::new(
            "a",
            arrow_schema::DataType::Struct(arrow_schema::Fields::from(vec![
                arrow_schema::Field::new("b", arrow_schema::DataType::Int64, true),
                arrow_schema::Field::new(
                    "c",
                    arrow_schema::DataType::Struct(arrow_schema::Fields::empty()),
                    true,
                ),
            ])),
            true,
        );

        /* Test record
            {"a":{"c": {}}}
            {"a":{"c": {}}}
            {"a":{"c": {}}}
        */

        // Create the record batch with the nested struct array
        let schema = Arc::new(Schema::new(vec![fields]));

        let b = Arc::new(Int64Array::from(vec![None, None, None]));
        let c = Arc::new(StructArray::new_empty_fields(
            3,
            Some(NullBuffer::from(vec![true, true, true])),
        ));
        let a = StructArray::new(
            vec![
                Field::new("b", DataType::Int64, true),
                Field::new("c", DataType::Struct(Fields::empty()), true),
            ]
            .into(),
            vec![b.clone(), c.clone()],
            Some(NullBuffer::from(vec![true, true, true])),
        );
        let record_batch = RecordBatch::try_new(schema, vec![Arc::new(a)]).unwrap();
        println!("{record_batch:?}");

        // Apply the filter
        let predicate = BooleanArray::from(vec![true, false, true]);
        let filtered_batch = filter_record_batch(&record_batch, &predicate).unwrap();

        // The filtered batch should have 2 rows (the 1st and 3rd)
        assert_eq!(filtered_batch.num_rows(), 2);
    }

    #[test]
    #[should_panic(expected = "buffer.len() >= predicate.filter.len()")]
    fn test_filter_bits_too_large() {
        let buffer = BooleanBuffer::from(vec![false; 8]);
        let predicate = BooleanArray::from(vec![true; 9]);
        let filter = FilterBuilder::new(&predicate).build();
        filter_bits(&buffer, &filter);
    }

    #[test]
    #[should_panic(expected = "values.len() >= predicate.filter.len()")]
    fn test_filter_native_too_large() {
        let values = vec![1; 8];
        let predicate = BooleanArray::from(vec![false; 9]);
        let filter = FilterBuilder::new(&predicate).build();
        filter_native(&values, &filter);
    }
}