vortex_array/scalar_fn/fns/binary/compare/
mod.rs1use std::cmp::Ordering;
15
16use vortex_buffer::BitBuffer;
17use vortex_buffer::BufferAllocatorRef;
18use vortex_buffer::BufferMut;
19use vortex_compute::lane_kernels::IndexedSourceExt;
20use vortex_compute::lane_kernels::LaneZip;
21use vortex_error::VortexResult;
22use vortex_error::vortex_bail;
23use vortex_error::vortex_err;
24
25use crate::ArrayRef;
26use crate::Canonical;
27use crate::ExecutionCtx;
28use crate::IntoArray;
29use crate::array::ArrayView;
30use crate::array::VTable;
31use crate::arrays::Constant;
32use crate::arrays::ConstantArray;
33use crate::arrays::ExtensionArray;
34use crate::arrays::ScalarFn;
35use crate::arrays::extension::ExtensionArrayExt;
36use crate::arrays::scalar_fn::ExactScalarFn;
37use crate::arrays::scalar_fn::ScalarFnArrayExt;
38use crate::arrays::scalar_fn::ScalarFnArrayView;
39use crate::dtype::DType;
40use crate::dtype::Nullability;
41use crate::kernel::ExecuteParentKernel;
42use crate::scalar::Scalar;
43use crate::scalar_fn::fns::binary::Binary;
44use crate::scalar_fn::fns::operators::CompareOperator;
45use crate::validity::Validity;
46
47mod boolean;
48mod bytes;
49mod decimal;
50mod nested;
51mod primitive;
52#[cfg(test)]
53mod tests;
54
55pub trait CompareKernel: VTable {
61 fn compare(
62 lhs: ArrayView<'_, Self>,
63 rhs: &ArrayRef,
64 operator: CompareOperator,
65 ctx: &mut ExecutionCtx,
66 ) -> VortexResult<Option<ArrayRef>>;
67}
68
69#[derive(Default, Debug)]
75pub struct CompareExecuteAdaptor<V>(pub V);
76
77impl<V> ExecuteParentKernel<V> for CompareExecuteAdaptor<V>
78where
79 V: CompareKernel,
80{
81 type Parent = ExactScalarFn<Binary>;
82
83 fn execute_parent(
84 &self,
85 array: ArrayView<'_, V>,
86 parent: ScalarFnArrayView<'_, Binary>,
87 child_idx: usize,
88 ctx: &mut ExecutionCtx,
89 ) -> VortexResult<Option<ArrayRef>> {
90 let Some(cmp_op) = parent.options.to_compare_operator() else {
93 return Ok(None);
94 };
95
96 let Some(scalar_fn_array) = parent.as_opt::<ScalarFn>() else {
98 return Ok(None);
99 };
100 let (cmp_op, other) = match child_idx {
103 0 => (cmp_op, scalar_fn_array.get_child(1)),
104 1 => (cmp_op.swap(), scalar_fn_array.get_child(0)),
105 _ => return Ok(None),
106 };
107
108 let len = array.len();
109 let nullable = array.dtype().is_nullable() || other.dtype().is_nullable();
110
111 if len == 0 {
113 return Ok(Some(
114 Canonical::empty(&DType::Bool(nullable.into())).into_array(),
115 ));
116 }
117
118 if other.as_constant().is_some_and(|s| s.is_null()) {
120 return Ok(Some(
121 ConstantArray::new(Scalar::null(DType::Bool(nullable.into())), len).into_array(),
122 ));
123 }
124
125 V::compare(array, other, cmp_op, ctx)
126 }
127}
128
129pub(crate) fn execute_compare(
135 lhs: &ArrayRef,
136 rhs: &ArrayRef,
137 op: CompareOperator,
138 ctx: &mut ExecutionCtx,
139) -> VortexResult<ArrayRef> {
140 let nullable = lhs.dtype().is_nullable() || rhs.dtype().is_nullable();
141
142 if lhs.len() != rhs.len() {
143 vortex_bail!(
144 "compare operator requires equal lengths, got {} and {}",
145 lhs.len(),
146 rhs.len()
147 );
148 }
149
150 if lhs.is_empty() {
151 return Ok(Canonical::empty(&DType::Bool(nullable.into())).into_array());
152 }
153
154 let left_constant_null = lhs.as_constant().map(|l| l.is_null()).unwrap_or(false);
155 let right_constant_null = rhs.as_constant().map(|r| r.is_null()).unwrap_or(false);
156 if left_constant_null || right_constant_null {
157 return Ok(
158 ConstantArray::new(Scalar::null(DType::Bool(nullable.into())), lhs.len()).into_array(),
159 );
160 }
161
162 if let (Some(lhs_const), Some(rhs_const)) = (lhs.as_opt::<Constant>(), rhs.as_opt::<Constant>())
164 {
165 let result = scalar_cmp(lhs_const.scalar(), rhs_const.scalar(), op)?;
166 return Ok(ConstantArray::new(result, lhs.len()).into_array());
167 }
168
169 compare_arrays(lhs, rhs, op, ctx)
170}
171
172fn compare_arrays(
174 lhs: &ArrayRef,
175 rhs: &ArrayRef,
176 op: CompareOperator,
177 ctx: &mut ExecutionCtx,
178) -> VortexResult<ArrayRef> {
179 let nullability = Nullability::from(lhs.dtype().is_nullable() || rhs.dtype().is_nullable());
180
181 if lhs.dtype().is_extension() || rhs.dtype().is_extension() {
185 if lhs.dtype().is_extension()
186 && rhs.dtype().is_extension()
187 && !lhs.dtype().eq_ignore_nullability(rhs.dtype())
188 {
189 vortex_bail!(
190 "Cannot compare extension dtypes {} and {}",
191 lhs.dtype(),
192 rhs.dtype()
193 );
194 }
195 let lhs = extension_storage(lhs, ctx)?;
196 let rhs = extension_storage(rhs, ctx)?;
197 return compare_arrays(&lhs, &rhs, op, ctx);
198 }
199
200 if !lhs.dtype().eq_ignore_nullability(rhs.dtype()) {
201 vortex_bail!(
202 "Cannot compare different DTypes {} and {}",
203 lhs.dtype(),
204 rhs.dtype()
205 );
206 }
207
208 match lhs.dtype() {
209 DType::Null => Ok(ConstantArray::new(
211 Scalar::null(DType::Bool(Nullability::Nullable)),
212 lhs.len(),
213 )
214 .into_array()),
215 DType::Bool(_) => boolean::compare_bool(lhs, rhs, op, nullability, ctx),
216 DType::Primitive(..) => primitive::compare_primitive(lhs, rhs, op, nullability, ctx),
217 DType::Decimal(..) => decimal::compare_decimal(lhs, rhs, op, nullability, ctx),
218 DType::Utf8(_) | DType::Binary(_) => bytes::compare_bytes(lhs, rhs, op, nullability, ctx),
219 DType::Struct(..) | DType::List(..) | DType::FixedSizeList(..) | DType::Map(..) => {
220 nested::compare_nested(lhs, rhs, op, nullability, ctx)
221 }
222 DType::Union(..) | DType::Variant(_) | DType::Extension(_) => {
223 vortex_bail!("compare is not supported for dtype {}", lhs.dtype())
224 }
225 }
226}
227
228fn extension_storage(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
231 if !array.dtype().is_extension() {
232 return Ok(array.clone());
233 }
234 if let Some(constant) = array.as_opt::<Constant>() {
235 return Ok(ConstantArray::new(
236 constant.scalar().as_extension().to_storage_scalar(),
237 constant.len(),
238 )
239 .into_array());
240 }
241 let ext = array.clone().execute::<ExtensionArray>(ctx)?;
242 Ok(ext.storage_array().clone())
243}
244
245pub fn scalar_cmp(lhs: &Scalar, rhs: &Scalar, operator: CompareOperator) -> VortexResult<Scalar> {
246 if lhs.is_null() | rhs.is_null() {
247 return Ok(Scalar::null(DType::Bool(Nullability::Nullable)));
248 }
249
250 let nullability = lhs.dtype().nullability() | rhs.dtype().nullability();
251
252 let ordering = lhs.partial_cmp(rhs).ok_or_else(|| {
254 vortex_err!(
255 "Cannot compare scalars with incompatible types: {} and {}",
256 lhs.dtype(),
257 rhs.dtype()
258 )
259 })?;
260
261 let b = match operator {
262 CompareOperator::Eq => ordering.is_eq(),
263 CompareOperator::NotEq => ordering.is_ne(),
264 CompareOperator::Gt => ordering.is_gt(),
265 CompareOperator::Gte => ordering.is_ge(),
266 CompareOperator::Lt => ordering.is_lt(),
267 CompareOperator::Lte => ordering.is_le(),
268 };
269
270 Ok(Scalar::bool(b, nullability))
271}
272
273#[inline]
275pub(super) fn ordering_predicate(op: CompareOperator) -> fn(Ordering) -> bool {
276 match op {
277 CompareOperator::Eq => Ordering::is_eq,
278 CompareOperator::NotEq => Ordering::is_ne,
279 CompareOperator::Gt => Ordering::is_gt,
280 CompareOperator::Gte => Ordering::is_ge,
281 CompareOperator::Lt => Ordering::is_lt,
282 CompareOperator::Lte => Ordering::is_le,
283 }
284}
285
286pub(super) fn bit_buffer_from_words(words: BufferMut<u64>, len: usize) -> BitBuffer {
288 debug_assert!(words.len() * 64 >= len);
289 let mut bytes = words.into_byte_buffer();
290 bytes.truncate(len.div_ceil(8));
291 BitBuffer::new(bytes.freeze(), len)
292}
293
294pub(super) fn collect_zip_bits<T: Copy>(
296 lhs: &[T],
297 rhs: &[T],
298 f: impl Fn(T, T) -> bool,
299 allocator: &BufferAllocatorRef,
300) -> BitBuffer {
301 let len = lhs.len();
302 let mut words = BufferMut::<u64>::zeroed_in(len.div_ceil(64), allocator.clone());
303 LaneZip::new(lhs, rhs).map_bits_into(words.as_mut_slice(), |(a, b)| f(a, b));
304 bit_buffer_from_words(words, len)
305}
306
307pub(super) fn collect_bits<T: Copy>(
309 values: &[T],
310 f: impl Fn(T) -> bool,
311 allocator: &BufferAllocatorRef,
312) -> BitBuffer {
313 let len = values.len();
314 let mut words = BufferMut::<u64>::zeroed_in(len.div_ceil(64), allocator.clone());
315 values.map_bits_into(words.as_mut_slice(), f);
316 bit_buffer_from_words(words, len)
317}
318
319pub(super) fn compare_validity(
322 lhs: Validity,
323 rhs: Validity,
324 nullability: Nullability,
325) -> VortexResult<Validity> {
326 Ok(lhs.and(rhs)?.union_nullability(nullability))
327}