1use std::cmp::Ordering;
16use std::collections::{BTreeMap, HashSet};
17use std::fmt::{self, Debug};
18use std::ops::Bound;
19use std::sync::LazyLock;
20
21use fixedbitset::FixedBitSet;
22use itertools::Itertools;
23use risingwave_common::catalog::Schema;
24use risingwave_common::types::{DataType, DefaultOrd, ScalarImpl};
25use risingwave_common::util::iter_util::ZipEqFast;
26use risingwave_common::util::scan_range::{ScanRange, is_full_range};
27use risingwave_common::util::sort_util::{OrderType, cmp_rows};
28
29use crate::TableCatalog;
30use crate::error::Result;
31use crate::expr::{
32 ExprDisplay, ExprImpl, ExprMutator, ExprRewriter, ExprType, ExprVisitor, FunctionCall,
33 InequalityInputPair, InputRef, collect_input_refs, column_self_eq_eliminate,
34 factorization_expr, fold_boolean_constant, push_down_not, to_conjunctions,
35 try_get_bool_constant,
36};
37use crate::utils::condition::cast_compare::{ResultForCmp, ResultForEq};
38
39#[derive(Debug, Clone, PartialEq, Eq, Hash)]
40pub struct Condition {
41 pub conjunctions: Vec<ExprImpl>,
43}
44
45impl IntoIterator for Condition {
46 type IntoIter = std::vec::IntoIter<ExprImpl>;
47 type Item = ExprImpl;
48
49 fn into_iter(self) -> Self::IntoIter {
50 self.conjunctions.into_iter()
51 }
52}
53
54impl fmt::Display for Condition {
55 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
56 let mut conjunctions = self.conjunctions.iter();
57 if let Some(expr) = conjunctions.next() {
58 write!(f, "{:?}", expr)?;
59 }
60 if self.always_true() {
61 write!(f, "true")?;
62 } else {
63 for expr in conjunctions {
64 write!(f, " AND {:?}", expr)?;
65 }
66 }
67 Ok(())
68 }
69}
70
71impl Condition {
72 pub fn with_expr(expr: ExprImpl) -> Self {
73 let conjunctions = to_conjunctions(expr);
74
75 Self { conjunctions }.simplify()
76 }
77
78 pub fn true_cond() -> Self {
79 Self {
80 conjunctions: vec![],
81 }
82 }
83
84 pub fn false_cond() -> Self {
85 Self {
86 conjunctions: vec![ExprImpl::literal_bool(false)],
87 }
88 }
89
90 pub fn always_true(&self) -> bool {
91 self.conjunctions.is_empty()
92 }
93
94 pub fn always_false(&self) -> bool {
95 static FALSE: LazyLock<ExprImpl> = LazyLock::new(|| ExprImpl::literal_bool(false));
96 !self.conjunctions.is_empty() && self.conjunctions.contains(&*FALSE)
98 }
99
100 pub fn as_expr_unless_true(&self) -> Option<ExprImpl> {
102 if self.always_true() {
103 None
104 } else {
105 Some(self.clone().into())
106 }
107 }
108
109 #[must_use]
110 pub fn and(self, other: Self) -> Self {
111 let mut ret = self;
112 ret.conjunctions.extend(other.conjunctions);
113 ret.simplify()
114 }
115
116 #[must_use]
117 pub fn or(self, other: Self) -> Self {
118 let or_expr = ExprImpl::FunctionCall(
119 FunctionCall::new_unchecked(
120 ExprType::Or,
121 vec![self.into(), other.into()],
122 DataType::Boolean,
123 )
124 .into(),
125 );
126 let ret = Self::with_expr(or_expr);
127 ret.simplify()
128 }
129
130 #[must_use]
132 pub fn split(self, left_col_num: usize, right_col_num: usize) -> (Self, Self, Self) {
133 let left_bit_map = FixedBitSet::from_iter(0..left_col_num);
134 let right_bit_map = FixedBitSet::from_iter(left_col_num..left_col_num + right_col_num);
135
136 self.group_by::<_, 3>(|expr| {
137 let input_bits = expr.collect_input_refs(left_col_num + right_col_num);
138 if input_bits.is_subset(&left_bit_map) {
139 0
140 } else if input_bits.is_subset(&right_bit_map) {
141 1
142 } else {
143 2
144 }
145 })
146 .into_iter()
147 .next_tuple()
148 .unwrap()
149 }
150
151 pub fn collect_input_refs(&self, input_col_num: usize) -> FixedBitSet {
156 collect_input_refs(input_col_num, &self.conjunctions)
157 }
158
159 #[must_use]
173 pub fn split_by_input_col_nums(
174 self,
175 input_col_nums: &[usize],
176 only_eq: bool,
177 ) -> (BTreeMap<(usize, usize), Self>, Self) {
178 let mut bitmaps = Vec::with_capacity(input_col_nums.len());
179 let mut cols_seen = 0;
180 for cols in input_col_nums {
181 bitmaps.push(FixedBitSet::from_iter(cols_seen..cols_seen + cols));
182 cols_seen += cols;
183 }
184
185 let mut pairwise_conditions = BTreeMap::new();
186 let mut non_eq_join = vec![];
187
188 for expr in self.conjunctions {
189 let input_bits = expr.collect_input_refs(cols_seen);
190 let mut subset_indices = Vec::with_capacity(input_col_nums.len());
191 for (idx, bitmap) in bitmaps.iter().enumerate() {
192 if !input_bits.is_disjoint(bitmap) {
193 subset_indices.push(idx);
194 }
195 }
196 if subset_indices.len() != 2 || (only_eq && expr.as_eq_cond().is_none()) {
197 non_eq_join.push(expr);
198 } else {
199 let key = if subset_indices[0] < subset_indices[1] {
201 (subset_indices[0], subset_indices[1])
202 } else {
203 (subset_indices[1], subset_indices[0])
204 };
205 let e = pairwise_conditions
206 .entry(key)
207 .or_insert_with(Condition::true_cond);
208 e.conjunctions.push(expr);
209 }
210 }
211 (
212 pairwise_conditions,
213 Condition {
214 conjunctions: non_eq_join,
215 },
216 )
217 }
218
219 #[must_use]
220 pub fn split_eq_keys(
227 self,
228 left_col_num: usize,
229 right_col_num: usize,
230 ) -> (Vec<(InputRef, InputRef, bool)>, Self) {
231 let left_bit_map = FixedBitSet::from_iter(0..left_col_num);
232 let right_bit_map = FixedBitSet::from_iter(left_col_num..left_col_num + right_col_num);
233
234 let (mut eq_keys, mut others) = (vec![], vec![]);
235 self.conjunctions.into_iter().for_each(|expr| {
236 let input_bits = expr.collect_input_refs(left_col_num + right_col_num);
237 if input_bits.is_disjoint(&left_bit_map) || input_bits.is_disjoint(&right_bit_map) {
238 others.push(expr)
239 } else if let Some(columns) = expr.as_eq_cond() {
240 eq_keys.push((columns.0, columns.1, false));
241 } else if let Some(columns) = expr.as_is_not_distinct_from_cond() {
242 eq_keys.push((columns.0, columns.1, true));
243 } else {
244 others.push(expr)
245 }
246 });
247
248 (
249 eq_keys,
250 Condition {
251 conjunctions: others,
252 },
253 )
254 }
255
256 pub(crate) fn extract_inequality_keys(
267 &self,
268 left_col_num: usize,
269 right_col_num: usize,
270 ) -> Vec<(usize, InequalityInputPair)> {
271 let left_bit_map = FixedBitSet::from_iter(0..left_col_num);
272 let right_bit_map = FixedBitSet::from_iter(left_col_num..left_col_num + right_col_num);
273
274 self.conjunctions
275 .iter()
276 .enumerate()
277 .filter_map(|(conjunction_idx, expr)| {
278 let input_bits = expr.collect_input_refs(left_col_num + right_col_num);
279 if input_bits.is_disjoint(&left_bit_map) || input_bits.is_disjoint(&right_bit_map) {
280 return None;
281 }
282
283 let (left_input, op, right_input) = expr.as_comparison_cond()?;
285
286 if left_input.index() < left_col_num
289 && right_input.index() >= left_col_num
290 && right_input.index() < left_col_num + right_col_num
291 {
292 Some((
293 conjunction_idx,
294 InequalityInputPair::new(
295 left_input.index(),
296 right_input.index() - left_col_num, op,
298 ),
299 ))
300 } else {
301 None
302 }
303 })
304 .collect_vec()
305 }
306
307 #[must_use]
310 pub fn split_disjoint(self, columns: &FixedBitSet) -> (Self, Self) {
311 self.group_by::<_, 2>(|expr| {
312 let input_bits = expr.collect_input_refs(columns.len());
313 input_bits.is_disjoint(columns) as usize
314 })
315 .into_iter()
316 .next_tuple()
317 .unwrap()
318 }
319
320 fn disjunctions_to_scan_ranges(
324 table: &TableCatalog,
325 max_split_range_gap: u64,
326 disjunctions: Vec<ExprImpl>,
327 ) -> Result<Option<(Vec<ScanRange>, bool)>> {
328 let disjunctions_result: Result<Vec<(Vec<ScanRange>, Self)>> = disjunctions
329 .into_iter()
330 .map(|x| {
331 Condition {
332 conjunctions: to_conjunctions(x),
333 }
334 .split_to_scan_ranges(table, max_split_range_gap)
335 })
336 .collect();
337
338 let disjunctions_result = disjunctions_result?;
340
341 let all_equal = disjunctions_result
344 .iter()
345 .all(|(scan_ranges, other_condition)| {
346 other_condition.always_true()
347 && scan_ranges
348 .iter()
349 .all(|x| !x.eq_conds.is_empty() && is_full_range(&x.range))
350 });
351
352 if all_equal {
353 let scan_ranges = disjunctions_result
357 .into_iter()
358 .flat_map(|(scan_ranges, _)| scan_ranges)
359 .sorted_by(|a, b| a.eq_conds.len().cmp(&b.eq_conds.len()))
361 .collect_vec();
362 let mut non_overlap_scan_ranges: Vec<ScanRange> = vec![];
364 for s1 in &scan_ranges {
365 let overlap = non_overlap_scan_ranges.iter().any(|s2| {
366 #[expect(clippy::disallowed_methods)]
367 s1.eq_conds
368 .iter()
369 .zip(s2.eq_conds.iter())
370 .all(|(a, b)| a == b)
371 });
372 if !overlap {
376 non_overlap_scan_ranges.push(s1.clone());
377 }
378 }
379
380 Ok(Some((non_overlap_scan_ranges, false)))
381 } else {
382 let mut scan_ranges = vec![];
383 for (scan_ranges_chunk, _) in disjunctions_result {
384 if scan_ranges_chunk.is_empty() {
385 return Ok(None);
387 }
388
389 scan_ranges.extend(scan_ranges_chunk);
390 }
391
392 let order_types = table
393 .pk
394 .iter()
395 .cloned()
396 .map(|x| {
397 if x.order_type.is_descending() {
398 x.order_type.reverse()
399 } else {
400 x.order_type
401 }
402 })
403 .collect_vec();
404 scan_ranges.sort_by(|left, right| {
405 let (left_start, _left_end) = &left.convert_to_range();
406 let (right_start, _right_end) = &right.convert_to_range();
407
408 let left_start_vec = match &left_start {
409 Bound::Included(vec) | Bound::Excluded(vec) => vec,
410 _ => &vec![],
411 };
412 let right_start_vec = match &right_start {
413 Bound::Included(vec) | Bound::Excluded(vec) => vec,
414 _ => &vec![],
415 };
416
417 if left_start_vec.is_empty() && right_start_vec.is_empty() {
418 return Ordering::Less;
419 }
420
421 if left_start_vec.is_empty() {
422 return Ordering::Less;
423 }
424
425 if right_start_vec.is_empty() {
426 return Ordering::Greater;
427 }
428
429 let cmp_column_len = left_start_vec.len().min(right_start_vec.len());
430 cmp_rows(
431 &left_start_vec[0..cmp_column_len],
432 &right_start_vec[0..cmp_column_len],
433 &order_types[0..cmp_column_len],
434 )
435 });
436
437 if scan_ranges.is_empty() {
438 return Ok(None);
439 }
440
441 if scan_ranges.len() == 1 {
442 return Ok(Some((scan_ranges, true)));
443 }
444
445 let mut output_scan_ranges: Vec<ScanRange> = vec![];
446 output_scan_ranges.push(scan_ranges[0].clone());
447 let mut idx = 1;
448 loop {
449 if idx >= scan_ranges.len() {
450 break;
451 }
452
453 let scan_range_left = output_scan_ranges.last_mut().unwrap();
454 let scan_range_right = &scan_ranges[idx];
455
456 if scan_range_left.eq_conds == scan_range_right.eq_conds {
457 if !ScanRange::is_overlap(scan_range_left, scan_range_right, &order_types) {
460 output_scan_ranges.push(scan_range_right.clone());
462 idx += 1;
463 continue;
464 }
465
466 fn merge_bound(
468 left_scan_range: &Bound<Vec<Option<ScalarImpl>>>,
469 right_scan_range: &Bound<Vec<Option<ScalarImpl>>>,
470 order_types: &[OrderType],
471 left_bound: bool,
472 ) -> Bound<Vec<Option<ScalarImpl>>> {
473 let left_scan_range = match left_scan_range {
474 Bound::Included(vec) | Bound::Excluded(vec) => vec,
475 Bound::Unbounded => return Bound::Unbounded,
476 };
477
478 let right_scan_range = match right_scan_range {
479 Bound::Included(vec) | Bound::Excluded(vec) => vec,
480 Bound::Unbounded => return Bound::Unbounded,
481 };
482
483 let cmp_len = left_scan_range.len().min(right_scan_range.len());
484
485 let cmp = cmp_rows(
486 &left_scan_range[..cmp_len],
487 &right_scan_range[..cmp_len],
488 &order_types[..cmp_len],
489 );
490
491 let bound = {
492 if (cmp.is_le() && left_bound) || (cmp.is_ge() && !left_bound) {
493 left_scan_range.clone()
494 } else {
495 right_scan_range.clone()
496 }
497 };
498
499 Bound::Included(bound)
501 }
502
503 scan_range_left.range.0 = merge_bound(
504 &scan_range_left.range.0,
505 &scan_range_right.range.0,
506 &order_types,
507 true,
508 );
509
510 scan_range_left.range.1 = merge_bound(
511 &scan_range_left.range.1,
512 &scan_range_right.range.1,
513 &order_types,
514 false,
515 );
516
517 if scan_range_left.is_full_table_scan() {
518 return Ok(None);
519 }
520 } else {
521 output_scan_ranges.push(scan_range_right.clone());
522 }
523
524 idx += 1;
525 }
526
527 Ok(Some((output_scan_ranges, true)))
528 }
529 }
530
531 fn split_row_cmp_to_scan_ranges(
532 &self,
533 table: &TableCatalog,
534 ) -> Result<Option<(Vec<ScanRange>, Self)>> {
535 let (mut row_conjunctions, row_conjunctions_without_struct): (Vec<_>, Vec<_>) =
536 self.conjunctions.clone().into_iter().partition(|expr| {
537 if let Some(f) = expr.as_function_call() {
538 if let Some(left_input) = f.inputs().get(0)
539 && let Some(left_input) = left_input.as_function_call()
540 && matches!(left_input.func_type(), ExprType::Row)
541 && left_input.inputs().iter().all(|x| x.is_input_ref())
542 && let Some(right_input) = f.inputs().get(1)
543 && right_input.is_literal()
544 {
545 true
546 } else {
547 false
548 }
549 } else {
550 false
551 }
552 });
553 if row_conjunctions.len() == 1 {
558 let row_conjunction = row_conjunctions.pop().unwrap();
559 let row_left_inputs = row_conjunction
560 .as_function_call()
561 .unwrap()
562 .inputs()
563 .get(0)
564 .unwrap()
565 .as_function_call()
566 .unwrap()
567 .inputs();
568 let row_right_literal = row_conjunction
569 .as_function_call()
570 .unwrap()
571 .inputs()
572 .get(1)
573 .unwrap()
574 .as_literal()
575 .unwrap();
576 if !matches!(row_right_literal.get_data(), Some(ScalarImpl::Struct(_))) {
577 return Ok(None);
578 }
579 let row_right_literal_data = row_right_literal.get_data().clone().unwrap();
580 let right_iter = row_right_literal_data.as_struct().fields();
581 let func_type = row_conjunction.as_function_call().unwrap().func_type();
582 if row_left_inputs.len() > 1
583 && (matches!(func_type, ExprType::LessThan)
584 || matches!(func_type, ExprType::GreaterThan))
585 {
586 let mut pk_struct = vec![];
587 let mut order_type = None;
588 let mut all_added = true;
589 let mut iter = row_left_inputs.iter().zip_eq_fast(right_iter);
590 for column_order in &table.pk {
591 if let Some((left_expr, right_expr)) = iter.next() {
592 if left_expr.as_input_ref().unwrap().index != column_order.column_index {
593 all_added = false;
594 break;
595 }
596 match order_type {
597 Some(o) => {
598 if o != column_order.order_type {
599 all_added = false;
600 break;
601 }
602 }
603 None => order_type = Some(column_order.order_type),
604 }
605 pk_struct.push(right_expr.clone());
606 }
607 }
608
609 if !pk_struct.is_empty() {
611 if !all_added {
612 let scan_range = ScanRange {
613 eq_conds: vec![],
614 range: match func_type {
615 ExprType::GreaterThan => {
616 (Bound::Included(pk_struct), Bound::Unbounded)
617 }
618 ExprType::LessThan => {
619 (Bound::Unbounded, Bound::Included(pk_struct))
620 }
621 _ => unreachable!(),
622 },
623 };
624 return Ok(Some((
625 vec![scan_range],
626 Condition {
627 conjunctions: self.conjunctions.clone(),
628 },
629 )));
630 } else {
631 let scan_range = ScanRange {
632 eq_conds: vec![],
633 range: match func_type {
634 ExprType::GreaterThan => {
635 (Bound::Excluded(pk_struct), Bound::Unbounded)
636 }
637 ExprType::LessThan => {
638 (Bound::Unbounded, Bound::Excluded(pk_struct))
639 }
640 _ => unreachable!(),
641 },
642 };
643 return Ok(Some((
644 vec![scan_range],
645 Condition {
646 conjunctions: row_conjunctions_without_struct,
647 },
648 )));
649 }
650 }
651 }
652 }
653 Ok(None)
654 }
655
656 pub fn get_eq_const_input_refs(&self) -> Vec<InputRef> {
658 self.conjunctions
659 .iter()
660 .filter_map(|expr| expr.as_eq_const().map(|(input_ref, _)| input_ref))
661 .collect()
662 }
663
664 pub fn split_to_scan_ranges(
666 self,
667 table: &TableCatalog,
668 max_split_range_gap: u64,
669 ) -> Result<(Vec<ScanRange>, Self)> {
670 fn false_cond() -> (Vec<ScanRange>, Condition) {
671 (vec![], Condition::false_cond())
672 }
673
674 if self.conjunctions.len() == 1
676 && let Some(disjunctions) = self.conjunctions[0].as_or_disjunctions()
677 {
678 if let Some((scan_ranges, maintaining_condition)) =
679 Self::disjunctions_to_scan_ranges(table, max_split_range_gap, disjunctions)?
680 {
681 if maintaining_condition {
682 return Ok((scan_ranges, self));
683 } else {
684 return Ok((scan_ranges, Condition::true_cond()));
685 }
686 } else {
687 return Ok((vec![], self));
688 }
689 }
690 if let Some((scan_ranges, other_condition)) = self.split_row_cmp_to_scan_ranges(table)? {
691 return Ok((scan_ranges, other_condition));
692 }
693
694 let mut groups = Self::classify_conjunctions_by_pk(self.conjunctions, table);
695 let mut other_conds = groups.pop().unwrap();
696
697 let mut scan_range = ScanRange::full_table_scan();
699 for i in 0..table.pk.len() {
700 let group = std::mem::take(&mut groups[i]);
701 if group.is_empty() {
702 groups.push(other_conds);
703 return Ok((
704 if scan_range.is_full_table_scan() {
705 vec![]
706 } else {
707 vec![scan_range]
708 },
709 Self {
710 conjunctions: groups[i + 1..].concat(),
711 },
712 ));
713 }
714
715 let Some((
716 lower_bound_conjunctions,
717 upper_bound_conjunctions,
718 eq_conds,
719 part_of_other_conds,
720 )) = Self::analyze_group(group)?
721 else {
722 return Ok(false_cond());
723 };
724 other_conds.extend(part_of_other_conds);
725
726 let lower_bound = Self::merge_lower_bound_conjunctions(lower_bound_conjunctions);
727 let upper_bound = Self::merge_upper_bound_conjunctions(upper_bound_conjunctions);
728
729 if Self::is_invalid_range(&lower_bound, &upper_bound) {
730 return Ok(false_cond());
731 }
732
733 match eq_conds.len() {
735 1 => {
736 let eq_conds =
737 Self::extract_eq_conds_within_range(eq_conds, &upper_bound, &lower_bound);
738 if eq_conds.is_empty() {
739 return Ok(false_cond());
740 }
741 scan_range.eq_conds.extend(eq_conds);
742 }
743 0 => {
744 let convert = |bound| match bound {
745 Bound::Included(l) => Bound::Included(vec![Some(l)]),
746 Bound::Excluded(l) => Bound::Excluded(vec![Some(l)]),
747 Bound::Unbounded => Bound::Unbounded,
748 };
749 scan_range.range = (convert(lower_bound), convert(upper_bound));
750 other_conds.extend(groups[i + 1..].iter().flatten().cloned());
751 break;
752 }
753 _ => {
754 let eq_conds =
761 Self::extract_eq_conds_within_range(eq_conds, &upper_bound, &lower_bound);
762 if eq_conds.is_empty() {
763 return Ok(false_cond());
764 }
765 other_conds.extend(groups[i + 1..].iter().flatten().cloned());
766 let scan_ranges = eq_conds
767 .into_iter()
768 .map(|lit| {
769 let mut scan_range = scan_range.clone();
770 scan_range.eq_conds.push(lit);
771 scan_range
772 })
773 .collect();
774 return Ok((
775 scan_ranges,
776 Self {
777 conjunctions: other_conds,
778 },
779 ));
780 }
781 }
782 }
783
784 Ok((
785 if scan_range.is_full_table_scan() {
786 vec![]
787 } else if table.columns[table.pk[0].column_index].data_type.is_int() {
788 match scan_range.split_small_range(max_split_range_gap) {
789 Some(scan_ranges) => scan_ranges,
790 None => vec![scan_range],
791 }
792 } else {
793 vec![scan_range]
794 },
795 Self {
796 conjunctions: other_conds,
797 },
798 ))
799 }
800
801 fn classify_conjunctions_by_pk(
805 conjunctions: Vec<ExprImpl>,
806 table: &TableCatalog,
807 ) -> Vec<Vec<ExprImpl>> {
808 let pk_cols_num = table.pk.len();
809 let cols_num = table.columns.len();
810
811 let mut col_idx_to_pk_idx = vec![None; cols_num];
812 table
813 .order_column_indices()
814 .enumerate()
815 .for_each(|(idx, pk_idx)| {
816 col_idx_to_pk_idx[pk_idx] = Some(idx);
817 });
818
819 let mut groups = vec![vec![]; pk_cols_num + 1];
820 for (key, group) in &conjunctions.into_iter().chunk_by(|expr| {
821 let input_bits = expr.collect_input_refs(cols_num);
822 if input_bits.count_ones(..) == 1 {
823 let col_idx = input_bits.ones().next().unwrap();
824 col_idx_to_pk_idx[col_idx].unwrap_or(pk_cols_num)
825 } else {
826 pk_cols_num
827 }
828 }) {
829 groups[key].extend(group);
830 }
831
832 groups
833 }
834
835 #[expect(clippy::type_complexity)]
843 fn analyze_group(
844 group: Vec<ExprImpl>,
845 ) -> Result<
846 Option<(
847 Vec<Bound<ScalarImpl>>,
848 Vec<Bound<ScalarImpl>>,
849 Vec<Option<ScalarImpl>>,
850 Vec<ExprImpl>,
851 )>,
852 > {
853 let mut lower_bound_conjunctions = vec![];
854 let mut upper_bound_conjunctions = vec![];
855 let mut eq_conds = vec![];
857 let mut other_conds = vec![];
858
859 'group_loop: for expr in group {
861 if let Some((input_ref, const_expr)) = expr.as_eq_const() {
862 let new_expr = if let Ok(expr) =
863 const_expr.clone().cast_implicit(&input_ref.data_type)
864 {
865 expr
866 } else {
867 match self::cast_compare::cast_compare_for_eq(const_expr, input_ref.data_type) {
868 Ok(ResultForEq::Success(expr)) => expr,
869 Ok(ResultForEq::NeverEqual) => {
870 return Ok(None);
871 }
872 Err(_) => {
873 other_conds.push(expr);
874 continue;
875 }
876 }
877 };
878
879 let Some(new_cond) = new_expr.fold_const()? else {
880 return Ok(None);
882 };
883 if Self::mutual_exclusive_with_eq_conds(&new_cond, &eq_conds) {
884 return Ok(None);
885 }
886 eq_conds = vec![Some(new_cond)];
887 } else if expr.as_is_null().is_some() {
888 if !eq_conds.is_empty() && eq_conds.into_iter().all(|l| l.is_some()) {
889 return Ok(None);
890 }
891 eq_conds = vec![None];
892 } else if let Some((input_ref, in_const_list)) = expr
893 .as_in_const_list()
894 .or_else(|| expr.as_some_eq_const_list())
895 {
896 let mut scalars = HashSet::new();
897 for const_expr in in_const_list {
898 let new_expr =
899 if let Ok(expr) = const_expr.clone().cast_implicit(&input_ref.data_type) {
900 expr
901 } else {
902 match self::cast_compare::cast_compare_for_eq(
903 const_expr,
904 input_ref.data_type.clone(),
905 ) {
906 Ok(ResultForEq::Success(expr)) => expr,
907 Ok(ResultForEq::NeverEqual) => {
908 continue;
909 }
910 Err(_) => {
911 other_conds.push(expr);
912 continue 'group_loop;
913 }
914 }
915 };
916
917 let value = new_expr.fold_const()?;
918 let Some(value) = value else {
919 continue;
920 };
921 scalars.insert(Some(value));
922 }
923 if scalars.is_empty() {
924 return Ok(None);
926 }
927 if !eq_conds.is_empty() {
928 scalars = scalars
929 .intersection(&HashSet::from_iter(eq_conds))
930 .cloned()
931 .collect();
932 if scalars.is_empty() {
933 return Ok(None);
934 }
935 }
936 eq_conds = scalars
938 .into_iter()
939 .sorted_by(DefaultOrd::default_cmp)
940 .collect();
941 } else if let Some((input_ref, op, const_expr)) = expr.as_comparison_const() {
942 let new_expr =
943 if let Ok(expr) = const_expr.clone().cast_implicit(&input_ref.data_type) {
944 expr
945 } else {
946 match self::cast_compare::cast_compare_for_cmp(
947 const_expr,
948 input_ref.data_type,
949 op,
950 ) {
951 Ok(ResultForCmp::Success(expr)) => expr,
952 _ => {
953 other_conds.push(expr);
954 continue;
955 }
956 }
957 };
958 let Some(value) = new_expr.fold_const()? else {
959 return Ok(None);
961 };
962 match op {
963 ExprType::LessThan => {
964 upper_bound_conjunctions.push(Bound::Excluded(value));
965 }
966 ExprType::LessThanOrEqual => {
967 upper_bound_conjunctions.push(Bound::Included(value));
968 }
969 ExprType::GreaterThan => {
970 lower_bound_conjunctions.push(Bound::Excluded(value));
971 }
972 ExprType::GreaterThanOrEqual => {
973 lower_bound_conjunctions.push(Bound::Included(value));
974 }
975 _ => unreachable!(),
976 }
977 } else {
978 other_conds.push(expr);
979 }
980 }
981 Ok(Some((
982 lower_bound_conjunctions,
983 upper_bound_conjunctions,
984 eq_conds,
985 other_conds,
986 )))
987 }
988
989 fn mutual_exclusive_with_eq_conds(
990 new_conds: &ScalarImpl,
991 eq_conds: &[Option<ScalarImpl>],
992 ) -> bool {
993 !eq_conds.is_empty()
994 && eq_conds.iter().all(|l| {
995 if let Some(l) = l {
996 l != new_conds
997 } else {
998 true
999 }
1000 })
1001 }
1002
1003 fn merge_lower_bound_conjunctions(lb: Vec<Bound<ScalarImpl>>) -> Bound<ScalarImpl> {
1004 lb.into_iter()
1005 .max_by(|a, b| {
1006 match (a, b) {
1008 (Bound::Included(_), Bound::Unbounded) => std::cmp::Ordering::Greater,
1009 (Bound::Excluded(_), Bound::Unbounded) => std::cmp::Ordering::Greater,
1010 (Bound::Unbounded, Bound::Included(_)) => std::cmp::Ordering::Less,
1011 (Bound::Unbounded, Bound::Excluded(_)) => std::cmp::Ordering::Less,
1012 (Bound::Unbounded, Bound::Unbounded) => std::cmp::Ordering::Equal,
1013 (Bound::Included(a), Bound::Included(b)) => a.default_cmp(b),
1014 (Bound::Excluded(a), Bound::Excluded(b)) => a.default_cmp(b),
1015 (Bound::Included(a), Bound::Excluded(b)) => match a.default_cmp(b) {
1017 std::cmp::Ordering::Equal => std::cmp::Ordering::Less,
1018 other => other,
1019 },
1020 (Bound::Excluded(a), Bound::Included(b)) => match a.default_cmp(b) {
1021 std::cmp::Ordering::Equal => std::cmp::Ordering::Greater,
1022 other => other,
1023 },
1024 }
1025 })
1026 .unwrap_or(Bound::Unbounded)
1027 }
1028
1029 fn merge_upper_bound_conjunctions(ub: Vec<Bound<ScalarImpl>>) -> Bound<ScalarImpl> {
1030 ub.into_iter()
1031 .min_by(|a, b| {
1032 match (a, b) {
1034 (Bound::Included(_), Bound::Unbounded) => std::cmp::Ordering::Less,
1035 (Bound::Excluded(_), Bound::Unbounded) => std::cmp::Ordering::Less,
1036 (Bound::Unbounded, Bound::Included(_)) => std::cmp::Ordering::Greater,
1037 (Bound::Unbounded, Bound::Excluded(_)) => std::cmp::Ordering::Greater,
1038 (Bound::Unbounded, Bound::Unbounded) => std::cmp::Ordering::Equal,
1039 (Bound::Included(a), Bound::Included(b)) => a.default_cmp(b),
1040 (Bound::Excluded(a), Bound::Excluded(b)) => a.default_cmp(b),
1041 (Bound::Included(a), Bound::Excluded(b)) => match a.default_cmp(b) {
1043 std::cmp::Ordering::Equal => std::cmp::Ordering::Greater,
1044 other => other,
1045 },
1046 (Bound::Excluded(a), Bound::Included(b)) => match a.default_cmp(b) {
1047 std::cmp::Ordering::Equal => std::cmp::Ordering::Less,
1048 other => other,
1049 },
1050 }
1051 })
1052 .unwrap_or(Bound::Unbounded)
1053 }
1054
1055 fn is_invalid_range(lower_bound: &Bound<ScalarImpl>, upper_bound: &Bound<ScalarImpl>) -> bool {
1056 match (lower_bound, upper_bound) {
1057 (Bound::Included(l), Bound::Included(u)) => l.default_cmp(u).is_gt(), (Bound::Included(l), Bound::Excluded(u)) => l.default_cmp(u).is_ge(), (Bound::Excluded(l), Bound::Included(u)) => l.default_cmp(u).is_ge(), (Bound::Excluded(l), Bound::Excluded(u)) => l.default_cmp(u).is_ge(), _ => false,
1062 }
1063 }
1064
1065 fn extract_eq_conds_within_range(
1066 eq_conds: Vec<Option<ScalarImpl>>,
1067 upper_bound: &Bound<ScalarImpl>,
1068 lower_bound: &Bound<ScalarImpl>,
1069 ) -> Vec<Option<ScalarImpl>> {
1070 if Self::is_invalid_range(lower_bound, upper_bound) {
1073 return vec![];
1074 }
1075
1076 let is_extract_null = upper_bound == &Bound::Unbounded && lower_bound == &Bound::Unbounded;
1077
1078 eq_conds
1079 .into_iter()
1080 .filter(|cond| {
1081 if let Some(cond) = cond {
1082 match lower_bound {
1083 Bound::Included(val) => {
1084 if cond.default_cmp(val).is_lt() {
1085 return false;
1087 }
1088 }
1089 Bound::Excluded(val) => {
1090 if cond.default_cmp(val).is_le() {
1091 return false;
1093 }
1094 }
1095 Bound::Unbounded => {}
1096 }
1097 match upper_bound {
1098 Bound::Included(val) => {
1099 if cond.default_cmp(val).is_gt() {
1100 return false;
1102 }
1103 }
1104 Bound::Excluded(val) => {
1105 if cond.default_cmp(val).is_ge() {
1106 return false;
1108 }
1109 }
1110 Bound::Unbounded => {}
1111 }
1112 true
1113 } else {
1114 is_extract_null
1115 }
1116 })
1117 .collect()
1118 }
1119
1120 #[must_use]
1126 pub fn group_by<F, const N: usize>(self, f: F) -> [Self; N]
1127 where
1128 F: Fn(&ExprImpl) -> usize,
1129 {
1130 const EMPTY: Vec<ExprImpl> = vec![];
1131 let mut groups = [EMPTY; N];
1132 for (key, group) in &self.conjunctions.into_iter().chunk_by(|expr| {
1133 let i = f(expr);
1135 assert!(i < N);
1136 i
1137 }) {
1138 groups[key].extend(group);
1139 }
1140
1141 groups.map(|group| Condition {
1142 conjunctions: group,
1143 })
1144 }
1145
1146 #[must_use]
1147 pub fn rewrite_expr(self, rewriter: &mut (impl ExprRewriter + ?Sized)) -> Self {
1148 Self {
1149 conjunctions: self
1150 .conjunctions
1151 .into_iter()
1152 .map(|expr| rewriter.rewrite_expr(expr))
1153 .collect(),
1154 }
1155 .simplify()
1156 }
1157
1158 pub fn visit_expr<V: ExprVisitor + ?Sized>(&self, visitor: &mut V) {
1159 self.conjunctions
1160 .iter()
1161 .for_each(|expr| visitor.visit_expr(expr));
1162 }
1163
1164 pub fn visit_expr_mut(&mut self, mutator: &mut (impl ExprMutator + ?Sized)) {
1165 self.conjunctions
1166 .iter_mut()
1167 .for_each(|expr| mutator.visit_expr(expr))
1168 }
1169
1170 fn simplify(self) -> Self {
1173 let conjunctions: Vec<_> = self
1175 .conjunctions
1176 .into_iter()
1177 .map(push_down_not)
1178 .map(fold_boolean_constant)
1179 .map(column_self_eq_eliminate)
1180 .flat_map(to_conjunctions)
1181 .collect();
1182 let mut res: Vec<ExprImpl> = Vec::new();
1183 let mut visited: HashSet<ExprImpl> = HashSet::new();
1184 for expr in conjunctions {
1185 if !expr.has_subquery() {
1187 let results_of_factorization = factorization_expr(expr);
1188 res.extend(
1189 results_of_factorization
1190 .clone()
1191 .into_iter()
1192 .filter(|expr| !visited.contains(expr)),
1193 );
1194 visited.extend(results_of_factorization);
1195 } else {
1196 res.push(expr);
1198 }
1199 }
1200 res.retain(|expr| {
1202 if let Some(v) = try_get_bool_constant(expr)
1203 && v
1204 {
1205 false
1206 } else {
1207 true
1208 }
1209 });
1210 for expr in &mut res {
1212 if let Some(v) = try_get_bool_constant(expr)
1213 && !v
1214 {
1215 res.clear();
1216 res.push(ExprImpl::literal_bool(false));
1217 break;
1218 }
1219 }
1220 Self { conjunctions: res }
1221 }
1222}
1223
1224pub struct ConditionDisplay<'a> {
1225 pub condition: &'a Condition,
1226 pub input_schema: &'a Schema,
1227}
1228
1229impl ConditionDisplay<'_> {
1230 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1231 if self.condition.always_true() {
1232 write!(f, "true")
1233 } else {
1234 write!(
1235 f,
1236 "{}",
1237 self.condition
1238 .conjunctions
1239 .iter()
1240 .format_with(" AND ", |expr, f| {
1241 f(&ExprDisplay {
1242 expr,
1243 input_schema: self.input_schema,
1244 })
1245 })
1246 )
1247 }
1248 }
1249}
1250
1251impl fmt::Display for ConditionDisplay<'_> {
1252 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1253 self.fmt(f)
1254 }
1255}
1256
1257impl fmt::Debug for ConditionDisplay<'_> {
1258 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1259 self.fmt(f)
1260 }
1261}
1262
1263mod cast_compare {
1268 use risingwave_common::types::DataType;
1269
1270 use crate::expr::{Expr, ExprImpl, ExprType};
1271
1272 enum ShrinkResult {
1273 OutUpperBound,
1274 OutLowerBound,
1275 InRange(ExprImpl),
1276 }
1277
1278 pub enum ResultForEq {
1279 Success(ExprImpl),
1280 NeverEqual,
1281 }
1282
1283 pub enum ResultForCmp {
1284 Success(ExprImpl),
1285 OutUpperBound,
1286 OutLowerBound,
1287 }
1288
1289 pub fn cast_compare_for_eq(const_expr: ExprImpl, target: DataType) -> Result<ResultForEq, ()> {
1290 match (const_expr.return_type(), &target) {
1291 (DataType::Int64, DataType::Int32)
1292 | (DataType::Int64, DataType::Int16)
1293 | (DataType::Int32, DataType::Int16) => match shrink_integral(const_expr, target)? {
1294 ShrinkResult::InRange(expr) => Ok(ResultForEq::Success(expr)),
1295 ShrinkResult::OutUpperBound | ShrinkResult::OutLowerBound => {
1296 Ok(ResultForEq::NeverEqual)
1297 }
1298 },
1299 _ => Err(()),
1300 }
1301 }
1302
1303 pub fn cast_compare_for_cmp(
1304 const_expr: ExprImpl,
1305 target: DataType,
1306 _op: ExprType,
1307 ) -> Result<ResultForCmp, ()> {
1308 match (const_expr.return_type(), &target) {
1309 (DataType::Int64, DataType::Int32)
1310 | (DataType::Int64, DataType::Int16)
1311 | (DataType::Int32, DataType::Int16) => match shrink_integral(const_expr, target)? {
1312 ShrinkResult::InRange(expr) => Ok(ResultForCmp::Success(expr)),
1313 ShrinkResult::OutUpperBound => Ok(ResultForCmp::OutUpperBound),
1314 ShrinkResult::OutLowerBound => Ok(ResultForCmp::OutLowerBound),
1315 },
1316 _ => Err(()),
1317 }
1318 }
1319
1320 fn shrink_integral(const_expr: ExprImpl, target: DataType) -> Result<ShrinkResult, ()> {
1321 let (upper_bound, lower_bound) = match (const_expr.return_type(), &target) {
1322 (DataType::Int64, DataType::Int32) => (i32::MAX as i64, i32::MIN as i64),
1323 (DataType::Int64, DataType::Int16) | (DataType::Int32, DataType::Int16) => {
1324 (i16::MAX as i64, i16::MIN as i64)
1325 }
1326 _ => unreachable!(),
1327 };
1328 match const_expr.fold_const().map_err(|_| ())? {
1329 Some(scalar) => {
1330 let value = scalar.as_integral();
1331 if value > upper_bound {
1332 Ok(ShrinkResult::OutUpperBound)
1333 } else if value < lower_bound {
1334 Ok(ShrinkResult::OutLowerBound)
1335 } else {
1336 Ok(ShrinkResult::InRange(
1337 const_expr.cast_explicit(&target).unwrap(),
1338 ))
1339 }
1340 }
1341 None => Ok(ShrinkResult::InRange(
1342 const_expr.cast_explicit(&target).unwrap(),
1343 )),
1344 }
1345 }
1346}
1347
1348#[cfg(test)]
1349mod tests {
1350 use rand::Rng;
1351
1352 use super::*;
1353
1354 #[test]
1355 fn test_split() {
1356 let left_col_num = 3;
1357 let right_col_num = 2;
1358
1359 let ty = DataType::Int32;
1360
1361 let mut rng = rand::rng();
1362
1363 let left: ExprImpl = FunctionCall::new(
1364 ExprType::LessThanOrEqual,
1365 vec![
1366 InputRef::new(rng.random_range(0..left_col_num), ty.clone()).into(),
1367 InputRef::new(rng.random_range(0..left_col_num), ty.clone()).into(),
1368 ],
1369 )
1370 .unwrap()
1371 .into();
1372
1373 let right: ExprImpl = FunctionCall::new(
1374 ExprType::LessThan,
1375 vec![
1376 InputRef::new(
1377 rng.random_range(left_col_num..left_col_num + right_col_num),
1378 ty.clone(),
1379 )
1380 .into(),
1381 InputRef::new(
1382 rng.random_range(left_col_num..left_col_num + right_col_num),
1383 ty.clone(),
1384 )
1385 .into(),
1386 ],
1387 )
1388 .unwrap()
1389 .into();
1390
1391 let other: ExprImpl = FunctionCall::new(
1392 ExprType::GreaterThan,
1393 vec![
1394 InputRef::new(rng.random_range(0..left_col_num), ty.clone()).into(),
1395 InputRef::new(
1396 rng.random_range(left_col_num..left_col_num + right_col_num),
1397 ty,
1398 )
1399 .into(),
1400 ],
1401 )
1402 .unwrap()
1403 .into();
1404
1405 let cond = Condition::with_expr(other.clone())
1406 .and(Condition::with_expr(right.clone()))
1407 .and(Condition::with_expr(left.clone()));
1408
1409 let res = cond.split(left_col_num, right_col_num);
1410
1411 assert_eq!(res.0.conjunctions, vec![left]);
1412 assert_eq!(res.1.conjunctions, vec![right]);
1413 assert_eq!(res.2.conjunctions, vec![other]);
1414 }
1415
1416 #[test]
1417 fn test_self_eq_eliminate() {
1418 let left_col_num = 3;
1419 let right_col_num = 2;
1420
1421 let ty = DataType::Int32;
1422
1423 let mut rng = rand::rng();
1424
1425 let x: ExprImpl = InputRef::new(rng.random_range(0..left_col_num), ty.clone()).into();
1426
1427 let left: ExprImpl = FunctionCall::new(ExprType::Equal, vec![x.clone(), x.clone()])
1428 .unwrap()
1429 .into();
1430
1431 let right: ExprImpl = FunctionCall::new(
1432 ExprType::LessThan,
1433 vec![
1434 InputRef::new(
1435 rng.random_range(left_col_num..left_col_num + right_col_num),
1436 ty.clone(),
1437 )
1438 .into(),
1439 InputRef::new(
1440 rng.random_range(left_col_num..left_col_num + right_col_num),
1441 ty.clone(),
1442 )
1443 .into(),
1444 ],
1445 )
1446 .unwrap()
1447 .into();
1448
1449 let other: ExprImpl = FunctionCall::new(
1450 ExprType::GreaterThan,
1451 vec![
1452 InputRef::new(rng.random_range(0..left_col_num), ty.clone()).into(),
1453 InputRef::new(
1454 rng.random_range(left_col_num..left_col_num + right_col_num),
1455 ty,
1456 )
1457 .into(),
1458 ],
1459 )
1460 .unwrap()
1461 .into();
1462
1463 let cond = Condition::with_expr(other.clone())
1464 .and(Condition::with_expr(right.clone()))
1465 .and(Condition::with_expr(left));
1466
1467 let res = cond.split(left_col_num, right_col_num);
1468
1469 let left_res = FunctionCall::new(ExprType::IsNotNull, vec![x])
1470 .unwrap()
1471 .into();
1472
1473 assert_eq!(res.0.conjunctions, vec![left_res]);
1474 assert_eq!(res.1.conjunctions, vec![right]);
1475 assert_eq!(res.2.conjunctions, vec![other]);
1476 }
1477}