1use std::{
5 ops::Bound,
6 sync::{Arc, LazyLock},
7};
8
9use arrow::array::BinaryBuilder;
10use arrow_array::{Array, RecordBatch, UInt32Array};
11use arrow_schema::{DataType, Field, Schema, SchemaRef};
12use async_recursion::async_recursion;
13use async_trait::async_trait;
14use datafusion_common::ScalarValue;
15use datafusion_expr::{
16 Between, BinaryExpr, Expr, Operator, ReturnFieldArgs, ScalarUDF,
17 expr::{InList, Like, ScalarFunction},
18};
19use tokio::try_join;
20
21use super::{
22 AnyQuery, BloomFilterQuery, LabelListQuery, MetricsCollector, SargableQuery, ScalarIndex,
23 SearchResult, TextQuery, TokenQuery,
24};
25#[cfg(feature = "geo")]
26use super::{GeoQuery, RelationQuery};
27use lance_core::{
28 Error, Result,
29 utils::mask::{NullableRowAddrMask, RowAddrMask},
30};
31use lance_datafusion::{expr::safe_coerce_scalar, planner::Planner};
32use roaring::RoaringBitmap;
33use tracing::instrument;
34
35const MAX_DEPTH: usize = 500;
36
37#[derive(Debug, PartialEq)]
65pub struct IndexedExpression {
66 pub scalar_query: Option<ScalarIndexExpr>,
68 pub refine_expr: Option<Expr>,
70}
71
72pub trait ScalarQueryParser: std::fmt::Debug + Send + Sync {
73 fn visit_between(
77 &self,
78 column: &str,
79 low: &Bound<ScalarValue>,
80 high: &Bound<ScalarValue>,
81 ) -> Option<IndexedExpression>;
82 fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression>;
86 fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression>;
90 fn visit_is_null(&self, column: &str) -> Option<IndexedExpression>;
94 fn visit_comparison(
98 &self,
99 column: &str,
100 value: &ScalarValue,
101 op: &Operator,
102 ) -> Option<IndexedExpression>;
103 fn visit_scalar_function(
108 &self,
109 column: &str,
110 data_type: &DataType,
111 func: &ScalarUDF,
112 args: &[Expr],
113 ) -> Option<IndexedExpression>;
114
115 fn visit_like(
130 &self,
131 _column: &str,
132 _like: &Like,
133 _pattern: &ScalarValue,
134 ) -> Option<IndexedExpression> {
135 None
136 }
137
138 fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
162 match func {
163 Expr::Column(_) => Some(data_type.clone()),
164 _ => None,
165 }
166 }
167}
168
169#[derive(Debug)]
173pub struct MultiQueryParser {
174 parsers: Vec<Box<dyn ScalarQueryParser>>,
175}
176
177impl MultiQueryParser {
178 pub fn single(parser: Box<dyn ScalarQueryParser>) -> Self {
180 Self {
181 parsers: vec![parser],
182 }
183 }
184
185 pub fn add(&mut self, other: Box<dyn ScalarQueryParser>) {
187 self.parsers.push(other);
188 }
189}
190
191impl ScalarQueryParser for MultiQueryParser {
192 fn visit_between(
193 &self,
194 column: &str,
195 low: &Bound<ScalarValue>,
196 high: &Bound<ScalarValue>,
197 ) -> Option<IndexedExpression> {
198 self.parsers
199 .iter()
200 .find_map(|parser| parser.visit_between(column, low, high))
201 }
202 fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
203 self.parsers
204 .iter()
205 .find_map(|parser| parser.visit_in_list(column, in_list))
206 }
207 fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
208 self.parsers
209 .iter()
210 .find_map(|parser| parser.visit_is_bool(column, value))
211 }
212 fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
213 self.parsers
214 .iter()
215 .find_map(|parser| parser.visit_is_null(column))
216 }
217 fn visit_comparison(
218 &self,
219 column: &str,
220 value: &ScalarValue,
221 op: &Operator,
222 ) -> Option<IndexedExpression> {
223 self.parsers
224 .iter()
225 .find_map(|parser| parser.visit_comparison(column, value, op))
226 }
227 fn visit_scalar_function(
228 &self,
229 column: &str,
230 data_type: &DataType,
231 func: &ScalarUDF,
232 args: &[Expr],
233 ) -> Option<IndexedExpression> {
234 self.parsers
235 .iter()
236 .find_map(|parser| parser.visit_scalar_function(column, data_type, func, args))
237 }
238 fn visit_like(
239 &self,
240 column: &str,
241 like: &Like,
242 pattern: &ScalarValue,
243 ) -> Option<IndexedExpression> {
244 self.parsers
245 .iter()
246 .find_map(|parser| parser.visit_like(column, like, pattern))
247 }
248 fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
255 self.parsers
256 .iter()
257 .find_map(|parser| parser.is_valid_reference(func, data_type))
258 }
259}
260
261#[derive(Debug)]
263pub struct SargableQueryParser {
264 index_name: String,
265 needs_recheck: bool,
266}
267
268impl SargableQueryParser {
269 pub fn new(index_name: String, needs_recheck: bool) -> Self {
270 Self {
271 index_name,
272 needs_recheck,
273 }
274 }
275}
276
277impl ScalarQueryParser for SargableQueryParser {
278 fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
279 match func {
280 Expr::Column(_) => Some(data_type.clone()),
281 Expr::ScalarFunction(udf) if udf.name() == "get_field" => Some(data_type.clone()),
283 _ => None,
284 }
285 }
286
287 fn visit_between(
288 &self,
289 column: &str,
290 low: &Bound<ScalarValue>,
291 high: &Bound<ScalarValue>,
292 ) -> Option<IndexedExpression> {
293 if let Bound::Included(val) | Bound::Excluded(val) = low
294 && val.is_null()
295 {
296 return None;
297 }
298 if let Bound::Included(val) | Bound::Excluded(val) = high
299 && val.is_null()
300 {
301 return None;
302 }
303 let query = SargableQuery::Range(low.clone(), high.clone());
304 Some(IndexedExpression::index_query_with_recheck(
305 column.to_string(),
306 self.index_name.clone(),
307 Arc::new(query),
308 self.needs_recheck,
309 ))
310 }
311
312 fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
313 if in_list.iter().any(|val| val.is_null()) {
314 return None;
315 }
316 let query = SargableQuery::IsIn(in_list.to_vec());
317 Some(IndexedExpression::index_query_with_recheck(
318 column.to_string(),
319 self.index_name.clone(),
320 Arc::new(query),
321 self.needs_recheck,
322 ))
323 }
324
325 fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
326 Some(IndexedExpression::index_query_with_recheck(
327 column.to_string(),
328 self.index_name.clone(),
329 Arc::new(SargableQuery::Equals(ScalarValue::Boolean(Some(value)))),
330 self.needs_recheck,
331 ))
332 }
333
334 fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
335 Some(IndexedExpression::index_query_with_recheck(
336 column.to_string(),
337 self.index_name.clone(),
338 Arc::new(SargableQuery::IsNull()),
339 self.needs_recheck,
340 ))
341 }
342
343 fn visit_comparison(
344 &self,
345 column: &str,
346 value: &ScalarValue,
347 op: &Operator,
348 ) -> Option<IndexedExpression> {
349 if value.is_null() {
350 return None;
351 }
352 let query = match op {
353 Operator::Lt => SargableQuery::Range(Bound::Unbounded, Bound::Excluded(value.clone())),
354 Operator::LtEq => {
355 SargableQuery::Range(Bound::Unbounded, Bound::Included(value.clone()))
356 }
357 Operator::Gt => SargableQuery::Range(Bound::Excluded(value.clone()), Bound::Unbounded),
358 Operator::GtEq => {
359 SargableQuery::Range(Bound::Included(value.clone()), Bound::Unbounded)
360 }
361 Operator::Eq => SargableQuery::Equals(value.clone()),
362 Operator::NotEq => SargableQuery::Equals(value.clone()),
364 _ => unreachable!(),
365 };
366 Some(IndexedExpression::index_query_with_recheck(
367 column.to_string(),
368 self.index_name.clone(),
369 Arc::new(query),
370 self.needs_recheck,
371 ))
372 }
373
374 fn visit_scalar_function(
375 &self,
376 column: &str,
377 _data_type: &DataType,
378 func: &ScalarUDF,
379 args: &[Expr],
380 ) -> Option<IndexedExpression> {
381 if func.name() == "starts_with" && args.len() == 2 {
383 let prefix = match &args[1] {
385 Expr::Literal(ScalarValue::Utf8(Some(s)), _) => ScalarValue::Utf8(Some(s.clone())),
386 Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => {
387 ScalarValue::LargeUtf8(Some(s.clone()))
388 }
389 _ => return None,
390 };
391
392 let query = SargableQuery::LikePrefix(prefix);
393 return Some(IndexedExpression::index_query_with_recheck(
394 column.to_string(),
395 self.index_name.clone(),
396 Arc::new(query),
397 self.needs_recheck,
398 ));
399 }
400
401 None
402 }
403
404 fn visit_like(
405 &self,
406 column: &str,
407 like: &Like,
408 pattern: &ScalarValue,
409 ) -> Option<IndexedExpression> {
410 if like.case_insensitive {
412 return None;
413 }
414
415 let pattern_str = match pattern {
417 ScalarValue::Utf8(Some(s)) => s.as_str(),
418 ScalarValue::LargeUtf8(Some(s)) => s.as_str(),
419 _ => return None,
420 };
421
422 let (prefix, needs_refine) = extract_like_leading_prefix(pattern_str, like.escape_char)?;
424
425 let prefix_value = match pattern {
427 ScalarValue::Utf8(_) => ScalarValue::Utf8(Some(prefix)),
428 ScalarValue::LargeUtf8(_) => ScalarValue::LargeUtf8(Some(prefix)),
429 _ => return None,
430 };
431
432 let query = SargableQuery::LikePrefix(prefix_value);
433 let scalar_query = Some(ScalarIndexExpr::Query(ScalarIndexSearch {
434 column: column.to_string(),
435 index_name: self.index_name.clone(),
436 query: Arc::new(query),
437 needs_recheck: self.needs_recheck,
438 }));
439
440 let refine_expr = if needs_refine {
442 Some(Expr::Like(like.clone()))
443 } else {
444 None
445 };
446
447 Some(IndexedExpression {
448 scalar_query,
449 refine_expr,
450 })
451 }
452}
453
454fn extract_like_leading_prefix(pattern: &str, escape_char: Option<char>) -> Option<(String, bool)> {
470 let chars: Vec<char> = pattern.chars().collect();
471 let len = chars.len();
472
473 if len == 0 {
474 return None;
475 }
476
477 let effective_escape_char = escape_char.or(Some('\\'));
482
483 let is_escaped = |i: usize| -> bool {
485 if let Some(esc) = effective_escape_char {
486 if i > 0 && chars[i - 1] == esc {
487 if i >= 2 && chars[i - 2] == esc {
489 false } else {
491 true }
493 } else {
494 false
495 }
496 } else {
497 false
499 }
500 };
501
502 let has_wildcard = chars.iter().enumerate().any(|(i, &c)| {
504 if c != '%' && c != '_' {
505 return false;
506 }
507 !is_escaped(i)
508 });
509
510 if !has_wildcard {
511 return None; }
513
514 if chars[0] == '%' || chars[0] == '_' {
516 return None; }
518
519 let mut prefix = String::new();
521 let mut i = 0;
522 let mut found_wildcard = false;
523
524 while i < len {
525 let c = chars[i];
526
527 if let Some(esc) = effective_escape_char
529 && c == esc
530 && i + 1 < len
531 {
532 let next = chars[i + 1];
533 if next == '%' || next == '_' || next == esc {
534 prefix.push(next);
536 i += 2;
537 continue;
538 }
539 }
540
541 if c == '%' || c == '_' {
543 found_wildcard = true;
544 break;
545 }
546
547 prefix.push(c);
548 i += 1;
549 }
550
551 if prefix.is_empty() {
552 return None;
553 }
554
555 let needs_refine = if found_wildcard && i < len {
557 if chars[i] == '%' && i + 1 == len {
559 false
561 } else {
562 true
564 }
565 } else {
566 false
568 };
569
570 Some((prefix, needs_refine))
571}
572
573#[derive(Debug)]
575pub struct BloomFilterQueryParser {
576 index_name: String,
577 needs_recheck: bool,
578}
579
580impl BloomFilterQueryParser {
581 pub fn new(index_name: String, needs_recheck: bool) -> Self {
582 Self {
583 index_name,
584 needs_recheck,
585 }
586 }
587}
588
589impl ScalarQueryParser for BloomFilterQueryParser {
590 fn visit_between(
591 &self,
592 _: &str,
593 _: &Bound<ScalarValue>,
594 _: &Bound<ScalarValue>,
595 ) -> Option<IndexedExpression> {
596 None
598 }
599
600 fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
601 let query = BloomFilterQuery::IsIn(in_list.to_vec());
602 Some(IndexedExpression::index_query_with_recheck(
603 column.to_string(),
604 self.index_name.clone(),
605 Arc::new(query),
606 self.needs_recheck,
607 ))
608 }
609
610 fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
611 Some(IndexedExpression::index_query_with_recheck(
612 column.to_string(),
613 self.index_name.clone(),
614 Arc::new(BloomFilterQuery::Equals(ScalarValue::Boolean(Some(value)))),
615 self.needs_recheck,
616 ))
617 }
618
619 fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
620 Some(IndexedExpression::index_query_with_recheck(
621 column.to_string(),
622 self.index_name.clone(),
623 Arc::new(BloomFilterQuery::IsNull()),
624 self.needs_recheck,
625 ))
626 }
627
628 fn visit_comparison(
629 &self,
630 column: &str,
631 value: &ScalarValue,
632 op: &Operator,
633 ) -> Option<IndexedExpression> {
634 let query = match op {
635 Operator::Eq => BloomFilterQuery::Equals(value.clone()),
637 Operator::NotEq => BloomFilterQuery::Equals(value.clone()),
639 _ => return None,
641 };
642 Some(IndexedExpression::index_query_with_recheck(
643 column.to_string(),
644 self.index_name.clone(),
645 Arc::new(query),
646 self.needs_recheck,
647 ))
648 }
649
650 fn visit_scalar_function(
651 &self,
652 _: &str,
653 _: &DataType,
654 _: &ScalarUDF,
655 _: &[Expr],
656 ) -> Option<IndexedExpression> {
657 None
659 }
660}
661
662#[derive(Debug)]
664pub struct LabelListQueryParser {
665 index_name: String,
666}
667
668impl LabelListQueryParser {
669 pub fn new(index_name: String) -> Self {
670 Self { index_name }
671 }
672}
673
674impl ScalarQueryParser for LabelListQueryParser {
675 fn visit_between(
676 &self,
677 _: &str,
678 _: &Bound<ScalarValue>,
679 _: &Bound<ScalarValue>,
680 ) -> Option<IndexedExpression> {
681 None
682 }
683
684 fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
685 None
686 }
687
688 fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
689 None
690 }
691
692 fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
693 None
694 }
695
696 fn visit_comparison(
697 &self,
698 _: &str,
699 _: &ScalarValue,
700 _: &Operator,
701 ) -> Option<IndexedExpression> {
702 None
703 }
704
705 fn visit_scalar_function(
706 &self,
707 column: &str,
708 data_type: &DataType,
709 func: &ScalarUDF,
710 args: &[Expr],
711 ) -> Option<IndexedExpression> {
712 if args.len() != 2 {
713 return None;
714 }
715 if func.name() == "array_has" {
717 let inner_type = match data_type {
718 DataType::List(field) | DataType::LargeList(field) => field.data_type(),
719 _ => return None,
720 };
721 let scalar = maybe_scalar(&args[1], inner_type)?;
722 if scalar.is_null() {
725 return None;
726 }
727 let query = LabelListQuery::HasAnyLabel(vec![scalar]);
728 return Some(IndexedExpression::index_query(
729 column.to_string(),
730 self.index_name.clone(),
731 Arc::new(query),
732 ));
733 }
734
735 let label_list = maybe_scalar(&args[1], data_type)?;
736 if let ScalarValue::List(list_arr) = label_list {
737 let list_values = list_arr.values();
738 if list_values.is_empty() {
739 return None;
740 }
741 let mut scalars = Vec::with_capacity(list_values.len());
742 for idx in 0..list_values.len() {
743 scalars.push(ScalarValue::try_from_array(list_values.as_ref(), idx).ok()?);
744 }
745 if func.name() == "array_has_all" {
746 let query = LabelListQuery::HasAllLabels(scalars);
747 Some(IndexedExpression::index_query(
748 column.to_string(),
749 self.index_name.clone(),
750 Arc::new(query),
751 ))
752 } else if func.name() == "array_has_any" {
753 let query = LabelListQuery::HasAnyLabel(scalars);
754 Some(IndexedExpression::index_query(
755 column.to_string(),
756 self.index_name.clone(),
757 Arc::new(query),
758 ))
759 } else {
760 None
761 }
762 } else {
763 None
764 }
765 }
766}
767
768#[derive(Debug, Clone)]
770pub struct TextQueryParser {
771 index_name: String,
772 needs_recheck: bool,
773}
774
775impl TextQueryParser {
776 pub fn new(index_name: String, needs_recheck: bool) -> Self {
777 Self {
778 index_name,
779 needs_recheck,
780 }
781 }
782}
783
784impl ScalarQueryParser for TextQueryParser {
785 fn visit_between(
786 &self,
787 _: &str,
788 _: &Bound<ScalarValue>,
789 _: &Bound<ScalarValue>,
790 ) -> Option<IndexedExpression> {
791 None
792 }
793
794 fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
795 None
796 }
797
798 fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
799 None
800 }
801
802 fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
803 None
804 }
805
806 fn visit_comparison(
807 &self,
808 _: &str,
809 _: &ScalarValue,
810 _: &Operator,
811 ) -> Option<IndexedExpression> {
812 None
813 }
814
815 fn visit_scalar_function(
816 &self,
817 column: &str,
818 data_type: &DataType,
819 func: &ScalarUDF,
820 args: &[Expr],
821 ) -> Option<IndexedExpression> {
822 if args.len() != 2 {
823 return None;
824 }
825 let scalar = maybe_scalar(&args[1], data_type)?;
826 match scalar {
827 ScalarValue::Utf8(Some(scalar_str)) | ScalarValue::LargeUtf8(Some(scalar_str)) => {
828 if func.name() == "contains" {
829 let query = TextQuery::StringContains(scalar_str);
830 Some(IndexedExpression::index_query_with_recheck(
831 column.to_string(),
832 self.index_name.clone(),
833 Arc::new(query),
834 self.needs_recheck,
835 ))
836 } else {
837 None
838 }
839 }
840 _ => {
841 None
843 }
844 }
845 }
846}
847
848#[derive(Debug, Clone)]
850pub struct FtsQueryParser {
851 index_name: String,
852}
853
854impl FtsQueryParser {
855 pub fn new(name: String) -> Self {
856 Self { index_name: name }
857 }
858}
859
860impl ScalarQueryParser for FtsQueryParser {
861 fn visit_between(
862 &self,
863 _: &str,
864 _: &Bound<ScalarValue>,
865 _: &Bound<ScalarValue>,
866 ) -> Option<IndexedExpression> {
867 None
868 }
869
870 fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
871 None
872 }
873
874 fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
875 None
876 }
877
878 fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
879 None
880 }
881
882 fn visit_comparison(
883 &self,
884 _: &str,
885 _: &ScalarValue,
886 _: &Operator,
887 ) -> Option<IndexedExpression> {
888 None
889 }
890
891 fn visit_scalar_function(
892 &self,
893 column: &str,
894 data_type: &DataType,
895 func: &ScalarUDF,
896 args: &[Expr],
897 ) -> Option<IndexedExpression> {
898 if args.len() != 2 {
899 return None;
900 }
901 let scalar = maybe_scalar(&args[1], data_type)?;
902 if let ScalarValue::Utf8(Some(scalar_str)) = scalar
903 && func.name() == "contains_tokens"
904 {
905 let query = TokenQuery::TokensContains(scalar_str);
906 return Some(IndexedExpression::index_query(
907 column.to_string(),
908 self.index_name.clone(),
909 Arc::new(query),
910 ));
911 }
912 None
913 }
914}
915
916#[cfg(feature = "geo")]
918#[derive(Debug, Clone)]
919pub struct GeoQueryParser {
920 index_name: String,
921}
922
923#[cfg(feature = "geo")]
924impl GeoQueryParser {
925 pub fn new(index_name: String) -> Self {
926 Self { index_name }
927 }
928}
929
930#[cfg(feature = "geo")]
931impl ScalarQueryParser for GeoQueryParser {
932 fn visit_between(
933 &self,
934 _: &str,
935 _: &Bound<ScalarValue>,
936 _: &Bound<ScalarValue>,
937 ) -> Option<IndexedExpression> {
938 None
939 }
940
941 fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
942 None
943 }
944
945 fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
946 None
947 }
948
949 fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
950 Some(IndexedExpression::index_query_with_recheck(
951 column.to_string(),
952 self.index_name.clone(),
953 Arc::new(GeoQuery::IsNull),
954 true,
955 ))
956 }
957
958 fn visit_comparison(
959 &self,
960 _: &str,
961 _: &ScalarValue,
962 _: &Operator,
963 ) -> Option<IndexedExpression> {
964 None
965 }
966
967 fn visit_scalar_function(
968 &self,
969 column: &str,
970 _data_type: &DataType,
971 func: &ScalarUDF,
972 args: &[Expr],
973 ) -> Option<IndexedExpression> {
974 if (func.name() == "st_intersects"
975 || func.name() == "st_contains"
976 || func.name() == "st_within"
977 || func.name() == "st_touches"
978 || func.name() == "st_crosses"
979 || func.name() == "st_overlaps"
980 || func.name() == "st_covers"
981 || func.name() == "st_coveredby")
982 && args.len() == 2
983 {
984 let left_arg = &args[0];
985 let right_arg = &args[1];
986 return match (left_arg, right_arg) {
987 (Expr::Literal(left_value, metadata), Expr::Column(_)) => {
988 let mut field = Field::new("_geo", left_value.data_type(), false);
989 if let Some(metadata) = metadata {
990 field = field.with_metadata(metadata.to_hashmap());
991 }
992 let query = GeoQuery::IntersectQuery(RelationQuery {
993 value: left_value.clone(),
994 field,
995 });
996 Some(IndexedExpression::index_query_with_recheck(
997 column.to_string(),
998 self.index_name.clone(),
999 Arc::new(query),
1000 true,
1001 ))
1002 }
1003 (Expr::Column(_), Expr::Literal(right_value, metadata)) => {
1004 let mut field = Field::new("_geo", right_value.data_type(), false);
1005 if let Some(metadata) = metadata {
1006 field = field.with_metadata(metadata.to_hashmap());
1007 }
1008 let query = GeoQuery::IntersectQuery(RelationQuery {
1009 value: right_value.clone(),
1010 field,
1011 });
1012 Some(IndexedExpression::index_query_with_recheck(
1013 column.to_string(),
1014 self.index_name.clone(),
1015 Arc::new(query),
1016 true,
1017 ))
1018 }
1019 _ => None,
1020 };
1021 }
1022 None
1023 }
1024}
1025
1026impl IndexedExpression {
1027 fn refine_only(refine_expr: Expr) -> Self {
1029 Self {
1030 scalar_query: None,
1031 refine_expr: Some(refine_expr),
1032 }
1033 }
1034
1035 fn index_query(column: String, index_name: String, query: Arc<dyn AnyQuery>) -> Self {
1037 Self {
1038 scalar_query: Some(ScalarIndexExpr::Query(ScalarIndexSearch {
1039 column,
1040 index_name,
1041 query,
1042 needs_recheck: false, })),
1044 refine_expr: None,
1045 }
1046 }
1047
1048 fn index_query_with_recheck(
1050 column: String,
1051 index_name: String,
1052 query: Arc<dyn AnyQuery>,
1053 needs_recheck: bool,
1054 ) -> Self {
1055 Self {
1056 scalar_query: Some(ScalarIndexExpr::Query(ScalarIndexSearch {
1057 column,
1058 index_name,
1059 query,
1060 needs_recheck,
1061 })),
1062 refine_expr: None,
1063 }
1064 }
1065
1066 fn maybe_not(self) -> Option<Self> {
1071 match (self.scalar_query, self.refine_expr) {
1072 (Some(_), Some(_)) => None,
1073 (Some(scalar_query), None) => {
1074 if scalar_query.needs_recheck() {
1075 return None;
1076 }
1077 Some(Self {
1078 scalar_query: Some(ScalarIndexExpr::Not(Box::new(scalar_query))),
1079 refine_expr: None,
1080 })
1081 }
1082 (None, Some(refine_expr)) => Some(Self {
1083 scalar_query: None,
1084 refine_expr: Some(Expr::Not(Box::new(refine_expr))),
1085 }),
1086 (None, None) => panic!("Empty node should not occur"),
1087 }
1088 }
1089
1090 fn and(self, other: Self) -> Self {
1095 let scalar_query = match (self.scalar_query, other.scalar_query) {
1096 (Some(scalar_query), Some(other_scalar_query)) => Some(ScalarIndexExpr::And(
1097 Box::new(scalar_query),
1098 Box::new(other_scalar_query),
1099 )),
1100 (Some(scalar_query), None) => Some(scalar_query),
1101 (None, Some(scalar_query)) => Some(scalar_query),
1102 (None, None) => None,
1103 };
1104 let refine_expr = match (self.refine_expr, other.refine_expr) {
1105 (Some(refine_expr), Some(other_refine_expr)) => {
1106 Some(refine_expr.and(other_refine_expr))
1107 }
1108 (Some(refine_expr), None) => Some(refine_expr),
1109 (None, Some(refine_expr)) => Some(refine_expr),
1110 (None, None) => None,
1111 };
1112 Self {
1113 scalar_query,
1114 refine_expr,
1115 }
1116 }
1117
1118 fn maybe_or(self, other: Self) -> Option<Self> {
1125 let scalar_query = self.scalar_query?;
1128 let other_scalar_query = other.scalar_query?;
1129 let scalar_query = Some(ScalarIndexExpr::Or(
1130 Box::new(scalar_query),
1131 Box::new(other_scalar_query),
1132 ));
1133
1134 let refine_expr = match (self.refine_expr, other.refine_expr) {
1135 (Some(_), Some(_)) => {
1145 return None;
1146 }
1147 (Some(_), None) => {
1148 return None;
1149 }
1150 (None, Some(_)) => {
1151 return None;
1152 }
1153 (None, None) => None,
1154 };
1155 Some(Self {
1156 scalar_query,
1157 refine_expr,
1158 })
1159 }
1160
1161 fn refine(self, expr: Expr) -> Self {
1162 match self.refine_expr {
1163 Some(refine_expr) => Self {
1164 scalar_query: self.scalar_query,
1165 refine_expr: Some(refine_expr.and(expr)),
1166 },
1167 None => Self {
1168 scalar_query: self.scalar_query,
1169 refine_expr: Some(expr),
1170 },
1171 }
1172 }
1173}
1174
1175#[async_trait]
1179pub trait ScalarIndexLoader: Send + Sync {
1180 async fn load_index(
1182 &self,
1183 column: &str,
1184 index_name: &str,
1185 metrics: &dyn MetricsCollector,
1186 ) -> Result<Arc<dyn ScalarIndex>>;
1187}
1188
1189#[derive(Debug, Clone)]
1191pub struct ScalarIndexSearch {
1192 pub column: String,
1194 pub index_name: String,
1196 pub query: Arc<dyn AnyQuery>,
1198 pub needs_recheck: bool,
1200}
1201
1202impl PartialEq for ScalarIndexSearch {
1203 fn eq(&self, other: &Self) -> bool {
1204 self.column == other.column
1205 && self.index_name == other.index_name
1206 && self.query.as_ref().eq(other.query.as_ref())
1207 }
1208}
1209
1210#[derive(Debug, Clone)]
1215pub enum ScalarIndexExpr {
1216 Not(Box<Self>),
1217 And(Box<Self>, Box<Self>),
1218 Or(Box<Self>, Box<Self>),
1219 Query(ScalarIndexSearch),
1220}
1221
1222impl PartialEq for ScalarIndexExpr {
1223 fn eq(&self, other: &Self) -> bool {
1224 match (self, other) {
1225 (Self::Not(l0), Self::Not(r0)) => l0 == r0,
1226 (Self::And(l0, l1), Self::And(r0, r1)) => l0 == r0 && l1 == r1,
1227 (Self::Or(l0, l1), Self::Or(r0, r1)) => l0 == r0 && l1 == r1,
1228 (Self::Query(l_search), Self::Query(r_search)) => l_search == r_search,
1229 _ => false,
1230 }
1231 }
1232}
1233
1234impl std::fmt::Display for ScalarIndexExpr {
1235 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1236 match self {
1237 Self::Not(inner) => write!(f, "NOT({})", inner),
1238 Self::And(lhs, rhs) => write!(f, "AND({},{})", lhs, rhs),
1239 Self::Or(lhs, rhs) => write!(f, "OR({},{})", lhs, rhs),
1240 Self::Query(search) => write!(
1241 f,
1242 "[{}]@{}",
1243 search.query.format(&search.column),
1244 search.index_name
1245 ),
1246 }
1247 }
1248}
1249
1250pub static INDEX_EXPR_RESULT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
1256 Arc::new(Schema::new(vec![
1257 Field::new("result".to_string(), DataType::Binary, true),
1258 Field::new("discriminant".to_string(), DataType::UInt32, true),
1259 Field::new("fragments_covered".to_string(), DataType::Binary, true),
1260 ]))
1261});
1262
1263#[derive(Debug)]
1264enum NullableIndexExprResult {
1265 Exact(NullableRowAddrMask),
1266 AtMost(NullableRowAddrMask),
1267 AtLeast(NullableRowAddrMask),
1268}
1269
1270impl From<SearchResult> for NullableIndexExprResult {
1271 fn from(result: SearchResult) -> Self {
1272 match result {
1273 SearchResult::Exact(mask) => Self::Exact(NullableRowAddrMask::AllowList(mask)),
1274 SearchResult::AtMost(mask) => Self::AtMost(NullableRowAddrMask::AllowList(mask)),
1275 SearchResult::AtLeast(mask) => Self::AtLeast(NullableRowAddrMask::AllowList(mask)),
1276 }
1277 }
1278}
1279
1280impl std::ops::BitAnd<Self> for NullableIndexExprResult {
1281 type Output = Self;
1282
1283 fn bitand(self, rhs: Self) -> Self {
1284 match (self, rhs) {
1285 (Self::Exact(lhs), Self::Exact(rhs)) => Self::Exact(lhs & rhs),
1286 (Self::Exact(lhs), Self::AtMost(rhs)) | (Self::AtMost(lhs), Self::Exact(rhs)) => {
1287 Self::AtMost(lhs & rhs)
1288 }
1289 (Self::Exact(exact), Self::AtLeast(_)) | (Self::AtLeast(_), Self::Exact(exact)) => {
1290 Self::AtMost(exact)
1294 }
1295 (Self::AtMost(lhs), Self::AtMost(rhs)) => Self::AtMost(lhs & rhs),
1296 (Self::AtLeast(lhs), Self::AtLeast(rhs)) => Self::AtLeast(lhs & rhs),
1297 (Self::AtMost(most), Self::AtLeast(_)) | (Self::AtLeast(_), Self::AtMost(most)) => {
1298 Self::AtMost(most)
1299 }
1300 }
1301 }
1302}
1303
1304impl std::ops::BitOr<Self> for NullableIndexExprResult {
1305 type Output = Self;
1306
1307 fn bitor(self, rhs: Self) -> Self {
1308 match (self, rhs) {
1309 (Self::Exact(lhs), Self::Exact(rhs)) => Self::Exact(lhs | rhs),
1310 (Self::Exact(lhs), Self::AtMost(rhs)) | (Self::AtMost(rhs), Self::Exact(lhs)) => {
1311 Self::AtMost(lhs | rhs)
1315 }
1316 (Self::Exact(lhs), Self::AtLeast(rhs)) | (Self::AtLeast(rhs), Self::Exact(lhs)) => {
1317 Self::AtLeast(lhs | rhs)
1318 }
1319 (Self::AtMost(lhs), Self::AtMost(rhs)) => Self::AtMost(lhs | rhs),
1320 (Self::AtLeast(lhs), Self::AtLeast(rhs)) => Self::AtLeast(lhs | rhs),
1321 (Self::AtMost(_), Self::AtLeast(least)) | (Self::AtLeast(least), Self::AtMost(_)) => {
1322 Self::AtLeast(least)
1323 }
1324 }
1325 }
1326}
1327
1328impl NullableIndexExprResult {
1329 pub fn drop_nulls(self) -> IndexExprResult {
1330 match self {
1331 Self::Exact(mask) => IndexExprResult::Exact(mask.drop_nulls()),
1332 Self::AtMost(mask) => IndexExprResult::AtMost(mask.drop_nulls()),
1333 Self::AtLeast(mask) => IndexExprResult::AtLeast(mask.drop_nulls()),
1334 }
1335 }
1336}
1337
1338#[derive(Debug)]
1339pub enum IndexExprResult {
1340 Exact(RowAddrMask),
1342 AtMost(RowAddrMask),
1346 AtLeast(RowAddrMask),
1350}
1351
1352impl IndexExprResult {
1353 pub fn row_addr_mask(&self) -> &RowAddrMask {
1354 match self {
1355 Self::Exact(mask) => mask,
1356 Self::AtMost(mask) => mask,
1357 Self::AtLeast(mask) => mask,
1358 }
1359 }
1360
1361 pub fn discriminant(&self) -> u32 {
1362 match self {
1363 Self::Exact(_) => 0,
1364 Self::AtMost(_) => 1,
1365 Self::AtLeast(_) => 2,
1366 }
1367 }
1368
1369 pub fn from_parts(mask: RowAddrMask, discriminant: u32) -> Result<Self> {
1370 match discriminant {
1371 0 => Ok(Self::Exact(mask)),
1372 1 => Ok(Self::AtMost(mask)),
1373 2 => Ok(Self::AtLeast(mask)),
1374 _ => Err(Error::invalid_input_source(
1375 format!("Invalid IndexExprResult discriminant: {}", discriminant).into(),
1376 )),
1377 }
1378 }
1379
1380 #[instrument(skip_all)]
1381 pub fn serialize_to_arrow(
1382 &self,
1383 fragments_covered_by_result: &RoaringBitmap,
1384 ) -> Result<RecordBatch> {
1385 let row_addr_mask = self.row_addr_mask();
1386 let row_addr_mask_arr = row_addr_mask.into_arrow()?;
1387 let discriminant = self.discriminant();
1388 let discriminant_arr =
1389 Arc::new(UInt32Array::from(vec![discriminant, discriminant])) as Arc<dyn Array>;
1390 let mut fragments_covered_builder = BinaryBuilder::new();
1391 let fragments_covered_bytes_len = fragments_covered_by_result.serialized_size();
1392 let mut fragments_covered_bytes = Vec::with_capacity(fragments_covered_bytes_len);
1393 fragments_covered_by_result.serialize_into(&mut fragments_covered_bytes)?;
1394 fragments_covered_builder.append_value(fragments_covered_bytes);
1395 fragments_covered_builder.append_null();
1396 let fragments_covered_arr = Arc::new(fragments_covered_builder.finish()) as Arc<dyn Array>;
1397 Ok(RecordBatch::try_new(
1398 INDEX_EXPR_RESULT_SCHEMA.clone(),
1399 vec![
1400 Arc::new(row_addr_mask_arr),
1401 Arc::new(discriminant_arr),
1402 Arc::new(fragments_covered_arr),
1403 ],
1404 )?)
1405 }
1406}
1407
1408impl ScalarIndexExpr {
1409 #[async_recursion]
1416 async fn evaluate_impl(
1417 &self,
1418 index_loader: &dyn ScalarIndexLoader,
1419 metrics: &dyn MetricsCollector,
1420 ) -> Result<NullableIndexExprResult> {
1421 match self {
1422 Self::Not(inner) => {
1423 let result = inner.evaluate_impl(index_loader, metrics).await?;
1424 Ok(match result {
1426 NullableIndexExprResult::Exact(mask) => NullableIndexExprResult::Exact(!mask),
1427 NullableIndexExprResult::AtMost(mask) => {
1428 NullableIndexExprResult::AtLeast(!mask)
1429 }
1430 NullableIndexExprResult::AtLeast(mask) => {
1431 NullableIndexExprResult::AtMost(!mask)
1432 }
1433 })
1434 }
1435 Self::And(lhs, rhs) => {
1436 let lhs_result = lhs.evaluate_impl(index_loader, metrics);
1437 let rhs_result = rhs.evaluate_impl(index_loader, metrics);
1438 let (lhs_result, rhs_result) = try_join!(lhs_result, rhs_result)?;
1439 Ok(lhs_result & rhs_result)
1440 }
1441 Self::Or(lhs, rhs) => {
1442 let lhs_result = lhs.evaluate_impl(index_loader, metrics);
1443 let rhs_result = rhs.evaluate_impl(index_loader, metrics);
1444 let (lhs_result, rhs_result) = try_join!(lhs_result, rhs_result)?;
1445 Ok(lhs_result | rhs_result)
1446 }
1447 Self::Query(search) => {
1448 let index = index_loader
1449 .load_index(&search.column, &search.index_name, metrics)
1450 .await?;
1451 let search_result = index.search(search.query.as_ref(), metrics).await?;
1452 Ok(search_result.into())
1453 }
1454 }
1455 }
1456
1457 #[instrument(level = "debug", skip_all)]
1458 pub async fn evaluate(
1459 &self,
1460 index_loader: &dyn ScalarIndexLoader,
1461 metrics: &dyn MetricsCollector,
1462 ) -> Result<IndexExprResult> {
1463 Ok(self
1464 .evaluate_impl(index_loader, metrics)
1465 .await?
1466 .drop_nulls())
1467 }
1468
1469 pub fn to_expr(&self) -> Expr {
1470 match self {
1471 Self::Not(inner) => Expr::Not(inner.to_expr().into()),
1472 Self::And(lhs, rhs) => {
1473 let lhs = lhs.to_expr();
1474 let rhs = rhs.to_expr();
1475 lhs.and(rhs)
1476 }
1477 Self::Or(lhs, rhs) => {
1478 let lhs = lhs.to_expr();
1479 let rhs = rhs.to_expr();
1480 lhs.or(rhs)
1481 }
1482 Self::Query(search) => search.query.to_expr(search.column.clone()),
1483 }
1484 }
1485
1486 pub fn needs_recheck(&self) -> bool {
1487 match self {
1488 Self::Not(inner) => inner.needs_recheck(),
1489 Self::And(lhs, rhs) | Self::Or(lhs, rhs) => lhs.needs_recheck() || rhs.needs_recheck(),
1490 Self::Query(search) => search.needs_recheck,
1491 }
1492 }
1493}
1494
1495fn maybe_column(expr: &Expr) -> Option<&str> {
1497 match expr {
1498 Expr::Column(col) => Some(&col.name),
1499 _ => None,
1500 }
1501}
1502
1503fn extract_nested_column_path(expr: &Expr) -> Option<String> {
1506 let mut current_expr = expr;
1507 let mut parts = Vec::new();
1508
1509 loop {
1511 match current_expr {
1512 Expr::ScalarFunction(udf) if udf.name() == "get_field" => {
1513 if udf.args.len() != 2 {
1514 return None;
1515 }
1516 if let Expr::Literal(ScalarValue::Utf8(Some(field_name)), _) = &udf.args[1] {
1519 parts.push(field_name.clone());
1520 } else {
1521 return None;
1522 }
1523 current_expr = &udf.args[0];
1525 }
1526 Expr::Column(col) => {
1527 parts.push(col.name.clone());
1529 break;
1530 }
1531 _ => {
1532 return None;
1533 }
1534 }
1535 }
1536
1537 parts.reverse();
1539
1540 let field_refs: Vec<&str> = parts.iter().map(|s| s.as_str()).collect();
1542 Some(lance_core::datatypes::format_field_path(&field_refs))
1543}
1544
1545fn maybe_indexed_column<'b>(
1552 expr: &Expr,
1553 index_info: &'b dyn IndexInformationProvider,
1554) -> Option<(String, DataType, &'b dyn ScalarQueryParser)> {
1555 if let Some(nested_path) = extract_nested_column_path(expr)
1557 && let Some((data_type, parser)) = index_info.get_index(&nested_path)
1558 && let Some(data_type) = parser.is_valid_reference(expr, data_type)
1559 {
1560 return Some((nested_path, data_type, parser));
1561 }
1562
1563 match expr {
1564 Expr::Column(col) => {
1565 let col = col.name.as_str();
1566 let (data_type, parser) = index_info.get_index(col)?;
1567 if let Some(data_type) = parser.is_valid_reference(expr, data_type) {
1568 Some((col.to_string(), data_type, parser))
1569 } else {
1570 None
1571 }
1572 }
1573 Expr::ScalarFunction(udf) => {
1574 if udf.args.is_empty() {
1575 return None;
1576 }
1577 let col = maybe_column(&udf.args[0])?;
1579 let (data_type, parser) = index_info.get_index(col)?;
1580 if let Some(data_type) = parser.is_valid_reference(expr, data_type) {
1581 Some((col.to_string(), data_type, parser))
1582 } else {
1583 None
1584 }
1585 }
1586 _ => None,
1587 }
1588}
1589
1590fn maybe_scalar(expr: &Expr, expected_type: &DataType) -> Option<ScalarValue> {
1592 match expr {
1593 Expr::Literal(value, _) => safe_coerce_scalar(value, expected_type),
1594 Expr::Cast(cast) => match cast.expr.as_ref() {
1602 Expr::Literal(value, _) => {
1603 let casted = value.cast_to(&cast.data_type).ok()?;
1604 safe_coerce_scalar(&casted, expected_type)
1605 }
1606 _ => None,
1607 },
1608 Expr::ScalarFunction(scalar_function) => {
1609 if scalar_function.name() == "arrow_cast" {
1610 if scalar_function.args.len() != 2 {
1611 return None;
1612 }
1613 match (&scalar_function.args[0], &scalar_function.args[1]) {
1614 (Expr::Literal(value, _), Expr::Literal(cast_type, _)) => {
1615 let target_type = scalar_function
1616 .func
1617 .return_field_from_args(ReturnFieldArgs {
1618 arg_fields: &[
1619 Arc::new(Field::new("expression", value.data_type(), false)),
1620 Arc::new(Field::new("datatype", cast_type.data_type(), false)),
1621 ],
1622 scalar_arguments: &[Some(value), Some(cast_type)],
1623 })
1624 .ok()?;
1625 let casted = value.cast_to(target_type.data_type()).ok()?;
1626 safe_coerce_scalar(&casted, expected_type)
1627 }
1628 _ => None,
1629 }
1630 } else {
1631 None
1632 }
1633 }
1634 _ => None,
1635 }
1636}
1637
1638fn maybe_scalar_list(exprs: &Vec<Expr>, expected_type: &DataType) -> Option<Vec<ScalarValue>> {
1640 let mut scalar_values = Vec::with_capacity(exprs.len());
1641 for expr in exprs {
1642 match maybe_scalar(expr, expected_type) {
1643 Some(scalar_val) => {
1644 scalar_values.push(scalar_val);
1645 }
1646 None => {
1647 return None;
1648 }
1649 }
1650 }
1651 Some(scalar_values)
1652}
1653
1654fn visit_between(
1655 between: &Between,
1656 index_info: &dyn IndexInformationProvider,
1657) -> Option<IndexedExpression> {
1658 let (column, col_type, query_parser) = maybe_indexed_column(&between.expr, index_info)?;
1659 let low = maybe_scalar(&between.low, &col_type)?;
1660 let high = maybe_scalar(&between.high, &col_type)?;
1661
1662 let indexed_expr =
1663 query_parser.visit_between(&column, &Bound::Included(low), &Bound::Included(high))?;
1664
1665 if between.negated {
1666 indexed_expr.maybe_not()
1667 } else {
1668 Some(indexed_expr)
1669 }
1670}
1671
1672fn visit_in_list(
1673 in_list: &InList,
1674 index_info: &dyn IndexInformationProvider,
1675) -> Option<IndexedExpression> {
1676 let (column, col_type, query_parser) = maybe_indexed_column(&in_list.expr, index_info)?;
1677 let values = maybe_scalar_list(&in_list.list, &col_type)?;
1678
1679 let indexed_expr = query_parser.visit_in_list(&column, &values)?;
1680
1681 if in_list.negated {
1682 indexed_expr.maybe_not()
1683 } else {
1684 Some(indexed_expr)
1685 }
1686}
1687
1688fn visit_is_bool(
1689 expr: &Expr,
1690 index_info: &dyn IndexInformationProvider,
1691 value: bool,
1692) -> Option<IndexedExpression> {
1693 let (column, col_type, query_parser) = maybe_indexed_column(expr, index_info)?;
1694 if col_type != DataType::Boolean {
1695 None
1696 } else {
1697 query_parser.visit_is_bool(&column, value)
1698 }
1699}
1700
1701fn visit_column(
1703 col: &Expr,
1704 index_info: &dyn IndexInformationProvider,
1705) -> Option<IndexedExpression> {
1706 let (column, col_type, query_parser) = maybe_indexed_column(col, index_info)?;
1707 if col_type != DataType::Boolean {
1708 None
1709 } else {
1710 query_parser.visit_is_bool(&column, true)
1711 }
1712}
1713
1714fn visit_is_null(
1715 expr: &Expr,
1716 index_info: &dyn IndexInformationProvider,
1717 negated: bool,
1718) -> Option<IndexedExpression> {
1719 let (column, _, query_parser) = maybe_indexed_column(expr, index_info)?;
1720 let indexed_expr = query_parser.visit_is_null(&column)?;
1721 if negated {
1722 indexed_expr.maybe_not()
1723 } else {
1724 Some(indexed_expr)
1725 }
1726}
1727
1728fn visit_not(
1729 expr: &Expr,
1730 index_info: &dyn IndexInformationProvider,
1731 depth: usize,
1732) -> Result<Option<IndexedExpression>> {
1733 let node = visit_node(expr, index_info, depth + 1)?;
1734 Ok(node.and_then(|node| node.maybe_not()))
1735}
1736
1737fn visit_comparison(
1738 expr: &BinaryExpr,
1739 index_info: &dyn IndexInformationProvider,
1740) -> Option<IndexedExpression> {
1741 let left_col = maybe_indexed_column(&expr.left, index_info);
1742 if let Some((column, col_type, query_parser)) = left_col {
1743 let scalar = maybe_scalar(&expr.right, &col_type)?;
1744 query_parser.visit_comparison(&column, &scalar, &expr.op)
1745 } else {
1746 None
1749 }
1750}
1751
1752fn maybe_range(
1753 expr: &BinaryExpr,
1754 index_info: &dyn IndexInformationProvider,
1755) -> Option<IndexedExpression> {
1756 let left_expr = match expr.left.as_ref() {
1757 Expr::BinaryExpr(binary_expr) => Some(binary_expr),
1758 _ => None,
1759 }?;
1760 let right_expr = match expr.right.as_ref() {
1761 Expr::BinaryExpr(binary_expr) => Some(binary_expr),
1762 _ => None,
1763 }?;
1764
1765 let (left_col, dt, parser) = maybe_indexed_column(&left_expr.left, index_info)?;
1766 let right_col = maybe_column(&right_expr.left)?;
1767
1768 if left_col != right_col {
1769 return None;
1770 }
1771
1772 let left_value = maybe_scalar(&left_expr.right, &dt)?;
1773 let right_value = maybe_scalar(&right_expr.right, &dt)?;
1774
1775 let (low, high) = match (left_expr.op, right_expr.op) {
1776 (Operator::GtEq, Operator::LtEq) => {
1778 (Bound::Included(left_value), Bound::Included(right_value))
1779 }
1780 (Operator::GtEq, Operator::Lt) => {
1782 (Bound::Included(left_value), Bound::Excluded(right_value))
1783 }
1784 (Operator::Gt, Operator::LtEq) => {
1786 (Bound::Excluded(left_value), Bound::Included(right_value))
1787 }
1788 (Operator::Gt, Operator::Lt) => (Bound::Excluded(left_value), Bound::Excluded(right_value)),
1790 (Operator::LtEq, Operator::GtEq) => {
1792 (Bound::Included(right_value), Bound::Included(left_value))
1793 }
1794 (Operator::LtEq, Operator::Gt) => {
1796 (Bound::Included(right_value), Bound::Excluded(left_value))
1797 }
1798 (Operator::Lt, Operator::GtEq) => {
1800 (Bound::Excluded(right_value), Bound::Included(left_value))
1801 }
1802 (Operator::Lt, Operator::Gt) => (Bound::Excluded(right_value), Bound::Excluded(left_value)),
1804 _ => return None,
1805 };
1806
1807 parser.visit_between(&left_col, &low, &high)
1808}
1809
1810fn visit_and(
1811 expr: &BinaryExpr,
1812 index_info: &dyn IndexInformationProvider,
1813 depth: usize,
1814) -> Result<Option<IndexedExpression>> {
1815 if let Some(range_expr) = maybe_range(expr, index_info) {
1823 return Ok(Some(range_expr));
1824 }
1825
1826 let left = visit_node(&expr.left, index_info, depth + 1)?;
1827 let right = visit_node(&expr.right, index_info, depth + 1)?;
1828 Ok(match (left, right) {
1829 (Some(left), Some(right)) => Some(left.and(right)),
1830 (Some(left), None) => Some(left.refine((*expr.right).clone())),
1831 (None, Some(right)) => Some(right.refine((*expr.left).clone())),
1832 (None, None) => None,
1833 })
1834}
1835
1836fn visit_or(
1837 expr: &BinaryExpr,
1838 index_info: &dyn IndexInformationProvider,
1839 depth: usize,
1840) -> Result<Option<IndexedExpression>> {
1841 let left = visit_node(&expr.left, index_info, depth + 1)?;
1842 let right = visit_node(&expr.right, index_info, depth + 1)?;
1843 Ok(match (left, right) {
1844 (Some(left), Some(right)) => left.maybe_or(right),
1845 (Some(_), None) => None,
1851 (None, Some(_)) => None,
1852 (None, None) => None,
1853 })
1854}
1855
1856fn visit_binary_expr(
1857 expr: &BinaryExpr,
1858 index_info: &dyn IndexInformationProvider,
1859 depth: usize,
1860) -> Result<Option<IndexedExpression>> {
1861 match &expr.op {
1862 Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq | Operator::Eq => {
1863 Ok(visit_comparison(expr, index_info))
1864 }
1865 Operator::NotEq => Ok(visit_comparison(expr, index_info).and_then(|node| node.maybe_not())),
1867 Operator::And => visit_and(expr, index_info, depth),
1868 Operator::Or => visit_or(expr, index_info, depth),
1869 _ => Ok(None),
1870 }
1871}
1872
1873fn visit_scalar_fn(
1874 scalar_fn: &ScalarFunction,
1875 index_info: &dyn IndexInformationProvider,
1876) -> Option<IndexedExpression> {
1877 if scalar_fn.args.is_empty() {
1878 return None;
1879 }
1880 let (col, data_type, query_parser) = maybe_indexed_column(&scalar_fn.args[0], index_info)?;
1881 query_parser.visit_scalar_function(&col, &data_type, &scalar_fn.func, &scalar_fn.args)
1882}
1883
1884fn visit_like_expr(
1885 like: &Like,
1886 index_info: &dyn IndexInformationProvider,
1887) -> Option<IndexedExpression> {
1888 let (column, _, query_parser) = maybe_indexed_column(&like.expr, index_info)?;
1889
1890 let pattern = match like.pattern.as_ref() {
1892 Expr::Literal(scalar, _) => scalar.clone(),
1893 _ => return None,
1894 };
1895
1896 query_parser.visit_like(&column, like, &pattern)
1897}
1898
1899fn visit_node(
1900 expr: &Expr,
1901 index_info: &dyn IndexInformationProvider,
1902 depth: usize,
1903) -> Result<Option<IndexedExpression>> {
1904 if depth >= MAX_DEPTH {
1905 return Err(Error::invalid_input(format!(
1906 "the filter expression is too long, lance limit the max number of conditions to {}",
1907 MAX_DEPTH
1908 )));
1909 }
1910 match expr {
1911 Expr::Between(between) => Ok(visit_between(between, index_info)),
1912 Expr::Alias(alias) => visit_node(alias.expr.as_ref(), index_info, depth),
1913 Expr::Column(_) => Ok(visit_column(expr, index_info)),
1914 Expr::InList(in_list) => Ok(visit_in_list(in_list, index_info)),
1915 Expr::IsFalse(expr) => Ok(visit_is_bool(expr.as_ref(), index_info, false)),
1916 Expr::IsTrue(expr) => Ok(visit_is_bool(expr.as_ref(), index_info, true)),
1917 Expr::IsNull(expr) => Ok(visit_is_null(expr.as_ref(), index_info, false)),
1918 Expr::IsNotNull(expr) => Ok(visit_is_null(expr.as_ref(), index_info, true)),
1919 Expr::Not(expr) => visit_not(expr.as_ref(), index_info, depth),
1920 Expr::BinaryExpr(binary_expr) => visit_binary_expr(binary_expr, index_info, depth),
1921 Expr::ScalarFunction(scalar_fn) => Ok(visit_scalar_fn(scalar_fn, index_info)),
1922 Expr::Like(like) => {
1923 if like.negated {
1924 Ok(None)
1926 } else {
1927 Ok(visit_like_expr(like, index_info))
1928 }
1929 }
1930 _ => Ok(None),
1931 }
1932}
1933
1934pub trait IndexInformationProvider {
1936 fn get_index(&self, col: &str) -> Option<(&DataType, &dyn ScalarQueryParser)>;
1939}
1940
1941pub fn apply_scalar_indices(
1944 expr: Expr,
1945 index_info: &dyn IndexInformationProvider,
1946) -> Result<IndexedExpression> {
1947 Ok(visit_node(&expr, index_info, 0)?.unwrap_or(IndexedExpression::refine_only(expr)))
1948}
1949
1950#[derive(Clone, Default, Debug)]
1951pub struct FilterPlan {
1952 pub index_query: Option<ScalarIndexExpr>,
1953 pub skip_recheck: bool,
1955 pub refine_expr: Option<Expr>,
1956 pub full_expr: Option<Expr>,
1957}
1958
1959impl FilterPlan {
1960 pub fn empty() -> Self {
1961 Self {
1962 index_query: None,
1963 skip_recheck: true,
1964 refine_expr: None,
1965 full_expr: None,
1966 }
1967 }
1968
1969 pub fn new_refine_only(expr: Expr) -> Self {
1970 Self {
1971 index_query: None,
1972 skip_recheck: true,
1973 refine_expr: Some(expr.clone()),
1974 full_expr: Some(expr),
1975 }
1976 }
1977
1978 pub fn is_empty(&self) -> bool {
1979 self.refine_expr.is_none() && self.index_query.is_none()
1980 }
1981
1982 pub fn all_columns(&self) -> Vec<String> {
1983 self.full_expr
1984 .as_ref()
1985 .map(Planner::column_names_in_expr)
1986 .unwrap_or_default()
1987 }
1988
1989 pub fn refine_columns(&self) -> Vec<String> {
1990 self.refine_expr
1991 .as_ref()
1992 .map(Planner::column_names_in_expr)
1993 .unwrap_or_default()
1994 }
1995
1996 pub fn has_refine(&self) -> bool {
1998 self.refine_expr.is_some()
1999 }
2000
2001 pub fn has_index_query(&self) -> bool {
2003 self.index_query.is_some()
2004 }
2005
2006 pub fn has_any_filter(&self) -> bool {
2007 self.refine_expr.is_some() || self.index_query.is_some()
2008 }
2009
2010 pub fn make_refine_only(&mut self) {
2011 self.index_query = None;
2012 self.refine_expr = self.full_expr.clone();
2013 }
2014
2015 pub fn is_exact_index_search(&self) -> bool {
2017 self.index_query.is_some() && self.refine_expr.is_none() && self.skip_recheck
2018 }
2019}
2020
2021pub trait PlannerIndexExt {
2022 fn create_filter_plan(
2029 &self,
2030 filter: Expr,
2031 index_info: &dyn IndexInformationProvider,
2032 use_scalar_index: bool,
2033 ) -> Result<FilterPlan>;
2034}
2035
2036impl PlannerIndexExt for Planner {
2037 fn create_filter_plan(
2038 &self,
2039 filter: Expr,
2040 index_info: &dyn IndexInformationProvider,
2041 use_scalar_index: bool,
2042 ) -> Result<FilterPlan> {
2043 let logical_expr = self.optimize_expr(filter)?;
2044 if use_scalar_index {
2045 let indexed_expr = apply_scalar_indices(logical_expr.clone(), index_info)?;
2046 let mut skip_recheck = false;
2047 if let Some(scalar_query) = indexed_expr.scalar_query.as_ref() {
2048 skip_recheck = !scalar_query.needs_recheck();
2049 }
2050 Ok(FilterPlan {
2051 index_query: indexed_expr.scalar_query,
2052 refine_expr: indexed_expr.refine_expr,
2053 full_expr: Some(logical_expr),
2054 skip_recheck,
2055 })
2056 } else {
2057 Ok(FilterPlan {
2058 index_query: None,
2059 skip_recheck: true,
2060 refine_expr: Some(logical_expr.clone()),
2061 full_expr: Some(logical_expr),
2062 })
2063 }
2064 }
2065}
2066
2067#[cfg(test)]
2068mod tests {
2069 use std::collections::HashMap;
2070
2071 use arrow_schema::{Field, Schema};
2072 use chrono::Utc;
2073 use datafusion_common::{Column, DFSchema};
2074 use datafusion_expr::execution_props::ExecutionProps;
2075 use datafusion_expr::simplify::SimplifyContext;
2076 use lance_datafusion::exec::{LanceExecutionOptions, get_session_context};
2077
2078 use crate::scalar::json::{JsonQuery, JsonQueryParser};
2079
2080 use super::*;
2081
2082 struct ColInfo {
2083 data_type: DataType,
2084 parser: Box<dyn ScalarQueryParser>,
2085 }
2086
2087 impl ColInfo {
2088 fn new(data_type: DataType, parser: Box<dyn ScalarQueryParser>) -> Self {
2089 Self { data_type, parser }
2090 }
2091 }
2092
2093 struct MockIndexInfoProvider {
2094 indexed_columns: HashMap<String, ColInfo>,
2095 }
2096
2097 impl MockIndexInfoProvider {
2098 fn new(indexed_columns: Vec<(&str, ColInfo)>) -> Self {
2099 Self {
2100 indexed_columns: HashMap::from_iter(
2101 indexed_columns
2102 .into_iter()
2103 .map(|(s, ty)| (s.to_string(), ty)),
2104 ),
2105 }
2106 }
2107 }
2108
2109 impl IndexInformationProvider for MockIndexInfoProvider {
2110 fn get_index(&self, col: &str) -> Option<(&DataType, &dyn ScalarQueryParser)> {
2111 self.indexed_columns
2112 .get(col)
2113 .map(|col_info| (&col_info.data_type, col_info.parser.as_ref()))
2114 }
2115 }
2116
2117 fn check(
2118 index_info: &dyn IndexInformationProvider,
2119 expr: &str,
2120 expected: Option<IndexedExpression>,
2121 optimize: bool,
2122 ) {
2123 let schema = Schema::new(vec![
2124 Field::new("color", DataType::Utf8, false),
2125 Field::new("size", DataType::Float32, false),
2126 Field::new("aisle", DataType::UInt32, false),
2127 Field::new("on_sale", DataType::Boolean, false),
2128 Field::new("price", DataType::Float32, false),
2129 Field::new("json", DataType::LargeBinary, false),
2130 ]);
2131 let df_schema: DFSchema = schema.try_into().unwrap();
2132
2133 let ctx = get_session_context(&LanceExecutionOptions::default());
2134 let state = ctx.state();
2135 let mut expr = state.create_logical_expr(expr, &df_schema).unwrap();
2136 if optimize {
2137 let props = ExecutionProps::new().with_query_execution_start_time(Utc::now());
2138 let simplify_context = SimplifyContext::new(&props).with_schema(Arc::new(df_schema));
2139 let simplifier =
2140 datafusion::optimizer::simplify_expressions::ExprSimplifier::new(simplify_context);
2141 expr = simplifier.simplify(expr).unwrap();
2142 }
2143
2144 let actual = apply_scalar_indices(expr.clone(), index_info).unwrap();
2145 if let Some(expected) = expected {
2146 assert_eq!(actual, expected);
2147 } else {
2148 assert!(actual.scalar_query.is_none());
2149 assert_eq!(actual.refine_expr.unwrap(), expr);
2150 }
2151 }
2152
2153 fn check_no_index(index_info: &dyn IndexInformationProvider, expr: &str) {
2154 check(index_info, expr, None, false)
2155 }
2156
2157 fn check_simple(
2158 index_info: &dyn IndexInformationProvider,
2159 expr: &str,
2160 col: &str,
2161 query: impl AnyQuery,
2162 ) {
2163 check(
2164 index_info,
2165 expr,
2166 Some(IndexedExpression::index_query(
2167 col.to_string(),
2168 format!("{}_idx", col),
2169 Arc::new(query),
2170 )),
2171 false,
2172 )
2173 }
2174
2175 fn check_range(
2176 index_info: &dyn IndexInformationProvider,
2177 expr: &str,
2178 col: &str,
2179 query: SargableQuery,
2180 ) {
2181 check(
2182 index_info,
2183 expr,
2184 Some(IndexedExpression::index_query(
2185 col.to_string(),
2186 format!("{}_idx", col),
2187 Arc::new(query),
2188 )),
2189 true,
2190 )
2191 }
2192
2193 fn check_simple_negated(
2194 index_info: &dyn IndexInformationProvider,
2195 expr: &str,
2196 col: &str,
2197 query: SargableQuery,
2198 ) {
2199 check(
2200 index_info,
2201 expr,
2202 Some(
2203 IndexedExpression::index_query(
2204 col.to_string(),
2205 format!("{}_idx", col),
2206 Arc::new(query),
2207 )
2208 .maybe_not()
2209 .unwrap(),
2210 ),
2211 false,
2212 )
2213 }
2214
2215 #[test]
2216 fn test_expressions() {
2217 let index_info = MockIndexInfoProvider::new(vec![
2218 (
2219 "color",
2220 ColInfo::new(
2221 DataType::Utf8,
2222 Box::new(SargableQueryParser::new("color_idx".to_string(), false)),
2223 ),
2224 ),
2225 (
2226 "aisle",
2227 ColInfo::new(
2228 DataType::UInt32,
2229 Box::new(SargableQueryParser::new("aisle_idx".to_string(), false)),
2230 ),
2231 ),
2232 (
2233 "on_sale",
2234 ColInfo::new(
2235 DataType::Boolean,
2236 Box::new(SargableQueryParser::new("on_sale_idx".to_string(), false)),
2237 ),
2238 ),
2239 (
2240 "price",
2241 ColInfo::new(
2242 DataType::Float32,
2243 Box::new(SargableQueryParser::new("price_idx".to_string(), false)),
2244 ),
2245 ),
2246 (
2247 "json",
2248 ColInfo::new(
2249 DataType::LargeBinary,
2250 Box::new(JsonQueryParser::new(
2251 "$.name".to_string(),
2252 Box::new(SargableQueryParser::new("json_idx".to_string(), false)),
2253 )),
2254 ),
2255 ),
2256 ]);
2257
2258 check_simple(
2259 &index_info,
2260 "json_extract(json, '$.name') = 'foo'",
2261 "json",
2262 JsonQuery::new(
2263 Arc::new(SargableQuery::Equals(ScalarValue::Utf8(Some(
2264 "foo".to_string(),
2265 )))),
2266 "$.name".to_string(),
2267 ),
2268 );
2269
2270 check_no_index(&index_info, "size BETWEEN 5 AND 10");
2271 check_simple(
2273 &index_info,
2274 "aisle = arrow_cast(5, 'Int16')",
2275 "aisle",
2276 SargableQuery::Equals(ScalarValue::UInt32(Some(5))),
2277 );
2278 check_range(
2280 &index_info,
2281 "aisle BETWEEN 5 AND 10",
2282 "aisle",
2283 SargableQuery::Range(
2284 Bound::Included(ScalarValue::UInt32(Some(5))),
2285 Bound::Included(ScalarValue::UInt32(Some(10))),
2286 ),
2287 );
2288 check_range(
2289 &index_info,
2290 "aisle >= 5 AND aisle <= 10",
2291 "aisle",
2292 SargableQuery::Range(
2293 Bound::Included(ScalarValue::UInt32(Some(5))),
2294 Bound::Included(ScalarValue::UInt32(Some(10))),
2295 ),
2296 );
2297
2298 check_range(
2299 &index_info,
2300 "aisle <= 10 AND aisle >= 5",
2301 "aisle",
2302 SargableQuery::Range(
2303 Bound::Included(ScalarValue::UInt32(Some(5))),
2304 Bound::Included(ScalarValue::UInt32(Some(10))),
2305 ),
2306 );
2307
2308 check_range(
2309 &index_info,
2310 "5 <= aisle AND 10 >= aisle",
2311 "aisle",
2312 SargableQuery::Range(
2313 Bound::Included(ScalarValue::UInt32(Some(5))),
2314 Bound::Included(ScalarValue::UInt32(Some(10))),
2315 ),
2316 );
2317
2318 check_range(
2319 &index_info,
2320 "10 >= aisle AND 5 <= aisle",
2321 "aisle",
2322 SargableQuery::Range(
2323 Bound::Included(ScalarValue::UInt32(Some(5))),
2324 Bound::Included(ScalarValue::UInt32(Some(10))),
2325 ),
2326 );
2327 check_simple(
2328 &index_info,
2329 "on_sale IS TRUE",
2330 "on_sale",
2331 SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2332 );
2333 check_simple(
2334 &index_info,
2335 "on_sale",
2336 "on_sale",
2337 SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2338 );
2339 check_simple_negated(
2340 &index_info,
2341 "NOT on_sale",
2342 "on_sale",
2343 SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2344 );
2345 check_simple(
2346 &index_info,
2347 "on_sale IS FALSE",
2348 "on_sale",
2349 SargableQuery::Equals(ScalarValue::Boolean(Some(false))),
2350 );
2351 check_simple_negated(
2352 &index_info,
2353 "aisle NOT BETWEEN 5 AND 10",
2354 "aisle",
2355 SargableQuery::Range(
2356 Bound::Included(ScalarValue::UInt32(Some(5))),
2357 Bound::Included(ScalarValue::UInt32(Some(10))),
2358 ),
2359 );
2360 check_simple(
2362 &index_info,
2363 "aisle IN (5, 6, 7)",
2364 "aisle",
2365 SargableQuery::IsIn(vec![
2366 ScalarValue::UInt32(Some(5)),
2367 ScalarValue::UInt32(Some(6)),
2368 ScalarValue::UInt32(Some(7)),
2369 ]),
2370 );
2371 check_simple_negated(
2372 &index_info,
2373 "NOT aisle IN (5, 6, 7)",
2374 "aisle",
2375 SargableQuery::IsIn(vec![
2376 ScalarValue::UInt32(Some(5)),
2377 ScalarValue::UInt32(Some(6)),
2378 ScalarValue::UInt32(Some(7)),
2379 ]),
2380 );
2381 check_simple_negated(
2382 &index_info,
2383 "aisle NOT IN (5, 6, 7)",
2384 "aisle",
2385 SargableQuery::IsIn(vec![
2386 ScalarValue::UInt32(Some(5)),
2387 ScalarValue::UInt32(Some(6)),
2388 ScalarValue::UInt32(Some(7)),
2389 ]),
2390 );
2391 check_simple(
2392 &index_info,
2393 "aisle IN (5, 6, 7, 8, 9)",
2394 "aisle",
2395 SargableQuery::IsIn(vec![
2396 ScalarValue::UInt32(Some(5)),
2397 ScalarValue::UInt32(Some(6)),
2398 ScalarValue::UInt32(Some(7)),
2399 ScalarValue::UInt32(Some(8)),
2400 ScalarValue::UInt32(Some(9)),
2401 ]),
2402 );
2403 check_simple_negated(
2404 &index_info,
2405 "NOT aisle IN (5, 6, 7, 8, 9)",
2406 "aisle",
2407 SargableQuery::IsIn(vec![
2408 ScalarValue::UInt32(Some(5)),
2409 ScalarValue::UInt32(Some(6)),
2410 ScalarValue::UInt32(Some(7)),
2411 ScalarValue::UInt32(Some(8)),
2412 ScalarValue::UInt32(Some(9)),
2413 ]),
2414 );
2415 check_simple_negated(
2416 &index_info,
2417 "aisle NOT IN (5, 6, 7, 8, 9)",
2418 "aisle",
2419 SargableQuery::IsIn(vec![
2420 ScalarValue::UInt32(Some(5)),
2421 ScalarValue::UInt32(Some(6)),
2422 ScalarValue::UInt32(Some(7)),
2423 ScalarValue::UInt32(Some(8)),
2424 ScalarValue::UInt32(Some(9)),
2425 ]),
2426 );
2427 check_simple(
2428 &index_info,
2429 "on_sale is false",
2430 "on_sale",
2431 SargableQuery::Equals(ScalarValue::Boolean(Some(false))),
2432 );
2433 check_simple(
2434 &index_info,
2435 "on_sale is true",
2436 "on_sale",
2437 SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2438 );
2439 check_simple(
2440 &index_info,
2441 "aisle < 10",
2442 "aisle",
2443 SargableQuery::Range(
2444 Bound::Unbounded,
2445 Bound::Excluded(ScalarValue::UInt32(Some(10))),
2446 ),
2447 );
2448 check_simple(
2449 &index_info,
2450 "aisle <= 10",
2451 "aisle",
2452 SargableQuery::Range(
2453 Bound::Unbounded,
2454 Bound::Included(ScalarValue::UInt32(Some(10))),
2455 ),
2456 );
2457 check_simple(
2458 &index_info,
2459 "aisle > 10",
2460 "aisle",
2461 SargableQuery::Range(
2462 Bound::Excluded(ScalarValue::UInt32(Some(10))),
2463 Bound::Unbounded,
2464 ),
2465 );
2466 check_no_index(&index_info, "10 > aisle");
2470 check_simple(
2471 &index_info,
2472 "aisle >= 10",
2473 "aisle",
2474 SargableQuery::Range(
2475 Bound::Included(ScalarValue::UInt32(Some(10))),
2476 Bound::Unbounded,
2477 ),
2478 );
2479 check_simple(
2480 &index_info,
2481 "aisle = 10",
2482 "aisle",
2483 SargableQuery::Equals(ScalarValue::UInt32(Some(10))),
2484 );
2485 check_simple_negated(
2486 &index_info,
2487 "aisle <> 10",
2488 "aisle",
2489 SargableQuery::Equals(ScalarValue::UInt32(Some(10))),
2490 );
2491 let left = Box::new(ScalarIndexExpr::Query(ScalarIndexSearch {
2493 column: "aisle".to_string(),
2494 index_name: "aisle_idx".to_string(),
2495 query: Arc::new(SargableQuery::Equals(ScalarValue::UInt32(Some(10)))),
2496 needs_recheck: false,
2497 }));
2498 let right = Box::new(ScalarIndexExpr::Query(ScalarIndexSearch {
2499 column: "color".to_string(),
2500 index_name: "color_idx".to_string(),
2501 query: Arc::new(SargableQuery::Equals(ScalarValue::Utf8(Some(
2502 "blue".to_string(),
2503 )))),
2504 needs_recheck: false,
2505 }));
2506 check(
2507 &index_info,
2508 "aisle = 10 AND color = 'blue'",
2509 Some(IndexedExpression {
2510 scalar_query: Some(ScalarIndexExpr::And(left.clone(), right.clone())),
2511 refine_expr: None,
2512 }),
2513 false,
2514 );
2515 let refine = Expr::Column(Column::new_unqualified("size")).gt(datafusion_expr::lit(30_i64));
2517 check(
2518 &index_info,
2519 "aisle = 10 AND color = 'blue' AND size > 30",
2520 Some(IndexedExpression {
2521 scalar_query: Some(ScalarIndexExpr::And(left.clone(), right.clone())),
2522 refine_expr: Some(refine.clone()),
2523 }),
2524 false,
2525 );
2526 check(
2528 &index_info,
2529 "aisle = 10 OR color = 'blue'",
2530 Some(IndexedExpression {
2531 scalar_query: Some(ScalarIndexExpr::Or(left.clone(), right.clone())),
2532 refine_expr: None,
2533 }),
2534 false,
2535 );
2536 check_no_index(&index_info, "aisle = 10 OR color = 'blue' OR size > 30");
2538 check(
2540 &index_info,
2541 "(aisle = 10 OR color = 'blue') AND size > 30",
2542 Some(IndexedExpression {
2543 scalar_query: Some(ScalarIndexExpr::Or(left, right)),
2544 refine_expr: Some(refine),
2545 }),
2546 false,
2547 );
2548 check_no_index(
2552 &index_info,
2553 "(aisle = 10 AND size > 30) OR (color = 'blue' AND size > 20)",
2554 );
2555
2556 check_no_index(&index_info, "aisle + 3 < 10");
2558
2559 check_no_index(&index_info, "aisle IN (5, 6, NULL)");
2564 check_no_index(&index_info, "aisle = 5 OR aisle = 6 OR NULL");
2567 check_no_index(&index_info, "aisle IN (5, 6, 7, 8, NULL)");
2568 check_no_index(&index_info, "aisle = NULL");
2569 check_no_index(&index_info, "aisle BETWEEN 5 AND NULL");
2570 check_no_index(&index_info, "aisle BETWEEN NULL AND 10");
2571 }
2572
2573 #[tokio::test]
2574 async fn test_not_flips_certainty() {
2575 use lance_core::utils::mask::{NullableRowAddrSet, RowAddrTreeMap};
2576
2577 fn apply_not(result: NullableIndexExprResult) -> NullableIndexExprResult {
2582 match result {
2583 NullableIndexExprResult::Exact(mask) => NullableIndexExprResult::Exact(!mask),
2584 NullableIndexExprResult::AtMost(mask) => NullableIndexExprResult::AtLeast(!mask),
2585 NullableIndexExprResult::AtLeast(mask) => NullableIndexExprResult::AtMost(!mask),
2586 }
2587 }
2588
2589 let at_most = NullableIndexExprResult::AtMost(NullableRowAddrMask::AllowList(
2591 NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2592 ));
2593 assert!(matches!(
2595 apply_not(at_most),
2596 NullableIndexExprResult::AtLeast(_)
2597 ));
2598
2599 let at_least = NullableIndexExprResult::AtLeast(NullableRowAddrMask::AllowList(
2601 NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2602 ));
2603 assert!(matches!(
2605 apply_not(at_least),
2606 NullableIndexExprResult::AtMost(_)
2607 ));
2608
2609 let exact = NullableIndexExprResult::Exact(NullableRowAddrMask::AllowList(
2611 NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2612 ));
2613 assert!(matches!(
2614 apply_not(exact),
2615 NullableIndexExprResult::Exact(_)
2616 ));
2617 }
2618
2619 #[tokio::test]
2620 async fn test_and_or_preserve_certainty() {
2621 use lance_core::utils::mask::{NullableRowAddrSet, RowAddrTreeMap};
2622
2623 let make_at_most = || {
2625 NullableIndexExprResult::AtMost(NullableRowAddrMask::AllowList(
2626 NullableRowAddrSet::new(
2627 RowAddrTreeMap::from_iter(&[1, 2, 3]),
2628 RowAddrTreeMap::new(),
2629 ),
2630 ))
2631 };
2632
2633 let make_at_least = || {
2634 NullableIndexExprResult::AtLeast(NullableRowAddrMask::AllowList(
2635 NullableRowAddrSet::new(
2636 RowAddrTreeMap::from_iter(&[2, 3, 4]),
2637 RowAddrTreeMap::new(),
2638 ),
2639 ))
2640 };
2641
2642 let make_exact = || {
2643 NullableIndexExprResult::Exact(NullableRowAddrMask::AllowList(NullableRowAddrSet::new(
2644 RowAddrTreeMap::from_iter(&[1, 2]),
2645 RowAddrTreeMap::new(),
2646 )))
2647 };
2648
2649 assert!(matches!(
2651 make_at_most() & make_at_most(),
2652 NullableIndexExprResult::AtMost(_)
2653 ));
2654
2655 assert!(matches!(
2657 make_at_least() & make_at_least(),
2658 NullableIndexExprResult::AtLeast(_)
2659 ));
2660
2661 assert!(matches!(
2663 make_at_most() & make_at_least(),
2664 NullableIndexExprResult::AtMost(_)
2665 ));
2666
2667 assert!(matches!(
2669 make_at_most() | make_at_most(),
2670 NullableIndexExprResult::AtMost(_)
2671 ));
2672
2673 assert!(matches!(
2675 make_at_least() | make_at_least(),
2676 NullableIndexExprResult::AtLeast(_)
2677 ));
2678
2679 assert!(matches!(
2681 make_at_most() | make_at_least(),
2682 NullableIndexExprResult::AtLeast(_)
2683 ));
2684
2685 assert!(matches!(
2687 make_exact() & make_at_most(),
2688 NullableIndexExprResult::AtMost(_)
2689 ));
2690
2691 assert!(matches!(
2693 make_exact() | make_at_least(),
2694 NullableIndexExprResult::AtLeast(_)
2695 ));
2696 }
2697
2698 #[test]
2699 fn test_extract_like_leading_prefix() {
2700 assert_eq!(
2702 extract_like_leading_prefix("foo%", None),
2703 Some(("foo".to_string(), false))
2704 );
2705 assert_eq!(
2706 extract_like_leading_prefix("abc%", None),
2707 Some(("abc".to_string(), false))
2708 );
2709
2710 assert_eq!(
2712 extract_like_leading_prefix("foo%bar%", None),
2713 Some(("foo".to_string(), true))
2714 );
2715 assert_eq!(
2716 extract_like_leading_prefix("foo_bar%", None),
2717 Some(("foo".to_string(), true))
2718 );
2719 assert_eq!(
2720 extract_like_leading_prefix("foo%bar", None),
2721 Some(("foo".to_string(), true))
2722 );
2723 assert_eq!(
2724 extract_like_leading_prefix("foo_", None),
2725 Some(("foo".to_string(), true))
2726 );
2727
2728 assert_eq!(extract_like_leading_prefix("%foo", None), None);
2730 assert_eq!(extract_like_leading_prefix("_foo%", None), None);
2731 assert_eq!(extract_like_leading_prefix("%", None), None);
2732
2733 assert_eq!(extract_like_leading_prefix("foo", None), None);
2735
2736 assert_eq!(
2738 extract_like_leading_prefix(r"foo\%bar%", Some('\\')),
2739 Some(("foo%bar".to_string(), false))
2740 );
2741 assert_eq!(
2742 extract_like_leading_prefix(r"foo\_bar%", Some('\\')),
2743 Some(("foo_bar".to_string(), false))
2744 );
2745 assert_eq!(
2746 extract_like_leading_prefix(r"foo\\bar%", Some('\\')),
2747 Some(("foo\\bar".to_string(), false))
2748 );
2749
2750 assert_eq!(extract_like_leading_prefix(r"foo\%", Some('\\')), None);
2752
2753 assert_eq!(extract_like_leading_prefix(r"foo\%", None), None);
2756 assert_eq!(
2758 extract_like_leading_prefix(r"foo\bar%", None),
2759 Some(("foo\\bar".to_string(), false))
2760 );
2761
2762 assert_eq!(extract_like_leading_prefix("", None), None);
2764
2765 assert_eq!(
2767 extract_like_leading_prefix(r"foo\%bar%baz%", Some('\\')),
2768 Some(("foo%bar".to_string(), true))
2769 );
2770 }
2771
2772 #[test]
2773 fn test_like_expression_parsing() {
2774 let index_info = MockIndexInfoProvider::new(vec![(
2777 "color",
2778 ColInfo::new(
2779 DataType::Utf8,
2780 Box::new(SargableQueryParser::new("color_idx".to_string(), false)),
2781 ),
2782 )]);
2783
2784 let schema = Schema::new(vec![Field::new("color", DataType::Utf8, false)]);
2786 let df_schema: DFSchema = schema.try_into().unwrap();
2787 let ctx = get_session_context(&LanceExecutionOptions::default());
2788 let state = ctx.state();
2789
2790 let expr = state
2791 .create_logical_expr("color LIKE 'foo%'", &df_schema)
2792 .unwrap();
2793 let result = apply_scalar_indices(expr, &index_info).unwrap();
2794
2795 assert!(result.scalar_query.is_some(), "Should have scalar_query");
2796 assert!(
2797 result.refine_expr.is_none(),
2798 "Simple prefix should not need refine_expr"
2799 );
2800
2801 if let Some(ScalarIndexExpr::Query(search)) = &result.scalar_query {
2803 let query = search.query.as_any().downcast_ref::<SargableQuery>();
2804 assert!(query.is_some(), "Query should be SargableQuery");
2805 match query.unwrap() {
2806 SargableQuery::LikePrefix(prefix) => {
2807 assert_eq!(prefix, &ScalarValue::Utf8(Some("foo".to_string())));
2808 }
2809 _ => panic!("Expected LikePrefix query"),
2810 }
2811 } else {
2812 panic!("Expected Query variant");
2813 }
2814
2815 let expr = state
2817 .create_logical_expr("color LIKE 'foo%bar%'", &df_schema)
2818 .unwrap();
2819 let result = apply_scalar_indices(expr, &index_info).unwrap();
2820
2821 assert!(result.scalar_query.is_some(), "Should have scalar_query");
2822 assert!(
2823 result.refine_expr.is_some(),
2824 "Complex pattern should have refine_expr"
2825 );
2826
2827 if let Some(ScalarIndexExpr::Query(search)) = &result.scalar_query {
2829 let query = search.query.as_any().downcast_ref::<SargableQuery>();
2830 assert!(query.is_some(), "Query should be SargableQuery");
2831 match query.unwrap() {
2832 SargableQuery::LikePrefix(prefix) => {
2833 assert_eq!(prefix, &ScalarValue::Utf8(Some("foo".to_string())));
2834 }
2835 _ => panic!("Expected LikePrefix query"),
2836 }
2837 }
2838
2839 let refine = result.refine_expr.unwrap();
2841 match refine {
2842 Expr::Like(like) => {
2843 assert!(!like.negated);
2844 assert!(!like.case_insensitive);
2845 if let Expr::Literal(ScalarValue::Utf8(Some(pattern)), _) = like.pattern.as_ref() {
2846 assert_eq!(pattern, "foo%bar%");
2847 } else {
2848 panic!("Expected Utf8 literal pattern");
2849 }
2850 }
2851 _ => panic!("Expected Like expression in refine_expr"),
2852 }
2853
2854 let expr = state
2856 .create_logical_expr("color LIKE '%foo'", &df_schema)
2857 .unwrap();
2858 let result = apply_scalar_indices(expr, &index_info).unwrap();
2859
2860 assert!(
2861 result.scalar_query.is_none(),
2862 "Pattern starting with wildcard should not use index"
2863 );
2864 assert!(result.refine_expr.is_some(), "Should fall back to refine");
2865 }
2866
2867 #[test]
2868 fn test_starts_with_with_underscore_after_optimization() {
2869 let index_info = MockIndexInfoProvider::new(vec![(
2873 "object_id",
2874 ColInfo::new(
2875 DataType::Utf8,
2876 Box::new(SargableQueryParser::new("object_id_idx".to_string(), false)),
2877 ),
2878 )]);
2879
2880 let schema = Schema::new(vec![Field::new("object_id", DataType::Utf8, false)]);
2881 let df_schema: DFSchema = schema.try_into().unwrap();
2882 let ctx = get_session_context(&LanceExecutionOptions::default());
2883 let state = ctx.state();
2884
2885 let expr = state
2887 .create_logical_expr("starts_with(object_id, 'test_ns$')", &df_schema)
2888 .unwrap();
2889
2890 let props = ExecutionProps::new().with_query_execution_start_time(Utc::now());
2892 let simplify_context = SimplifyContext::new(&props).with_schema(Arc::new(df_schema));
2893 let simplifier =
2894 datafusion::optimizer::simplify_expressions::ExprSimplifier::new(simplify_context);
2895 let simplified_expr = simplifier.simplify(expr).unwrap();
2896
2897 let result = apply_scalar_indices(simplified_expr, &index_info).unwrap();
2899
2900 if let Some(ScalarIndexExpr::Query(search)) = &result.scalar_query {
2903 let query = search
2904 .query
2905 .as_any()
2906 .downcast_ref::<SargableQuery>()
2907 .unwrap();
2908 match query {
2909 SargableQuery::LikePrefix(prefix) => {
2910 let prefix_str = match prefix {
2911 ScalarValue::Utf8(Some(s)) => s.clone(),
2912 _ => panic!("Expected Utf8 prefix"),
2913 };
2914 assert_eq!(
2916 prefix_str, "test_ns$",
2917 "Prefix should be 'test_ns$', not 'test' (underscore should not be a wildcard)"
2918 );
2919 }
2920 _ => panic!("Expected LikePrefix query"),
2921 }
2922 } else {
2923 panic!("Expected scalar_query to be present");
2925 }
2926 }
2927
2928 #[test]
2929 fn test_starts_with_to_like_conversion() {
2930 let index_info = MockIndexInfoProvider::new(vec![(
2932 "color",
2933 ColInfo::new(
2934 DataType::Utf8,
2935 Box::new(SargableQueryParser::new("color_idx".to_string(), false)),
2936 ),
2937 )]);
2938
2939 let schema = Schema::new(vec![Field::new("color", DataType::Utf8, false)]);
2940 let df_schema: DFSchema = schema.try_into().unwrap();
2941 let ctx = get_session_context(&LanceExecutionOptions::default());
2942 let state = ctx.state();
2943
2944 let expr = state
2946 .create_logical_expr("starts_with(color, 'foo')", &df_schema)
2947 .unwrap();
2948 let result = apply_scalar_indices(expr, &index_info).unwrap();
2949
2950 assert!(
2951 result.scalar_query.is_some(),
2952 "starts_with should use index"
2953 );
2954 assert!(
2955 result.refine_expr.is_none(),
2956 "Pure prefix starts_with should not need refine_expr"
2957 );
2958
2959 if let Some(ScalarIndexExpr::Query(search)) = &result.scalar_query {
2961 let query = search.query.as_any().downcast_ref::<SargableQuery>();
2962 assert!(query.is_some(), "Query should be SargableQuery");
2963 match query.unwrap() {
2964 SargableQuery::LikePrefix(prefix) => {
2965 assert_eq!(prefix, &ScalarValue::Utf8(Some("foo".to_string())));
2966 }
2967 _ => panic!("Expected LikePrefix query"),
2968 }
2969 } else {
2970 panic!("Expected Query variant");
2971 }
2972
2973 let like_expr = state
2975 .create_logical_expr("color LIKE 'foo%'", &df_schema)
2976 .unwrap();
2977 let like_result = apply_scalar_indices(like_expr, &index_info).unwrap();
2978
2979 if let (
2981 Some(ScalarIndexExpr::Query(starts_with_search)),
2982 Some(ScalarIndexExpr::Query(like_search)),
2983 ) = (&result.scalar_query, &like_result.scalar_query)
2984 {
2985 let sw_query = starts_with_search
2986 .query
2987 .as_any()
2988 .downcast_ref::<SargableQuery>()
2989 .unwrap();
2990 let like_query = like_search
2991 .query
2992 .as_any()
2993 .downcast_ref::<SargableQuery>()
2994 .unwrap();
2995 assert_eq!(
2996 sw_query, like_query,
2997 "starts_with and LIKE 'prefix%' should produce identical queries"
2998 );
2999 }
3000 }
3001}