1use std::fmt;
14use std::fmt::Formatter;
15use std::hash::Hash;
16use std::sync::Arc;
17
18use prost::Message;
19use vortex_error::VortexResult;
20use vortex_error::vortex_bail;
21use vortex_mask::AllOr;
22use vortex_mask::Mask;
23use vortex_proto::expr as pb;
24use vortex_session::VortexSession;
25use vortex_session::registry::CachedId;
26
27use crate::ArrayRef;
28use crate::ExecutionCtx;
29use crate::IntoArray;
30use crate::arrays::BoolArray;
31use crate::arrays::ConstantArray;
32use crate::arrays::bool::BoolArrayExt;
33use crate::builders::ArrayBuilder;
34use crate::builders::builder_with_capacity;
35use crate::builtins::ArrayBuiltins;
36use crate::dtype::DType;
37use crate::expr::Expression;
38use crate::expr::display::ExprDisplay;
39use crate::scalar::Scalar;
40use crate::scalar_fn::Arity;
41use crate::scalar_fn::ChildName;
42use crate::scalar_fn::ExecutionArgs;
43use crate::scalar_fn::ScalarFnId;
44use crate::scalar_fn::ScalarFnVTable;
45use crate::scalar_fn::SimplifyCtx;
46use crate::scalar_fn::fns::is_not_null::IsNotNull;
47use crate::scalar_fn::fns::is_null::IsNull;
48use crate::scalar_fn::fns::literal::Literal;
49use crate::scalar_fn::fns::zip::zip_impl;
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
53pub struct CaseWhenOptions {
54 pub num_when_then_pairs: u32,
56 pub has_else: bool,
59}
60
61impl CaseWhenOptions {
62 pub fn num_children(&self) -> usize {
64 self.num_when_then_pairs as usize * 2 + usize::from(self.has_else)
65 }
66}
67
68impl fmt::Display for CaseWhenOptions {
69 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
70 write!(
71 f,
72 "case_when(pairs={}, else={})",
73 self.num_when_then_pairs, self.has_else
74 )
75 }
76}
77
78#[derive(Clone)]
82pub struct CaseWhen;
83
84impl ScalarFnVTable for CaseWhen {
85 type Options = CaseWhenOptions;
86
87 fn id(&self) -> ScalarFnId {
88 static ID: CachedId = CachedId::new("vortex.case_when");
89 *ID
90 }
91
92 fn serialize(&self, _options: &Self::Options) -> VortexResult<Option<Vec<u8>>> {
93 vortex_bail!("cannot serialize")
97 }
98
99 fn deserialize(
100 &self,
101 metadata: &[u8],
102 _session: &VortexSession,
103 ) -> VortexResult<Self::Options> {
104 let opts = pb::CaseWhenOpts::decode(metadata)?;
105 if opts.num_children < 2 {
106 vortex_bail!(
107 "CaseWhen expects at least 2 children, got {}",
108 opts.num_children
109 );
110 }
111 Ok(CaseWhenOptions {
112 num_when_then_pairs: opts.num_children / 2,
113 has_else: opts.num_children % 2 == 1,
114 })
115 }
116
117 fn arity(&self, options: &Self::Options) -> Arity {
118 Arity::Exact(options.num_children())
119 }
120
121 fn child_name(&self, options: &Self::Options, child_idx: usize) -> ChildName {
122 let num_pair_children = options.num_when_then_pairs as usize * 2;
123 if child_idx < num_pair_children {
124 let pair_idx = child_idx / 2;
125 if child_idx.is_multiple_of(2) {
126 ChildName::from(Arc::from(format!("when_{pair_idx}")))
127 } else {
128 ChildName::from(Arc::from(format!("then_{pair_idx}")))
129 }
130 } else if options.has_else && child_idx == num_pair_children {
131 ChildName::from("else")
132 } else {
133 unreachable!("Invalid child index {} for CaseWhen", child_idx)
134 }
135 }
136
137 fn fmt_sql(
138 &self,
139 options: &Self::Options,
140 expr: &dyn ExprDisplay,
141 f: &mut Formatter<'_>,
142 ) -> fmt::Result {
143 write!(f, "CASE")?;
144 for i in 0..options.num_when_then_pairs as usize {
145 write!(
146 f,
147 " WHEN {} THEN {}",
148 expr.display_child(i * 2),
149 expr.display_child(i * 2 + 1)
150 )?;
151 }
152 if options.has_else {
153 let else_idx = options.num_when_then_pairs as usize * 2;
154 write!(f, " ELSE {}", expr.display_child(else_idx))?;
155 }
156 write!(f, " END")
157 }
158
159 fn return_dtype(&self, options: &Self::Options, arg_dtypes: &[DType]) -> VortexResult<DType> {
160 if options.num_when_then_pairs == 0 {
161 vortex_bail!("CaseWhen must have at least one WHEN/THEN pair");
162 }
163
164 let expected_len = options.num_children();
165 if arg_dtypes.len() != expected_len {
166 vortex_bail!(
167 "CaseWhen expects {expected_len} argument dtypes, got {}",
168 arg_dtypes.len()
169 );
170 }
171
172 let first_then = &arg_dtypes[1];
176 let mut result_dtype = first_then.clone();
177
178 for i in 1..options.num_when_then_pairs as usize {
179 let then_i = &arg_dtypes[i * 2 + 1];
180 if !first_then.eq_ignore_nullability(then_i) {
181 vortex_bail!(
182 "CaseWhen THEN dtypes must match (ignoring nullability), got {} and {}",
183 first_then,
184 then_i
185 );
186 }
187 result_dtype = result_dtype.union_nullability(then_i.nullability());
188 }
189
190 if options.has_else {
191 let else_dtype = &arg_dtypes[options.num_when_then_pairs as usize * 2];
192 if !result_dtype.eq_ignore_nullability(else_dtype) {
193 vortex_bail!(
194 "CaseWhen THEN and ELSE dtypes must match (ignoring nullability), got {} and {}",
195 first_then,
196 else_dtype
197 );
198 }
199 result_dtype = result_dtype.union_nullability(else_dtype.nullability());
200 } else {
201 result_dtype = result_dtype.as_nullable();
203 }
204
205 Ok(result_dtype)
206 }
207
208 fn execute(
209 &self,
210 options: &Self::Options,
211 args: &dyn ExecutionArgs,
212 ctx: &mut ExecutionCtx,
213 ) -> VortexResult<ArrayRef> {
214 let row_count = args.row_count();
221 let num_pairs = options.num_when_then_pairs as usize;
222
223 let mut remaining = Mask::new_true(row_count);
224 let mut branches: Vec<(Mask, ArrayRef)> = Vec::with_capacity(num_pairs);
225
226 for i in 0..num_pairs {
227 if remaining.all_false() {
228 break;
229 }
230
231 let condition = args.get(i * 2)?;
232 let cond_bool = condition.execute::<BoolArray>(ctx)?;
233 let cond_mask = cond_bool.to_mask_fill_null_false(ctx);
234 let effective_mask = &remaining & &cond_mask;
235
236 if effective_mask.all_false() {
237 continue;
238 }
239
240 let then_value = args.get(i * 2 + 1)?;
241 remaining = remaining.bitand_not(&cond_mask);
242 branches.push((effective_mask, then_value));
243 }
244
245 let else_value: ArrayRef = if options.has_else {
246 args.get(num_pairs * 2)?
247 } else {
248 let then_dtype = args.get(1)?.dtype().as_nullable();
249 ConstantArray::new(Scalar::null(then_dtype), row_count).into_array()
250 };
251
252 if branches.is_empty() {
253 return Ok(else_value);
254 }
255
256 merge_case_branches(branches, else_value, ctx)
257 }
258
259 fn simplify(
260 &self,
261 options: &Self::Options,
262 expr: &Expression,
263 _ctx: &dyn SimplifyCtx,
264 ) -> VortexResult<Option<Expression>> {
265 if options.num_when_then_pairs != 1 || !options.has_else {
275 return Ok(None);
276 }
277
278 let when = expr.child(0);
279 let then = expr.child(1);
280 let els = expr.child(2);
281
282 let (x, fill) = if when.is::<IsNull>() && when.child(0) == els {
284 (els, then)
285 } else if when.is::<IsNotNull>() && when.child(0) == then {
287 (then, els)
288 } else {
289 return Ok(None);
290 };
291
292 let Some(scalar) = fill.as_opt::<Literal>() else {
293 return Ok(None);
294 };
295
296 if scalar.is_null() {
297 return Ok(Some(x.clone()));
299 }
300
301 Ok(Some(crate::expr::fill_null(x.clone(), fill.clone())))
302 }
303
304 fn is_strict(&self, _options: &Self::Options) -> bool {
305 false
307 }
308
309 fn is_fallible(&self, _options: &Self::Options) -> bool {
310 false
311 }
312}
313
314const SLICE_CROSSOVER_RUN_LEN: usize = 4;
317
318fn merge_case_branches(
322 branches: Vec<(Mask, ArrayRef)>,
323 else_value: ArrayRef,
324 ctx: &mut ExecutionCtx,
325) -> VortexResult<ArrayRef> {
326 if branches.len() == 1 {
327 let (mask, then_value) = &branches[0];
328 return zip_impl(then_value, &else_value, mask, ctx);
329 }
330
331 let output_nullability = branches
332 .iter()
333 .fold(else_value.dtype().nullability(), |acc, (_, arr)| {
334 acc | arr.dtype().nullability()
335 });
336 let output_dtype = else_value.dtype().with_nullability(output_nullability);
337 let branch_arrays: Vec<&ArrayRef> = branches.iter().map(|(_, arr)| arr).collect();
338
339 let mut spans: Vec<(usize, usize, usize)> = Vec::new();
340 for (branch_idx, (mask, _)) in branches.iter().enumerate() {
341 match mask.slices() {
342 AllOr::All => return branch_arrays[branch_idx].cast(output_dtype),
343 AllOr::None => {}
344 AllOr::Some(slices) => {
345 for &(start, end) in slices {
346 spans.push((start, end, branch_idx));
347 }
348 }
349 }
350 }
351 spans.sort_unstable_by_key(|&(start, ..)| start);
352
353 if spans.is_empty() {
354 return else_value.cast(output_dtype);
355 }
356
357 let builder = builder_with_capacity(&output_dtype, else_value.len());
358
359 let fragmented = spans.len() > else_value.len() / SLICE_CROSSOVER_RUN_LEN;
360 if fragmented {
361 merge_row_by_row(
362 &branch_arrays,
363 &else_value,
364 &spans,
365 &output_dtype,
366 builder,
367 ctx,
368 )
369 } else {
370 merge_run_by_run(
371 &branch_arrays,
372 &else_value,
373 &spans,
374 &output_dtype,
375 builder,
376 ctx,
377 )
378 }
379}
380
381fn merge_row_by_row(
384 branch_arrays: &[&ArrayRef],
385 else_value: &ArrayRef,
386 spans: &[(usize, usize, usize)],
387 output_dtype: &DType,
388 mut builder: Box<dyn ArrayBuilder>,
389 ctx: &mut ExecutionCtx,
390) -> VortexResult<ArrayRef> {
391 let mut pos = 0;
392 for &(start, end, branch_idx) in spans {
393 for row in pos..start {
394 let scalar = else_value.execute_scalar(row, ctx)?;
395 builder.append_scalar(&scalar.cast(output_dtype)?)?;
396 }
397 for row in start..end {
398 let scalar = branch_arrays[branch_idx].execute_scalar(row, ctx)?;
399 builder.append_scalar(&scalar.cast(output_dtype)?)?;
400 }
401 pos = end;
402 }
403 for row in pos..else_value.len() {
404 let scalar = else_value.execute_scalar(row, ctx)?;
405 builder.append_scalar(&scalar.cast(output_dtype)?)?;
406 }
407
408 Ok(builder.finish())
409}
410
411fn merge_run_by_run(
415 branch_arrays: &[&ArrayRef],
416 else_value: &ArrayRef,
417 spans: &[(usize, usize, usize)],
418 output_dtype: &DType,
419 mut builder: Box<dyn ArrayBuilder>,
420 ctx: &mut ExecutionCtx,
421) -> VortexResult<ArrayRef> {
422 let else_value = else_value.cast(output_dtype.clone())?;
423 let len = else_value.len();
424 for (start, end, branch_idx) in spans {
425 if builder.len() < *start {
426 else_value
427 .slice(builder.len()..*start)?
428 .append_to_builder(builder.as_mut(), ctx)?;
429 }
430 branch_arrays[*branch_idx]
431 .cast(output_dtype.clone())?
432 .slice(*start..*end)?
433 .append_to_builder(builder.as_mut(), ctx)?;
434 }
435 if builder.len() < len {
436 else_value
437 .slice(builder.len()..len)?
438 .append_to_builder(builder.as_mut(), ctx)?;
439 }
440
441 Ok(builder.finish())
442}
443
444#[cfg(test)]
445mod tests {
446 use std::sync::LazyLock;
447
448 use vortex_buffer::buffer;
449 use vortex_error::VortexExpect as _;
450 use vortex_session::VortexSession;
451
452 use super::*;
453 use crate::Canonical;
454 use crate::IntoArray;
455 use crate::VortexSessionExecute;
456 use crate::arrays::BoolArray;
457 use crate::arrays::PrimitiveArray;
458 use crate::arrays::StructArray;
459 use crate::assert_arrays_eq;
460 use crate::dtype::DType;
461 use crate::dtype::Nullability;
462 use crate::dtype::PType;
463 use crate::dtype::StructFields;
464 use crate::expr::case_when;
465 use crate::expr::case_when_no_else;
466 use crate::expr::col;
467 use crate::expr::eq;
468 use crate::expr::get_item;
469 use crate::expr::gt;
470 use crate::expr::is_not_null;
471 use crate::expr::is_null;
472 use crate::expr::lit;
473 use crate::expr::nested_case_when;
474 use crate::expr::root;
475 use crate::expr::test_harness;
476 use crate::scalar::Scalar;
477
478 static SESSION: LazyLock<VortexSession> = LazyLock::new(crate::array_session);
479
480 fn evaluate_expr(expr: &Expression, array: &ArrayRef) -> ArrayRef {
482 let mut ctx = SESSION.create_execution_ctx();
483 array
484 .clone()
485 .apply(expr)
486 .unwrap()
487 .execute::<Canonical>(&mut ctx)
488 .unwrap()
489 .into_array()
490 }
491
492 #[test]
495 #[should_panic(expected = "cannot serialize")]
496 fn test_serialization_roundtrip() {
497 let options = CaseWhenOptions {
498 num_when_then_pairs: 1,
499 has_else: true,
500 };
501 let serialized = CaseWhen.serialize(&options).unwrap().unwrap();
502 let deserialized = CaseWhen
503 .deserialize(&serialized, &VortexSession::empty())
504 .unwrap();
505 assert_eq!(options, deserialized);
506 }
507
508 #[test]
509 #[should_panic(expected = "cannot serialize")]
510 fn test_serialization_no_else() {
511 let options = CaseWhenOptions {
512 num_when_then_pairs: 1,
513 has_else: false,
514 };
515 let serialized = CaseWhen.serialize(&options).unwrap().unwrap();
516 let deserialized = CaseWhen
517 .deserialize(&serialized, &VortexSession::empty())
518 .unwrap();
519 assert_eq!(options, deserialized);
520 }
521
522 #[test]
525 fn test_display_with_else() {
526 let expr = case_when(gt(col("value"), lit(0i32)), lit(100i32), lit(0i32));
527 let display = format!("{}", expr);
528 assert!(display.contains("CASE"));
529 assert!(display.contains("WHEN"));
530 assert!(display.contains("THEN"));
531 assert!(display.contains("ELSE"));
532 assert!(display.contains("END"));
533 }
534
535 #[test]
536 fn test_display_no_else() {
537 let expr = case_when_no_else(gt(col("value"), lit(0i32)), lit(100i32));
538 let display = format!("{}", expr);
539 assert!(display.contains("CASE"));
540 assert!(display.contains("WHEN"));
541 assert!(display.contains("THEN"));
542 assert!(!display.contains("ELSE"));
543 assert!(display.contains("END"));
544 }
545
546 #[test]
547 fn test_display_nested_nary() {
548 let expr = nested_case_when(
550 vec![
551 (gt(col("x"), lit(10i32)), lit("high")),
552 (gt(col("x"), lit(5i32)), lit("medium")),
553 ],
554 Some(lit("low")),
555 );
556 let display = format!("{}", expr);
557 assert_eq!(display.matches("CASE").count(), 1);
558 assert_eq!(display.matches("WHEN").count(), 2);
559 assert_eq!(display.matches("THEN").count(), 2);
560 }
561
562 #[test]
565 fn test_return_dtype_with_else() {
566 let expr = case_when(lit(true), lit(100i32), lit(0i32));
567 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
568 let result_dtype = expr.return_dtype(&input_dtype).unwrap();
569 assert_eq!(
570 result_dtype,
571 DType::Primitive(PType::I32, Nullability::NonNullable)
572 );
573 }
574
575 #[test]
576 fn test_return_dtype_with_nullable_else() {
577 let expr = case_when(
578 lit(true),
579 lit(100i32),
580 lit(Scalar::null(DType::Primitive(
581 PType::I32,
582 Nullability::Nullable,
583 ))),
584 );
585 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
586 let result_dtype = expr.return_dtype(&input_dtype).unwrap();
587 assert_eq!(
588 result_dtype,
589 DType::Primitive(PType::I32, Nullability::Nullable)
590 );
591 }
592
593 #[test]
594 fn test_return_dtype_without_else_is_nullable() {
595 let expr = case_when_no_else(lit(true), lit(100i32));
596 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
597 let result_dtype = expr.return_dtype(&input_dtype).unwrap();
598 assert_eq!(
599 result_dtype,
600 DType::Primitive(PType::I32, Nullability::Nullable)
601 );
602 }
603
604 #[test]
605 fn test_return_dtype_with_struct_input() {
606 let dtype = test_harness::struct_dtype();
607 let expr = case_when(
608 gt(get_item("col1", root()), lit(10u16)),
609 lit(100i32),
610 lit(0i32),
611 );
612 let result_dtype = expr.return_dtype(&dtype).unwrap();
613 assert_eq!(
614 result_dtype,
615 DType::Primitive(PType::I32, Nullability::NonNullable)
616 );
617 }
618
619 #[test]
620 fn test_return_dtype_mismatched_then_else_errors() {
621 let expr = case_when(lit(true), lit(100i32), lit("zero"));
622 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
623 let err = expr.return_dtype(&input_dtype).unwrap_err();
624 assert!(
625 err.to_string()
626 .contains("THEN and ELSE dtypes must match (ignoring nullability)")
627 );
628 }
629
630 #[test]
633 fn test_arity_with_else() {
634 let options = CaseWhenOptions {
635 num_when_then_pairs: 1,
636 has_else: true,
637 };
638 assert_eq!(CaseWhen.arity(&options), Arity::Exact(3));
639 }
640
641 #[test]
642 fn test_arity_without_else() {
643 let options = CaseWhenOptions {
644 num_when_then_pairs: 1,
645 has_else: false,
646 };
647 assert_eq!(CaseWhen.arity(&options), Arity::Exact(2));
648 }
649
650 #[test]
653 fn test_child_names() {
654 let options = CaseWhenOptions {
655 num_when_then_pairs: 1,
656 has_else: true,
657 };
658 assert_eq!(CaseWhen.child_name(&options, 0).to_string(), "when_0");
659 assert_eq!(CaseWhen.child_name(&options, 1).to_string(), "then_0");
660 assert_eq!(CaseWhen.child_name(&options, 2).to_string(), "else");
661 }
662
663 #[test]
666 #[should_panic(expected = "cannot serialize")]
667 fn test_serialization_roundtrip_nary() {
668 let options = CaseWhenOptions {
669 num_when_then_pairs: 3,
670 has_else: true,
671 };
672 let serialized = CaseWhen.serialize(&options).unwrap().unwrap();
673 let deserialized = CaseWhen
674 .deserialize(&serialized, &VortexSession::empty())
675 .unwrap();
676 assert_eq!(options, deserialized);
677 }
678
679 #[test]
680 #[should_panic(expected = "cannot serialize")]
681 fn test_serialization_roundtrip_nary_no_else() {
682 let options = CaseWhenOptions {
683 num_when_then_pairs: 4,
684 has_else: false,
685 };
686 let serialized = CaseWhen.serialize(&options).unwrap().unwrap();
687 let deserialized = CaseWhen
688 .deserialize(&serialized, &VortexSession::empty())
689 .unwrap();
690 assert_eq!(options, deserialized);
691 }
692
693 #[test]
696 fn test_arity_nary_with_else() {
697 let options = CaseWhenOptions {
698 num_when_then_pairs: 3,
699 has_else: true,
700 };
701 assert_eq!(CaseWhen.arity(&options), Arity::Exact(7));
703 }
704
705 #[test]
706 fn test_arity_nary_without_else() {
707 let options = CaseWhenOptions {
708 num_when_then_pairs: 3,
709 has_else: false,
710 };
711 assert_eq!(CaseWhen.arity(&options), Arity::Exact(6));
713 }
714
715 #[test]
718 fn test_child_names_nary() {
719 let options = CaseWhenOptions {
720 num_when_then_pairs: 3,
721 has_else: true,
722 };
723 assert_eq!(CaseWhen.child_name(&options, 0).to_string(), "when_0");
724 assert_eq!(CaseWhen.child_name(&options, 1).to_string(), "then_0");
725 assert_eq!(CaseWhen.child_name(&options, 2).to_string(), "when_1");
726 assert_eq!(CaseWhen.child_name(&options, 3).to_string(), "then_1");
727 assert_eq!(CaseWhen.child_name(&options, 4).to_string(), "when_2");
728 assert_eq!(CaseWhen.child_name(&options, 5).to_string(), "then_2");
729 assert_eq!(CaseWhen.child_name(&options, 6).to_string(), "else");
730 }
731
732 #[test]
735 fn test_return_dtype_nary_mismatched_then_types_errors() {
736 let expr = nested_case_when(
737 vec![(lit(true), lit(100i32)), (lit(false), lit("oops"))],
738 Some(lit(0i32)),
739 );
740 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
741 let err = expr.return_dtype(&input_dtype).unwrap_err();
742 assert!(err.to_string().contains("THEN dtypes must match"));
743 }
744
745 #[test]
746 fn test_return_dtype_nary_mixed_nullability() {
747 let non_null_then = lit(100i32);
750 let nullable_then = lit(Scalar::null(DType::Primitive(
751 PType::I32,
752 Nullability::Nullable,
753 )));
754 let expr = nested_case_when(
755 vec![(lit(true), non_null_then), (lit(false), nullable_then)],
756 Some(lit(0i32)),
757 );
758 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
759 let result = expr.return_dtype(&input_dtype).unwrap();
760 assert_eq!(result, DType::Primitive(PType::I32, Nullability::Nullable));
761 }
762
763 #[test]
764 fn test_return_dtype_nary_no_else_is_nullable() {
765 let expr = nested_case_when(
766 vec![(lit(true), lit(10i32)), (lit(false), lit(20i32))],
767 None,
768 );
769 let input_dtype = DType::Primitive(PType::I32, Nullability::NonNullable);
770 let result = expr.return_dtype(&input_dtype).unwrap();
771 assert_eq!(result, DType::Primitive(PType::I32, Nullability::Nullable));
772 }
773
774 #[test]
777 fn test_replace_children() {
778 let expr = case_when(lit(true), lit(1i32), lit(0i32));
779 expr.with_children([lit(false), lit(2i32), lit(3i32)])
780 .vortex_expect("operation should succeed in test");
781 }
782
783 #[test]
786 fn test_evaluate_simple_condition() {
787 let mut ctx = SESSION.create_execution_ctx();
788 let test_array =
789 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
790 .unwrap()
791 .into_array();
792
793 let expr = case_when(
794 gt(get_item("value", root()), lit(2i32)),
795 lit(100i32),
796 lit(0i32),
797 );
798
799 let result = evaluate_expr(&expr, &test_array);
800 assert_arrays_eq!(
801 result,
802 buffer![0i32, 0, 100, 100, 100].into_array(),
803 &mut ctx
804 );
805 }
806
807 #[test]
808 fn test_evaluate_nary_multiple_conditions() {
809 let mut ctx = SESSION.create_execution_ctx();
810 let test_array =
812 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
813 .unwrap()
814 .into_array();
815
816 let expr = nested_case_when(
817 vec![
818 (eq(get_item("value", root()), lit(1i32)), lit(10i32)),
819 (eq(get_item("value", root()), lit(3i32)), lit(30i32)),
820 ],
821 Some(lit(0i32)),
822 );
823
824 let result = evaluate_expr(&expr, &test_array);
825 assert_arrays_eq!(result, buffer![10i32, 0, 30, 0, 0].into_array(), &mut ctx);
826 }
827
828 #[test]
829 fn test_evaluate_nary_first_match_wins() {
830 let mut ctx = SESSION.create_execution_ctx();
831 let test_array =
832 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
833 .unwrap()
834 .into_array();
835
836 let expr = nested_case_when(
838 vec![
839 (gt(get_item("value", root()), lit(2i32)), lit(100i32)),
840 (gt(get_item("value", root()), lit(3i32)), lit(200i32)),
841 ],
842 Some(lit(0i32)),
843 );
844
845 let result = evaluate_expr(&expr, &test_array);
846 assert_arrays_eq!(
847 result,
848 buffer![0i32, 0, 100, 100, 100].into_array(),
849 &mut ctx
850 );
851 }
852
853 #[test]
854 fn test_evaluate_no_else_returns_null() {
855 let mut ctx = SESSION.create_execution_ctx();
856 let test_array =
857 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
858 .unwrap()
859 .into_array();
860
861 let expr = case_when_no_else(gt(get_item("value", root()), lit(3i32)), lit(100i32));
862
863 let result = evaluate_expr(&expr, &test_array);
864 assert!(result.dtype().is_nullable());
865 assert_arrays_eq!(
866 result,
867 PrimitiveArray::from_option_iter([None::<i32>, None, None, Some(100), Some(100)])
868 .into_array(),
869 &mut ctx
870 );
871 }
872
873 #[test]
874 fn test_evaluate_all_conditions_false() {
875 let mut ctx = SESSION.create_execution_ctx();
876 let test_array =
877 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
878 .unwrap()
879 .into_array();
880
881 let expr = case_when(
882 gt(get_item("value", root()), lit(100i32)),
883 lit(1i32),
884 lit(0i32),
885 );
886
887 let result = evaluate_expr(&expr, &test_array);
888 assert_arrays_eq!(result, buffer![0i32, 0, 0, 0, 0].into_array(), &mut ctx);
889 }
890
891 #[test]
892 fn test_evaluate_all_conditions_true() {
893 let mut ctx = SESSION.create_execution_ctx();
894 let test_array =
895 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
896 .unwrap()
897 .into_array();
898
899 let expr = case_when(
900 gt(get_item("value", root()), lit(0i32)),
901 lit(100i32),
902 lit(0i32),
903 );
904
905 let result = evaluate_expr(&expr, &test_array);
906 assert_arrays_eq!(
907 result,
908 buffer![100i32, 100, 100, 100, 100].into_array(),
909 &mut ctx
910 );
911 }
912
913 #[test]
914 fn test_evaluate_all_true_no_else_returns_correct_dtype() {
915 let mut ctx = SESSION.create_execution_ctx();
916 let test_array = StructArray::from_fields(&[("value", buffer![1i32, 2, 3].into_array())])
919 .unwrap()
920 .into_array();
921
922 let expr = case_when_no_else(gt(get_item("value", root()), lit(0i32)), lit(100i32));
923
924 let result = evaluate_expr(&expr, &test_array);
925 assert!(
926 result.dtype().is_nullable(),
927 "result dtype must be Nullable, got {:?}",
928 result.dtype()
929 );
930 assert_arrays_eq!(
931 result,
932 PrimitiveArray::from_option_iter([Some(100i32), Some(100), Some(100)]).into_array(),
933 &mut ctx
934 );
935 }
936
937 #[test]
938 fn test_merge_case_branches_widens_nullability_of_later_branch() -> VortexResult<()> {
939 let mut ctx = SESSION.create_execution_ctx();
940 let test_array =
948 StructArray::from_fields(&[("value", buffer![0i32, 1, 2].into_array())])?.into_array();
949
950 let nullable_20 =
951 Scalar::from(20i32).cast(&DType::Primitive(PType::I32, Nullability::Nullable))?;
952
953 let expr = nested_case_when(
954 vec![
955 (eq(get_item("value", root()), lit(0i32)), lit(10i32)),
956 (eq(get_item("value", root()), lit(1i32)), lit(nullable_20)),
957 ],
958 Some(lit(0i32)),
959 );
960
961 let result = evaluate_expr(&expr, &test_array);
962 assert!(
963 result.dtype().is_nullable(),
964 "result dtype must be Nullable, got {:?}",
965 result.dtype()
966 );
967 assert_arrays_eq!(
968 result,
969 PrimitiveArray::from_option_iter([Some(10), Some(20), Some(0)]).into_array(),
970 &mut ctx
971 );
972 Ok(())
973 }
974
975 #[test]
976 fn test_evaluate_with_literal_condition() {
977 let mut ctx = SESSION.create_execution_ctx();
978 let test_array = buffer![1i32, 2, 3].into_array();
979 let expr = case_when(lit(true), lit(100i32), lit(0i32));
980 let result = evaluate_expr(&expr, &test_array);
981
982 assert_arrays_eq!(result, buffer![100i32, 100, 100].into_array(), &mut ctx);
983 }
984
985 #[test]
986 fn test_evaluate_with_bool_column_result() {
987 let mut ctx = SESSION.create_execution_ctx();
988 let test_array =
989 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
990 .unwrap()
991 .into_array();
992
993 let expr = case_when(
994 gt(get_item("value", root()), lit(2i32)),
995 lit(true),
996 lit(false),
997 );
998
999 let result = evaluate_expr(&expr, &test_array);
1000 assert_arrays_eq!(
1001 result,
1002 BoolArray::from_iter([false, false, true, true, true]).into_array(),
1003 &mut ctx
1004 );
1005 }
1006
1007 #[test]
1008 fn test_evaluate_with_nullable_condition() {
1009 let mut ctx = SESSION.create_execution_ctx();
1010 let test_array = StructArray::from_fields(&[(
1011 "cond",
1012 BoolArray::from_iter([Some(true), None, Some(false), None, Some(true)]).into_array(),
1013 )])
1014 .unwrap()
1015 .into_array();
1016
1017 let expr = case_when(get_item("cond", root()), lit(100i32), lit(0i32));
1018
1019 let result = evaluate_expr(&expr, &test_array);
1020 assert_arrays_eq!(result, buffer![100i32, 0, 0, 0, 100].into_array(), &mut ctx);
1021 }
1022
1023 #[test]
1024 fn test_evaluate_with_nullable_result_values() {
1025 let mut ctx = SESSION.create_execution_ctx();
1026 let test_array = StructArray::from_fields(&[
1027 ("value", buffer![1i32, 2, 3, 4, 5].into_array()),
1028 (
1029 "result",
1030 PrimitiveArray::from_option_iter([Some(10), None, Some(30), Some(40), Some(50)])
1031 .into_array(),
1032 ),
1033 ])
1034 .unwrap()
1035 .into_array();
1036
1037 let expr = case_when(
1038 gt(get_item("value", root()), lit(2i32)),
1039 get_item("result", root()),
1040 lit(0i32),
1041 );
1042
1043 let result = evaluate_expr(&expr, &test_array);
1044 assert_arrays_eq!(
1045 result,
1046 PrimitiveArray::from_option_iter([Some(0i32), Some(0), Some(30), Some(40), Some(50)])
1047 .into_array(),
1048 &mut ctx
1049 );
1050 }
1051
1052 #[test]
1053 fn test_evaluate_with_all_null_condition() {
1054 let mut ctx = SESSION.create_execution_ctx();
1055 let test_array = StructArray::from_fields(&[(
1056 "cond",
1057 BoolArray::from_iter([None, None, None]).into_array(),
1058 )])
1059 .unwrap()
1060 .into_array();
1061
1062 let expr = case_when(get_item("cond", root()), lit(100i32), lit(0i32));
1063
1064 let result = evaluate_expr(&expr, &test_array);
1065 assert_arrays_eq!(result, buffer![0i32, 0, 0].into_array(), &mut ctx);
1066 }
1067
1068 #[test]
1071 fn test_evaluate_nary_no_else_returns_null() {
1072 let mut ctx = SESSION.create_execution_ctx();
1073 let test_array =
1074 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
1075 .unwrap()
1076 .into_array();
1077
1078 let expr = nested_case_when(
1080 vec![
1081 (eq(get_item("value", root()), lit(1i32)), lit(10i32)),
1082 (eq(get_item("value", root()), lit(3i32)), lit(30i32)),
1083 ],
1084 None,
1085 );
1086
1087 let result = evaluate_expr(&expr, &test_array);
1088 assert!(result.dtype().is_nullable());
1089 assert_arrays_eq!(
1090 result,
1091 PrimitiveArray::from_option_iter([Some(10i32), None, Some(30), None, None])
1092 .into_array(),
1093 &mut ctx
1094 );
1095 }
1096
1097 #[test]
1098 fn test_evaluate_nary_many_conditions() {
1099 let mut ctx = SESSION.create_execution_ctx();
1100 let test_array =
1101 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4, 5].into_array())])
1102 .unwrap()
1103 .into_array();
1104
1105 let expr = nested_case_when(
1107 vec![
1108 (eq(get_item("value", root()), lit(1i32)), lit(10i32)),
1109 (eq(get_item("value", root()), lit(2i32)), lit(20i32)),
1110 (eq(get_item("value", root()), lit(3i32)), lit(30i32)),
1111 (eq(get_item("value", root()), lit(4i32)), lit(40i32)),
1112 (eq(get_item("value", root()), lit(5i32)), lit(50i32)),
1113 ],
1114 Some(lit(0i32)),
1115 );
1116
1117 let result = evaluate_expr(&expr, &test_array);
1118 assert_arrays_eq!(
1119 result,
1120 buffer![10i32, 20, 30, 40, 50].into_array(),
1121 &mut ctx
1122 );
1123 }
1124
1125 #[test]
1126 fn test_evaluate_nary_all_false_no_else() {
1127 let mut ctx = SESSION.create_execution_ctx();
1128 let test_array = StructArray::from_fields(&[("value", buffer![1i32, 2, 3].into_array())])
1129 .unwrap()
1130 .into_array();
1131
1132 let expr = nested_case_when(
1134 vec![
1135 (gt(get_item("value", root()), lit(100i32)), lit(10i32)),
1136 (gt(get_item("value", root()), lit(200i32)), lit(20i32)),
1137 ],
1138 None,
1139 );
1140
1141 let result = evaluate_expr(&expr, &test_array);
1142 assert!(result.dtype().is_nullable());
1143 assert_arrays_eq!(
1144 result,
1145 PrimitiveArray::from_option_iter([None::<i32>, None, None]).into_array(),
1146 &mut ctx
1147 );
1148 }
1149
1150 #[test]
1151 fn test_evaluate_nary_overlapping_conditions_first_wins() {
1152 let mut ctx = SESSION.create_execution_ctx();
1153 let test_array =
1154 StructArray::from_fields(&[("value", buffer![10i32, 20, 30].into_array())])
1155 .unwrap()
1156 .into_array();
1157
1158 let expr = nested_case_when(
1162 vec![
1163 (gt(get_item("value", root()), lit(5i32)), lit(1i32)),
1164 (gt(get_item("value", root()), lit(0i32)), lit(2i32)),
1165 (gt(get_item("value", root()), lit(15i32)), lit(3i32)),
1166 ],
1167 Some(lit(0i32)),
1168 );
1169
1170 let result = evaluate_expr(&expr, &test_array);
1171 assert_arrays_eq!(result, buffer![1i32, 1, 1].into_array(), &mut ctx);
1173 }
1174
1175 #[test]
1176 fn test_evaluate_nary_early_exit_when_remaining_empty() {
1177 let mut ctx = SESSION.create_execution_ctx();
1178 let test_array = StructArray::from_fields(&[("value", buffer![1i32, 2, 3].into_array())])
1181 .unwrap()
1182 .into_array();
1183
1184 let expr = nested_case_when(
1185 vec![
1186 (gt(get_item("value", root()), lit(0i32)), lit(100i32)),
1187 (gt(get_item("value", root()), lit(0i32)), lit(999i32)),
1189 ],
1190 Some(lit(0i32)),
1191 );
1192
1193 let result = evaluate_expr(&expr, &test_array);
1194 assert_arrays_eq!(result, buffer![100i32, 100, 100].into_array(), &mut ctx);
1195 }
1196
1197 #[test]
1198 fn test_evaluate_nary_skips_branch_with_empty_effective_mask() {
1199 let mut ctx = SESSION.create_execution_ctx();
1200 let test_array = StructArray::from_fields(&[("value", buffer![1i32, 2, 3].into_array())])
1203 .unwrap()
1204 .into_array();
1205
1206 let expr = nested_case_when(
1207 vec![
1208 (eq(get_item("value", root()), lit(1i32)), lit(10i32)),
1209 (eq(get_item("value", root()), lit(1i32)), lit(999i32)),
1212 (eq(get_item("value", root()), lit(2i32)), lit(20i32)),
1213 ],
1214 Some(lit(0i32)),
1215 );
1216
1217 let result = evaluate_expr(&expr, &test_array);
1218 assert_arrays_eq!(result, buffer![10i32, 20, 0].into_array(), &mut ctx);
1219 }
1220
1221 #[test]
1222 fn test_evaluate_nary_string_output() -> VortexResult<()> {
1223 let test_array =
1225 StructArray::from_fields(&[("value", buffer![1i32, 2, 3, 4].into_array())])?
1226 .into_array();
1227
1228 let expr = nested_case_when(
1232 vec![
1233 (gt(get_item("value", root()), lit(2i32)), lit("high")),
1234 (gt(get_item("value", root()), lit(0i32)), lit("low")),
1235 ],
1236 Some(lit("none")),
1237 );
1238
1239 let result = evaluate_expr(&expr, &test_array);
1240 assert_eq!(
1241 result.execute_scalar(0, &mut SESSION.create_execution_ctx())?,
1242 Scalar::utf8("low", Nullability::NonNullable)
1243 );
1244 assert_eq!(
1245 result.execute_scalar(1, &mut SESSION.create_execution_ctx())?,
1246 Scalar::utf8("low", Nullability::NonNullable)
1247 );
1248 assert_eq!(
1249 result.execute_scalar(2, &mut SESSION.create_execution_ctx())?,
1250 Scalar::utf8("high", Nullability::NonNullable)
1251 );
1252 assert_eq!(
1253 result.execute_scalar(3, &mut SESSION.create_execution_ctx())?,
1254 Scalar::utf8("high", Nullability::NonNullable)
1255 );
1256 Ok(())
1257 }
1258
1259 #[test]
1260 fn test_evaluate_nary_with_nullable_conditions() {
1261 let mut ctx = SESSION.create_execution_ctx();
1262 let test_array = StructArray::from_fields(&[
1263 (
1264 "cond1",
1265 BoolArray::from_iter([Some(true), None, Some(false)]).into_array(),
1266 ),
1267 (
1268 "cond2",
1269 BoolArray::from_iter([Some(false), Some(true), None]).into_array(),
1270 ),
1271 ])
1272 .unwrap()
1273 .into_array();
1274
1275 let expr = nested_case_when(
1276 vec![
1277 (get_item("cond1", root()), lit(10i32)),
1278 (get_item("cond2", root()), lit(20i32)),
1279 ],
1280 Some(lit(0i32)),
1281 );
1282
1283 let result = evaluate_expr(&expr, &test_array);
1284 assert_arrays_eq!(result, buffer![10i32, 20, 0].into_array(), &mut ctx);
1288 }
1289
1290 fn nullable_i64_scope(fields: &[&str]) -> DType {
1294 DType::Struct(
1295 StructFields::new(
1296 fields.to_vec().into(),
1297 vec![DType::Primitive(PType::I64, Nullability::Nullable); fields.len()],
1298 ),
1299 Nullability::NonNullable,
1300 )
1301 }
1302
1303 #[test]
1304 fn test_simplify_coalesce_is_null_rewrites_to_fill_null() -> VortexResult<()> {
1305 let expr = case_when(is_null(col("x")), lit(0i64), col("x"));
1307 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x"]))?;
1308 assert!(
1309 optimized.to_string().starts_with("vortex.fill_null"),
1310 "expected fill_null, got {optimized}"
1311 );
1312 Ok(())
1313 }
1314
1315 #[test]
1316 fn test_simplify_coalesce_is_not_null_rewrites_to_fill_null() -> VortexResult<()> {
1317 let expr = case_when(is_not_null(col("x")), col("x"), lit(0i64));
1319 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x"]))?;
1320 assert!(
1321 optimized.to_string().starts_with("vortex.fill_null"),
1322 "expected fill_null, got {optimized}"
1323 );
1324 Ok(())
1325 }
1326
1327 #[test]
1328 fn test_simplify_does_not_fire_when_operands_differ() -> VortexResult<()> {
1329 let expr = case_when(is_null(col("x")), lit(0i64), col("y"));
1331 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x", "y"]))?;
1332 let s = optimized.to_string();
1333 assert!(s.contains("CASE"), "expected CASE WHEN to remain, got {s}");
1334 assert!(!s.contains("fill_null"), "must not rewrite, got {s}");
1335 Ok(())
1336 }
1337
1338 #[test]
1339 fn test_simplify_does_not_fire_for_non_constant_fill() -> VortexResult<()> {
1340 let expr = case_when(is_null(col("x")), col("c"), col("x"));
1343 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x", "c"]))?;
1344 let s = optimized.to_string();
1345 assert!(s.contains("CASE"), "expected CASE WHEN to remain, got {s}");
1346 assert!(!s.contains("fill_null"), "must not rewrite, got {s}");
1347 Ok(())
1348 }
1349
1350 #[test]
1351 fn test_simplify_null_fill_collapses_to_input() -> VortexResult<()> {
1352 let null_fill = || {
1356 lit(Scalar::null(DType::Primitive(
1357 PType::I64,
1358 Nullability::Nullable,
1359 )))
1360 };
1361
1362 for expr in [
1363 case_when(is_null(col("x")), null_fill(), col("x")),
1364 case_when(is_not_null(col("x")), col("x"), null_fill()),
1365 ] {
1366 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x"]))?;
1367 assert_eq!(
1368 optimized.to_string(),
1369 "$.x",
1370 "expected collapse to input column, got {optimized}"
1371 );
1372 }
1373 Ok(())
1374 }
1375
1376 #[test]
1377 fn test_simplify_null_fill_semantic_equivalence() -> VortexResult<()> {
1378 let mut ctx = SESSION.create_execution_ctx();
1379 let array = PrimitiveArray::from_option_iter([Some(1i64), None, Some(3)]).into_array();
1381 let scope = DType::Primitive(PType::I64, Nullability::Nullable);
1382 let null_fill = lit(Scalar::null(DType::Primitive(
1383 PType::I64,
1384 Nullability::Nullable,
1385 )));
1386
1387 let original = case_when(is_null(root()), null_fill, root());
1388 let optimized = original.optimize_recursive(&scope)?;
1389 assert_eq!(
1390 optimized.to_string(),
1391 "$",
1392 "expected collapse to root, got {optimized}"
1393 );
1394
1395 let expected = PrimitiveArray::from_option_iter([Some(1i64), None, Some(3)]).into_array();
1396 assert_arrays_eq!(evaluate_expr(&original, &array), expected, &mut ctx);
1397 assert_arrays_eq!(evaluate_expr(&optimized, &array), expected, &mut ctx);
1398 Ok(())
1399 }
1400
1401 #[test]
1402 fn test_simplify_does_not_fire_without_else() -> VortexResult<()> {
1403 let expr = case_when_no_else(is_null(col("x")), lit(0i64));
1404 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x"]))?;
1405 assert!(
1406 !optimized.to_string().contains("fill_null"),
1407 "must not rewrite a no-ELSE case_when, got {optimized}"
1408 );
1409 Ok(())
1410 }
1411
1412 #[test]
1413 fn test_simplify_does_not_fire_for_multi_pair() -> VortexResult<()> {
1414 let expr = nested_case_when(
1415 vec![
1416 (is_null(col("x")), lit(0i64)),
1417 (gt(col("x"), lit(5i64)), lit(1i64)),
1418 ],
1419 Some(col("x")),
1420 );
1421 let optimized = expr.optimize_recursive(&nullable_i64_scope(&["x"]))?;
1422 assert!(
1423 !optimized.to_string().contains("fill_null"),
1424 "must not rewrite a multi-pair case_when, got {optimized}"
1425 );
1426 Ok(())
1427 }
1428
1429 #[test]
1430 fn test_simplify_semantic_equivalence() -> VortexResult<()> {
1431 let mut ctx = SESSION.create_execution_ctx();
1432 let array = PrimitiveArray::from_option_iter([Some(1i64), None, Some(3)]).into_array();
1434 let scope = DType::Primitive(PType::I64, Nullability::Nullable);
1435
1436 let original = case_when(is_null(root()), lit(0i64), root());
1437 let optimized = original.optimize_recursive(&scope)?;
1438 assert!(
1439 optimized.to_string().starts_with("vortex.fill_null"),
1440 "expected fill_null, got {optimized}"
1441 );
1442
1443 assert_arrays_eq!(
1446 evaluate_expr(&original, &array),
1447 PrimitiveArray::from_option_iter([Some(1i64), Some(0), Some(3)]).into_array(),
1448 &mut ctx
1449 );
1450 assert_arrays_eq!(
1451 evaluate_expr(&optimized, &array),
1452 buffer![1i64, 0, 3].into_array(),
1453 &mut ctx
1454 );
1455 Ok(())
1456 }
1457
1458 #[test]
1459 fn test_merge_case_branches_alternating_mask() -> VortexResult<()> {
1460 let mut ctx = SESSION.create_execution_ctx();
1461 let n = 100usize;
1464
1465 let branch0_mask = Mask::from_indices(n, (0..n).step_by(2));
1467 let branch1_mask = Mask::from_indices(n, (1..n).step_by(2));
1468
1469 let result = merge_case_branches(
1470 vec![
1471 (
1472 branch0_mask,
1473 PrimitiveArray::from_option_iter(vec![Some(0i32); n]).into_array(),
1474 ),
1475 (
1476 branch1_mask,
1477 PrimitiveArray::from_option_iter(vec![Some(1i32); n]).into_array(),
1478 ),
1479 ],
1480 PrimitiveArray::from_option_iter(vec![Some(99i32); n]).into_array(),
1481 &mut SESSION.create_execution_ctx(),
1482 )?;
1483
1484 let expected: Vec<Option<i32>> = (0..n)
1486 .map(|v| if v % 2 == 0 { Some(0) } else { Some(1) })
1487 .collect();
1488 assert_arrays_eq!(
1489 result,
1490 PrimitiveArray::from_option_iter(expected).into_array(),
1491 &mut ctx
1492 );
1493 Ok(())
1494 }
1495}