Skip to content

Commit 6a1869f

Browse files
Implement Branchless Filter for small primitive lists
Adds a const-generic unrolled comparison chain that avoids CPU branching. Outperforms hash lookups for very small lists. Triggers for primitives when list size <= 32 (4-byte), 16 (8-byte), or 4 (16-byte).
1 parent 08fbe39 commit 6a1869f

3 files changed

Lines changed: 341 additions & 27 deletions

File tree

datafusion/physical-expr/src/expressions/in_list/primitive_filter.rs

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -176,6 +176,98 @@ impl<C: BitmapFilterConfig> StaticFilter for BitmapFilter<C> {
176176
}
177177
}
178178

179+
/// A branchless filter for very small fixed-width primitive IN lists.
180+
///
181+
/// Uses const generics to unroll the membership check into a fixed-size
182+
/// comparison chain, outperforming hash lookups for small lists due to:
183+
/// - No branching (uses bitwise OR to combine comparisons)
184+
/// - Better CPU pipelining
185+
/// - No hash computation overhead
186+
pub(super) struct BranchlessFilter<T: ArrowPrimitiveType, const N: usize> {
187+
null_count: usize,
188+
values: [T::Native; N],
189+
}
190+
191+
impl<T: ArrowPrimitiveType, const N: usize> BranchlessFilter<T, N>
192+
where
193+
T::Native: Copy + PartialEq,
194+
{
195+
/// Try to create a branchless filter if the array has exactly N non-null values.
196+
pub(super) fn try_new(in_array: &ArrayRef) -> Option<Result<Self>> {
197+
let in_array = in_array.as_primitive_opt::<T>()?;
198+
let non_null_count = in_array.len() - in_array.null_count();
199+
if non_null_count != N {
200+
return None;
201+
}
202+
// Use default_value() from ArrowPrimitiveType trait instead of Default::default()
203+
let mut arr = [T::default_value(); N];
204+
let mut i = 0;
205+
for value in in_array.iter().flatten() {
206+
arr[i] = value;
207+
i += 1;
208+
}
209+
debug_assert_eq!(i, N);
210+
Some(Ok(Self {
211+
null_count: in_array.null_count(),
212+
values: arr,
213+
}))
214+
}
215+
216+
/// Branchless membership check using OR-chain.
217+
#[inline(always)]
218+
fn check(&self, needle: T::Native) -> bool {
219+
self.values
220+
.iter()
221+
.fold(false, |acc, &v| acc | (v == needle))
222+
}
223+
224+
/// Check membership using a raw values slice (zero-copy path for type reinterpretation).
225+
#[inline]
226+
pub(super) fn contains_slice(
227+
&self,
228+
values: &[T::Native],
229+
nulls: Option<&NullBuffer>,
230+
negated: bool,
231+
) -> BooleanArray {
232+
build_in_list_result(values.len(), nulls, self.null_count > 0, negated, |i| {
233+
// SAFETY: `build_in_list_result` invokes this closure for
234+
// indices in `0..values.len()`.
235+
let needle = unsafe { *values.get_unchecked(i) };
236+
self.check(needle)
237+
})
238+
}
239+
}
240+
241+
impl<T: ArrowPrimitiveType, const N: usize> StaticFilter for BranchlessFilter<T, N>
242+
where
243+
T::Native: Copy + PartialEq + Send + Sync,
244+
{
245+
fn null_count(&self) -> usize {
246+
self.null_count
247+
}
248+
249+
fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
250+
handle_dictionary!(self, v, negated);
251+
let v = v.as_primitive_opt::<T>().ok_or_else(|| {
252+
exec_datafusion_err!("Failed to downcast array to primitive type")
253+
})?;
254+
let input_values = v.values();
255+
Ok(build_in_list_result(
256+
v.len(),
257+
v.nulls(),
258+
self.null_count > 0,
259+
negated,
260+
#[inline(always)]
261+
|i| {
262+
// SAFETY: `build_in_list_result` invokes this closure for
263+
// indices in `0..v.len()`, which matches `input_values.len()`.
264+
let needle = unsafe { *input_values.get_unchecked(i) };
265+
self.check(needle)
266+
},
267+
))
268+
}
269+
}
270+
179271
/// Wrapper for f32 that implements Hash and Eq using bit comparison.
180272
/// This treats NaN values as equal to each other when they have the same bit pattern.
181273
#[derive(Clone, Copy)]

datafusion/physical-expr/src/expressions/in_list/strategy.rs

Lines changed: 96 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,66 @@ use std::sync::Arc;
1919

2020
use arrow::array::ArrayRef;
2121
use arrow::compute::cast;
22-
use arrow::datatypes::DataType;
23-
use datafusion_common::Result;
22+
use arrow::datatypes::*;
23+
use datafusion_common::{Result, exec_datafusion_err};
2424

2525
use super::array_static_filter::ArrayStaticFilter;
2626
use super::primitive_filter::*;
2727
use super::static_filter::StaticFilter;
28-
use super::transform::make_bitmap_filter;
28+
use super::transform::{make_bitmap_filter, make_branchless_filter};
2929

30+
/// Maximum list size for branchless lookup on 4-byte primitives (Int32, UInt32, Float32).
31+
const BRANCHLESS_MAX_4B: usize = 32;
32+
33+
/// Maximum list size for branchless lookup on 8-byte primitives (Int64, UInt64, Float64).
34+
const BRANCHLESS_MAX_8B: usize = 16;
35+
36+
/// Maximum list size for branchless lookup on 16-byte types (Decimal128).
37+
const BRANCHLESS_MAX_16B: usize = 4;
38+
39+
/// The lookup strategy to use for a given data type and list size.
40+
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
41+
enum FilterStrategy {
42+
/// Bitmap filter for u8/u16 domains.
43+
Bitmap1B,
44+
Bitmap2B,
45+
/// Branchless OR-chain for small lists.
46+
Branchless,
47+
/// Generic ArrayStaticFilter fallback.
48+
Generic,
49+
}
50+
51+
/// Selects the lookup strategy based on data type and list size.
52+
fn select_strategy(dt: &DataType, len: usize) -> FilterStrategy {
53+
match dt.primitive_width() {
54+
Some(1) => FilterStrategy::Bitmap1B,
55+
Some(2) => FilterStrategy::Bitmap2B,
56+
Some(4) => {
57+
if len <= BRANCHLESS_MAX_4B {
58+
FilterStrategy::Branchless
59+
} else {
60+
FilterStrategy::Generic
61+
}
62+
}
63+
Some(8) => {
64+
if len <= BRANCHLESS_MAX_8B {
65+
FilterStrategy::Branchless
66+
} else {
67+
FilterStrategy::Generic
68+
}
69+
}
70+
Some(16) => {
71+
if len <= BRANCHLESS_MAX_16B {
72+
FilterStrategy::Branchless
73+
} else {
74+
FilterStrategy::Generic
75+
}
76+
}
77+
_ => FilterStrategy::Generic,
78+
}
79+
}
80+
81+
/// Creates the optimal static filter for the given array.
3082
pub(super) fn instantiate_static_filter(
3183
in_array: ArrayRef,
3284
) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
@@ -37,23 +89,46 @@ pub(super) fn instantiate_static_filter(
3789
DataType::Dictionary(_, value_type) => cast(&in_array, value_type.as_ref())?,
3890
_ => in_array,
3991
};
40-
match in_array.data_type() {
41-
DataType::Int8 | DataType::UInt8 => {
42-
make_bitmap_filter::<UInt8BitmapConfig>(&in_array)
43-
}
44-
DataType::Int16 | DataType::UInt16 => {
45-
make_bitmap_filter::<UInt16BitmapConfig>(&in_array)
46-
}
47-
DataType::Int32 => Ok(Arc::new(Int32StaticFilter::try_new(&in_array)?)),
48-
DataType::Int64 => Ok(Arc::new(Int64StaticFilter::try_new(&in_array)?)),
49-
DataType::UInt32 => Ok(Arc::new(UInt32StaticFilter::try_new(&in_array)?)),
50-
DataType::UInt64 => Ok(Arc::new(UInt64StaticFilter::try_new(&in_array)?)),
51-
// Float primitive types (use ordered wrappers for Hash/Eq)
52-
DataType::Float32 => Ok(Arc::new(Float32StaticFilter::try_new(&in_array)?)),
53-
DataType::Float64 => Ok(Arc::new(Float64StaticFilter::try_new(&in_array)?)),
54-
_ => {
55-
/* fall through to generic implementation for unsupported types (Struct, etc.) */
56-
Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?))
57-
}
92+
use FilterStrategy::*;
93+
94+
let len = in_array.len();
95+
let dt = in_array.data_type();
96+
let strategy = select_strategy(dt, len);
97+
98+
match (dt, strategy) {
99+
// Bitmap filters for 1-byte and 2-byte types
100+
(_, Bitmap1B) => make_bitmap_filter::<UInt8BitmapConfig>(&in_array),
101+
(_, Bitmap2B) => make_bitmap_filter::<UInt16BitmapConfig>(&in_array),
102+
103+
// Branchless filters for small lists of primitives
104+
(_, Branchless) => dispatch_branchless(&in_array).ok_or_else(|| {
105+
exec_datafusion_err!(
106+
"Branchless strategy selected but no filter for {:?}",
107+
dt
108+
)
109+
})?,
110+
111+
// Fallback for larger primitive lists or complex types.
112+
(_, Generic) => match dt {
113+
DataType::Int32 => Ok(Arc::new(Int32StaticFilter::try_new(&in_array)?)),
114+
DataType::Int64 => Ok(Arc::new(Int64StaticFilter::try_new(&in_array)?)),
115+
DataType::UInt32 => Ok(Arc::new(UInt32StaticFilter::try_new(&in_array)?)),
116+
DataType::UInt64 => Ok(Arc::new(UInt64StaticFilter::try_new(&in_array)?)),
117+
DataType::Float32 => Ok(Arc::new(Float32StaticFilter::try_new(&in_array)?)),
118+
DataType::Float64 => Ok(Arc::new(Float64StaticFilter::try_new(&in_array)?)),
119+
_ => Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?)),
120+
},
121+
}
122+
}
123+
124+
fn dispatch_branchless(
125+
arr: &ArrayRef,
126+
) -> Option<Result<Arc<dyn StaticFilter + Send + Sync>>> {
127+
// Dispatch to width-specific branchless filter.
128+
match arr.data_type().primitive_width() {
129+
Some(4) => Some(make_branchless_filter::<UInt32Type>(arr)),
130+
Some(8) => Some(make_branchless_filter::<UInt64Type>(arr)),
131+
Some(16) => Some(make_branchless_filter::<Decimal128Type>(arr)),
132+
_ => None,
58133
}
59134
}

0 commit comments

Comments
 (0)