Skip to main content

lance_index/scalar/
expression.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4use 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, 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/// An indexed expression consists of a scalar index query with a post-scan filter
38///
39/// When a user wants to filter the data returned by a scan we may be able to use
40/// one or more scalar indices to reduce the amount of data we load from the disk.
41///
42/// For example, if a user provides the filter "x = 7", and we have a scalar index
43/// on x, then we can possibly identify the exact row that the user desires with our
44/// index.  A full-table scan can then turn into a take operation fetching the rows
45/// desired.  This would create an IndexedExpression with a scalar_query but no
46/// refine.
47///
48/// If the user asked for "type = 'dog' && z = 3" and we had a scalar index on the
49/// "type" column then we could convert this to an indexed scan for "type='dog'"
50/// followed by an in-memory filter for z=3.  This would create an IndexedExpression
51/// with both a scalar_query AND a refine.
52///
53/// Finally, if the user asked for "z = 3" and we do not have a scalar index on the
54/// "z" column then we must fallback to an IndexedExpression with no scalar_query and
55/// only a refine.
56///
57/// Two IndexedExpressions can be AND'd together.  Each part is AND'd together.
58/// Two IndexedExpressions cannot be OR'd together unless both are scalar_query only
59///   or both are refine only
60/// An IndexedExpression cannot be negated if it has both a refine and a scalar_query
61///
62/// When an operation cannot be performed we fallback to the original expression-only
63/// representation
64#[derive(Debug, PartialEq)]
65pub struct IndexedExpression {
66    /// The portion of the query that can be satisfied by scalar indices
67    pub scalar_query: Option<ScalarIndexExpr>,
68    /// The portion of the query that cannot be satisfied by scalar indices
69    pub refine_expr: Option<Expr>,
70}
71
72pub trait ScalarQueryParser: std::fmt::Debug + Send + Sync {
73    /// Visit a between expression
74    ///
75    /// Returns an IndexedExpression if the index can accelerate between expressions
76    fn visit_between(
77        &self,
78        column: &str,
79        low: &Bound<ScalarValue>,
80        high: &Bound<ScalarValue>,
81    ) -> Option<IndexedExpression>;
82    /// Visit an in list expression
83    ///
84    /// Returns an IndexedExpression if the index can accelerate in list expressions
85    fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression>;
86    /// Visit an is bool expression
87    ///
88    /// Returns an IndexedExpression if the index can accelerate is bool expressions
89    fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression>;
90    /// Visit an is null expression
91    ///
92    /// Returns an IndexedExpression if the index can accelerate is null expressions
93    fn visit_is_null(&self, column: &str) -> Option<IndexedExpression>;
94    /// Visit a comparison expression
95    ///
96    /// Returns an IndexedExpression if the index can accelerate comparison expressions
97    fn visit_comparison(
98        &self,
99        column: &str,
100        value: &ScalarValue,
101        op: &Operator,
102    ) -> Option<IndexedExpression>;
103    /// Visit a scalar function expression
104    ///
105    /// Returns an IndexedExpression if the index can accelerate the given scalar function.
106    /// For example, an ngram index can accelerate the contains function.
107    fn visit_scalar_function(
108        &self,
109        column: &str,
110        data_type: &DataType,
111        func: &ScalarUDF,
112        args: &[Expr],
113    ) -> Option<IndexedExpression>;
114
115    /// Visits a potential reference to a column
116    ///
117    /// This function is a little different from the other visitors.  It is used to test if a potential
118    /// column reference is a reference the index handles.
119    ///
120    /// Most indexes are designed to run on references to the indexed column.  For example, if a query
121    /// is "x = 7" and we have a scalar index on "x" then we apply the index to the "x" column reference.
122    ///
123    /// However, some indexes are designed to run on projections of the indexed column.  For example,
124    /// if a query is "json_extract(json, '$.name') = 'books'" and we have a JSON index on the "json" column
125    /// then we apply the index to the projection of the "json" column.
126    ///
127    /// This function is used to test if a potential column reference is a reference the index handles.
128    /// The default implementation matches column references but this can be overridden by indexes that
129    /// handle projections.
130    ///
131    /// The function is also passed in the data type of the column and should return the data type of the
132    /// reference.  Normally this is the same as the input for a direct column reference and possibly something
133    /// different for a projection.  E.g. a JSON column (LargeBinary) might be projected to a string or float
134    ///
135    /// Note: higher logic in the expression parser already limits references to either Expr::Column or Expr::ScalarFunction
136    /// where the first argument is an Expr::Column.  If your projection doesn't fit that mold then the
137    /// expression parser will need to be modified.
138    fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
139        match func {
140            Expr::Column(_) => Some(data_type.clone()),
141            _ => None,
142        }
143    }
144}
145
146/// A generic parser that wraps multiple scalar query parsers
147///
148/// It will search each parser in order and return the first non-None result
149#[derive(Debug)]
150pub struct MultiQueryParser {
151    parsers: Vec<Box<dyn ScalarQueryParser>>,
152}
153
154impl MultiQueryParser {
155    /// Create a new MultiQueryParser with a single parser
156    pub fn single(parser: Box<dyn ScalarQueryParser>) -> Self {
157        Self {
158            parsers: vec![parser],
159        }
160    }
161
162    /// Add a new parser to the MultiQueryParser
163    pub fn add(&mut self, other: Box<dyn ScalarQueryParser>) {
164        self.parsers.push(other);
165    }
166}
167
168impl ScalarQueryParser for MultiQueryParser {
169    fn visit_between(
170        &self,
171        column: &str,
172        low: &Bound<ScalarValue>,
173        high: &Bound<ScalarValue>,
174    ) -> Option<IndexedExpression> {
175        self.parsers
176            .iter()
177            .find_map(|parser| parser.visit_between(column, low, high))
178    }
179    fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
180        self.parsers
181            .iter()
182            .find_map(|parser| parser.visit_in_list(column, in_list))
183    }
184    fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
185        self.parsers
186            .iter()
187            .find_map(|parser| parser.visit_is_bool(column, value))
188    }
189    fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
190        self.parsers
191            .iter()
192            .find_map(|parser| parser.visit_is_null(column))
193    }
194    fn visit_comparison(
195        &self,
196        column: &str,
197        value: &ScalarValue,
198        op: &Operator,
199    ) -> Option<IndexedExpression> {
200        self.parsers
201            .iter()
202            .find_map(|parser| parser.visit_comparison(column, value, op))
203    }
204    fn visit_scalar_function(
205        &self,
206        column: &str,
207        data_type: &DataType,
208        func: &ScalarUDF,
209        args: &[Expr],
210    ) -> Option<IndexedExpression> {
211        self.parsers
212            .iter()
213            .find_map(|parser| parser.visit_scalar_function(column, data_type, func, args))
214    }
215    /// TODO(low-priority): This is maybe not quite right.  We should filter down the list of parsers based
216    /// on those that consider the reference valid.  Instead what we are doing is checking all parsers if any one
217    /// parser considers the reference valid.
218    ///
219    /// This will be a problem if the user creates two indexes (e.g. btree and json) on the same column and those two
220    /// indexes have different reference schemes.
221    fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
222        self.parsers
223            .iter()
224            .find_map(|parser| parser.is_valid_reference(func, data_type))
225    }
226}
227
228/// A parser for indices that handle SARGable queries
229#[derive(Debug)]
230pub struct SargableQueryParser {
231    index_name: String,
232    needs_recheck: bool,
233}
234
235impl SargableQueryParser {
236    pub fn new(index_name: String, needs_recheck: bool) -> Self {
237        Self {
238            index_name,
239            needs_recheck,
240        }
241    }
242}
243
244impl ScalarQueryParser for SargableQueryParser {
245    fn is_valid_reference(&self, func: &Expr, data_type: &DataType) -> Option<DataType> {
246        match func {
247            Expr::Column(_) => Some(data_type.clone()),
248            // Also accept get_field expressions for nested field access
249            Expr::ScalarFunction(udf) if udf.name() == "get_field" => Some(data_type.clone()),
250            _ => None,
251        }
252    }
253
254    fn visit_between(
255        &self,
256        column: &str,
257        low: &Bound<ScalarValue>,
258        high: &Bound<ScalarValue>,
259    ) -> Option<IndexedExpression> {
260        if let Bound::Included(val) | Bound::Excluded(val) = low
261            && val.is_null()
262        {
263            return None;
264        }
265        if let Bound::Included(val) | Bound::Excluded(val) = high
266            && val.is_null()
267        {
268            return None;
269        }
270        let query = SargableQuery::Range(low.clone(), high.clone());
271        Some(IndexedExpression::index_query_with_recheck(
272            column.to_string(),
273            self.index_name.clone(),
274            Arc::new(query),
275            self.needs_recheck,
276        ))
277    }
278
279    fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
280        if in_list.iter().any(|val| val.is_null()) {
281            return None;
282        }
283        let query = SargableQuery::IsIn(in_list.to_vec());
284        Some(IndexedExpression::index_query_with_recheck(
285            column.to_string(),
286            self.index_name.clone(),
287            Arc::new(query),
288            self.needs_recheck,
289        ))
290    }
291
292    fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
293        Some(IndexedExpression::index_query_with_recheck(
294            column.to_string(),
295            self.index_name.clone(),
296            Arc::new(SargableQuery::Equals(ScalarValue::Boolean(Some(value)))),
297            self.needs_recheck,
298        ))
299    }
300
301    fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
302        Some(IndexedExpression::index_query_with_recheck(
303            column.to_string(),
304            self.index_name.clone(),
305            Arc::new(SargableQuery::IsNull()),
306            self.needs_recheck,
307        ))
308    }
309
310    fn visit_comparison(
311        &self,
312        column: &str,
313        value: &ScalarValue,
314        op: &Operator,
315    ) -> Option<IndexedExpression> {
316        if value.is_null() {
317            return None;
318        }
319        let query = match op {
320            Operator::Lt => SargableQuery::Range(Bound::Unbounded, Bound::Excluded(value.clone())),
321            Operator::LtEq => {
322                SargableQuery::Range(Bound::Unbounded, Bound::Included(value.clone()))
323            }
324            Operator::Gt => SargableQuery::Range(Bound::Excluded(value.clone()), Bound::Unbounded),
325            Operator::GtEq => {
326                SargableQuery::Range(Bound::Included(value.clone()), Bound::Unbounded)
327            }
328            Operator::Eq => SargableQuery::Equals(value.clone()),
329            // This will be negated by the caller
330            Operator::NotEq => SargableQuery::Equals(value.clone()),
331            _ => unreachable!(),
332        };
333        Some(IndexedExpression::index_query_with_recheck(
334            column.to_string(),
335            self.index_name.clone(),
336            Arc::new(query),
337            self.needs_recheck,
338        ))
339    }
340
341    fn visit_scalar_function(
342        &self,
343        _: &str,
344        _: &DataType,
345        _: &ScalarUDF,
346        _: &[Expr],
347    ) -> Option<IndexedExpression> {
348        None
349    }
350}
351
352/// A parser for bloom filter indices that only support equals, is_null, and is_in operations
353#[derive(Debug)]
354pub struct BloomFilterQueryParser {
355    index_name: String,
356    needs_recheck: bool,
357}
358
359impl BloomFilterQueryParser {
360    pub fn new(index_name: String, needs_recheck: bool) -> Self {
361        Self {
362            index_name,
363            needs_recheck,
364        }
365    }
366}
367
368impl ScalarQueryParser for BloomFilterQueryParser {
369    fn visit_between(
370        &self,
371        _: &str,
372        _: &Bound<ScalarValue>,
373        _: &Bound<ScalarValue>,
374    ) -> Option<IndexedExpression> {
375        // Bloom filters don't support range queries
376        None
377    }
378
379    fn visit_in_list(&self, column: &str, in_list: &[ScalarValue]) -> Option<IndexedExpression> {
380        let query = BloomFilterQuery::IsIn(in_list.to_vec());
381        Some(IndexedExpression::index_query_with_recheck(
382            column.to_string(),
383            self.index_name.clone(),
384            Arc::new(query),
385            self.needs_recheck,
386        ))
387    }
388
389    fn visit_is_bool(&self, column: &str, value: bool) -> Option<IndexedExpression> {
390        Some(IndexedExpression::index_query_with_recheck(
391            column.to_string(),
392            self.index_name.clone(),
393            Arc::new(BloomFilterQuery::Equals(ScalarValue::Boolean(Some(value)))),
394            self.needs_recheck,
395        ))
396    }
397
398    fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
399        Some(IndexedExpression::index_query_with_recheck(
400            column.to_string(),
401            self.index_name.clone(),
402            Arc::new(BloomFilterQuery::IsNull()),
403            self.needs_recheck,
404        ))
405    }
406
407    fn visit_comparison(
408        &self,
409        column: &str,
410        value: &ScalarValue,
411        op: &Operator,
412    ) -> Option<IndexedExpression> {
413        let query = match op {
414            // Bloom filters only support equality comparisons
415            Operator::Eq => BloomFilterQuery::Equals(value.clone()),
416            // This will be negated by the caller
417            Operator::NotEq => BloomFilterQuery::Equals(value.clone()),
418            // Bloom filters don't support range operations
419            _ => return None,
420        };
421        Some(IndexedExpression::index_query_with_recheck(
422            column.to_string(),
423            self.index_name.clone(),
424            Arc::new(query),
425            self.needs_recheck,
426        ))
427    }
428
429    fn visit_scalar_function(
430        &self,
431        _: &str,
432        _: &DataType,
433        _: &ScalarUDF,
434        _: &[Expr],
435    ) -> Option<IndexedExpression> {
436        // Bloom filters don't support scalar functions
437        None
438    }
439}
440
441/// A parser for indices that handle label list queries
442#[derive(Debug)]
443pub struct LabelListQueryParser {
444    index_name: String,
445}
446
447impl LabelListQueryParser {
448    pub fn new(index_name: String) -> Self {
449        Self { index_name }
450    }
451}
452
453impl ScalarQueryParser for LabelListQueryParser {
454    fn visit_between(
455        &self,
456        _: &str,
457        _: &Bound<ScalarValue>,
458        _: &Bound<ScalarValue>,
459    ) -> Option<IndexedExpression> {
460        None
461    }
462
463    fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
464        None
465    }
466
467    fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
468        None
469    }
470
471    fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
472        None
473    }
474
475    fn visit_comparison(
476        &self,
477        _: &str,
478        _: &ScalarValue,
479        _: &Operator,
480    ) -> Option<IndexedExpression> {
481        None
482    }
483
484    fn visit_scalar_function(
485        &self,
486        column: &str,
487        data_type: &DataType,
488        func: &ScalarUDF,
489        args: &[Expr],
490    ) -> Option<IndexedExpression> {
491        if args.len() != 2 {
492            return None;
493        }
494        // DataFusion normalizes array_contains to array_has
495        if func.name() == "array_has" {
496            let inner_type = match data_type {
497                DataType::List(field) | DataType::LargeList(field) => field.data_type(),
498                _ => return None,
499            };
500            let scalar = maybe_scalar(&args[1], inner_type)?;
501            // array_has(..., NULL) returns no matches in datafusion, but the index would
502            // match rows containing NULL. Fallback to match datafusion behavior.
503            if scalar.is_null() {
504                return None;
505            }
506            let query = LabelListQuery::HasAnyLabel(vec![scalar]);
507            return Some(IndexedExpression::index_query(
508                column.to_string(),
509                self.index_name.clone(),
510                Arc::new(query),
511            ));
512        }
513
514        let label_list = maybe_scalar(&args[1], data_type)?;
515        if let ScalarValue::List(list_arr) = label_list {
516            let list_values = list_arr.values();
517            if list_values.is_empty() {
518                return None;
519            }
520            let mut scalars = Vec::with_capacity(list_values.len());
521            for idx in 0..list_values.len() {
522                scalars.push(ScalarValue::try_from_array(list_values.as_ref(), idx).ok()?);
523            }
524            if func.name() == "array_has_all" {
525                let query = LabelListQuery::HasAllLabels(scalars);
526                Some(IndexedExpression::index_query(
527                    column.to_string(),
528                    self.index_name.clone(),
529                    Arc::new(query),
530                ))
531            } else if func.name() == "array_has_any" {
532                let query = LabelListQuery::HasAnyLabel(scalars);
533                Some(IndexedExpression::index_query(
534                    column.to_string(),
535                    self.index_name.clone(),
536                    Arc::new(query),
537                ))
538            } else {
539                None
540            }
541        } else {
542            None
543        }
544    }
545}
546
547/// A parser for indices that handle string contains queries
548#[derive(Debug, Clone)]
549pub struct TextQueryParser {
550    index_name: String,
551    needs_recheck: bool,
552}
553
554impl TextQueryParser {
555    pub fn new(index_name: String, needs_recheck: bool) -> Self {
556        Self {
557            index_name,
558            needs_recheck,
559        }
560    }
561}
562
563impl ScalarQueryParser for TextQueryParser {
564    fn visit_between(
565        &self,
566        _: &str,
567        _: &Bound<ScalarValue>,
568        _: &Bound<ScalarValue>,
569    ) -> Option<IndexedExpression> {
570        None
571    }
572
573    fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
574        None
575    }
576
577    fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
578        None
579    }
580
581    fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
582        None
583    }
584
585    fn visit_comparison(
586        &self,
587        _: &str,
588        _: &ScalarValue,
589        _: &Operator,
590    ) -> Option<IndexedExpression> {
591        None
592    }
593
594    fn visit_scalar_function(
595        &self,
596        column: &str,
597        data_type: &DataType,
598        func: &ScalarUDF,
599        args: &[Expr],
600    ) -> Option<IndexedExpression> {
601        if args.len() != 2 {
602            return None;
603        }
604        let scalar = maybe_scalar(&args[1], data_type)?;
605        match scalar {
606            ScalarValue::Utf8(Some(scalar_str)) | ScalarValue::LargeUtf8(Some(scalar_str)) => {
607                if func.name() == "contains" {
608                    let query = TextQuery::StringContains(scalar_str);
609                    Some(IndexedExpression::index_query_with_recheck(
610                        column.to_string(),
611                        self.index_name.clone(),
612                        Arc::new(query),
613                        self.needs_recheck,
614                    ))
615                } else {
616                    None
617                }
618            }
619            _ => {
620                // If the scalar is not a string, we cannot handle it
621                None
622            }
623        }
624    }
625}
626
627/// A parser for indices that handle queries with the contains_tokens function
628#[derive(Debug, Clone)]
629pub struct FtsQueryParser {
630    index_name: String,
631}
632
633impl FtsQueryParser {
634    pub fn new(name: String) -> Self {
635        Self { index_name: name }
636    }
637}
638
639impl ScalarQueryParser for FtsQueryParser {
640    fn visit_between(
641        &self,
642        _: &str,
643        _: &Bound<ScalarValue>,
644        _: &Bound<ScalarValue>,
645    ) -> Option<IndexedExpression> {
646        None
647    }
648
649    fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
650        None
651    }
652
653    fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
654        None
655    }
656
657    fn visit_is_null(&self, _: &str) -> Option<IndexedExpression> {
658        None
659    }
660
661    fn visit_comparison(
662        &self,
663        _: &str,
664        _: &ScalarValue,
665        _: &Operator,
666    ) -> Option<IndexedExpression> {
667        None
668    }
669
670    fn visit_scalar_function(
671        &self,
672        column: &str,
673        data_type: &DataType,
674        func: &ScalarUDF,
675        args: &[Expr],
676    ) -> Option<IndexedExpression> {
677        if args.len() != 2 {
678            return None;
679        }
680        let scalar = maybe_scalar(&args[1], data_type)?;
681        if let ScalarValue::Utf8(Some(scalar_str)) = scalar
682            && func.name() == "contains_tokens"
683        {
684            let query = TokenQuery::TokensContains(scalar_str);
685            return Some(IndexedExpression::index_query(
686                column.to_string(),
687                self.index_name.clone(),
688                Arc::new(query),
689            ));
690        }
691        None
692    }
693}
694
695/// A parser for geo indices that handles spatial queries
696#[cfg(feature = "geo")]
697#[derive(Debug, Clone)]
698pub struct GeoQueryParser {
699    index_name: String,
700}
701
702#[cfg(feature = "geo")]
703impl GeoQueryParser {
704    pub fn new(index_name: String) -> Self {
705        Self { index_name }
706    }
707}
708
709#[cfg(feature = "geo")]
710impl ScalarQueryParser for GeoQueryParser {
711    fn visit_between(
712        &self,
713        _: &str,
714        _: &Bound<ScalarValue>,
715        _: &Bound<ScalarValue>,
716    ) -> Option<IndexedExpression> {
717        None
718    }
719
720    fn visit_in_list(&self, _: &str, _: &[ScalarValue]) -> Option<IndexedExpression> {
721        None
722    }
723
724    fn visit_is_bool(&self, _: &str, _: bool) -> Option<IndexedExpression> {
725        None
726    }
727
728    fn visit_is_null(&self, column: &str) -> Option<IndexedExpression> {
729        Some(IndexedExpression::index_query_with_recheck(
730            column.to_string(),
731            self.index_name.clone(),
732            Arc::new(GeoQuery::IsNull),
733            true,
734        ))
735    }
736
737    fn visit_comparison(
738        &self,
739        _: &str,
740        _: &ScalarValue,
741        _: &Operator,
742    ) -> Option<IndexedExpression> {
743        None
744    }
745
746    fn visit_scalar_function(
747        &self,
748        column: &str,
749        _data_type: &DataType,
750        func: &ScalarUDF,
751        args: &[Expr],
752    ) -> Option<IndexedExpression> {
753        if (func.name() == "st_intersects"
754            || func.name() == "st_contains"
755            || func.name() == "st_within"
756            || func.name() == "st_touches"
757            || func.name() == "st_crosses"
758            || func.name() == "st_overlaps"
759            || func.name() == "st_covers"
760            || func.name() == "st_coveredby")
761            && args.len() == 2
762        {
763            let left_arg = &args[0];
764            let right_arg = &args[1];
765            return match (left_arg, right_arg) {
766                (Expr::Literal(left_value, metadata), Expr::Column(_)) => {
767                    let mut field = Field::new("_geo", left_value.data_type(), false);
768                    if let Some(metadata) = metadata {
769                        field = field.with_metadata(metadata.to_hashmap());
770                    }
771                    let query = GeoQuery::IntersectQuery(RelationQuery {
772                        value: left_value.clone(),
773                        field,
774                    });
775                    Some(IndexedExpression::index_query_with_recheck(
776                        column.to_string(),
777                        self.index_name.clone(),
778                        Arc::new(query),
779                        true,
780                    ))
781                }
782                (Expr::Column(_), Expr::Literal(right_value, metadata)) => {
783                    let mut field = Field::new("_geo", right_value.data_type(), false);
784                    if let Some(metadata) = metadata {
785                        field = field.with_metadata(metadata.to_hashmap());
786                    }
787                    let query = GeoQuery::IntersectQuery(RelationQuery {
788                        value: right_value.clone(),
789                        field,
790                    });
791                    Some(IndexedExpression::index_query_with_recheck(
792                        column.to_string(),
793                        self.index_name.clone(),
794                        Arc::new(query),
795                        true,
796                    ))
797                }
798                _ => None,
799            };
800        }
801        None
802    }
803}
804
805impl IndexedExpression {
806    /// Create an expression that only does refine
807    fn refine_only(refine_expr: Expr) -> Self {
808        Self {
809            scalar_query: None,
810            refine_expr: Some(refine_expr),
811        }
812    }
813
814    /// Create an expression that is only an index query
815    fn index_query(column: String, index_name: String, query: Arc<dyn AnyQuery>) -> Self {
816        Self {
817            scalar_query: Some(ScalarIndexExpr::Query(ScalarIndexSearch {
818                column,
819                index_name,
820                query,
821                needs_recheck: false, // Default to false, will be set by parser
822            })),
823            refine_expr: None,
824        }
825    }
826
827    /// Create an expression that is only an index query with explicit needs_recheck
828    fn index_query_with_recheck(
829        column: String,
830        index_name: String,
831        query: Arc<dyn AnyQuery>,
832        needs_recheck: bool,
833    ) -> Self {
834        Self {
835            scalar_query: Some(ScalarIndexExpr::Query(ScalarIndexSearch {
836                column,
837                index_name,
838                query,
839                needs_recheck,
840            })),
841            refine_expr: None,
842        }
843    }
844
845    /// Try and negate the expression
846    ///
847    /// If the expression contains both an index query and a refine expression then it
848    /// cannot be negated today and None will be returned (we give up trying to use indices)
849    fn maybe_not(self) -> Option<Self> {
850        match (self.scalar_query, self.refine_expr) {
851            (Some(_), Some(_)) => None,
852            (Some(scalar_query), None) => {
853                if scalar_query.needs_recheck() {
854                    return None;
855                }
856                Some(Self {
857                    scalar_query: Some(ScalarIndexExpr::Not(Box::new(scalar_query))),
858                    refine_expr: None,
859                })
860            }
861            (None, Some(refine_expr)) => Some(Self {
862                scalar_query: None,
863                refine_expr: Some(Expr::Not(Box::new(refine_expr))),
864            }),
865            (None, None) => panic!("Empty node should not occur"),
866        }
867    }
868
869    /// Perform a logical AND of two indexed expressions
870    ///
871    /// This is straightforward because we can just AND the individual parts
872    /// because (A && B) && (C && D) == (A && C) && (B && D)
873    fn and(self, other: Self) -> Self {
874        let scalar_query = match (self.scalar_query, other.scalar_query) {
875            (Some(scalar_query), Some(other_scalar_query)) => Some(ScalarIndexExpr::And(
876                Box::new(scalar_query),
877                Box::new(other_scalar_query),
878            )),
879            (Some(scalar_query), None) => Some(scalar_query),
880            (None, Some(scalar_query)) => Some(scalar_query),
881            (None, None) => None,
882        };
883        let refine_expr = match (self.refine_expr, other.refine_expr) {
884            (Some(refine_expr), Some(other_refine_expr)) => {
885                Some(refine_expr.and(other_refine_expr))
886            }
887            (Some(refine_expr), None) => Some(refine_expr),
888            (None, Some(refine_expr)) => Some(refine_expr),
889            (None, None) => None,
890        };
891        Self {
892            scalar_query,
893            refine_expr,
894        }
895    }
896
897    /// Try and perform a logical OR of two indexed expressions
898    ///
899    /// This is a bit tricky because something like:
900    ///   (color == 'blue' AND size < 20) OR (color == 'green' AND size < 50)
901    /// is not equivalent to:
902    ///   (color == 'blue' OR color == 'green') AND (size < 20 OR size < 50)
903    fn maybe_or(self, other: Self) -> Option<Self> {
904        // If either expression is missing a scalar_query then we need to load all rows from
905        // the database and so we short-circuit and return None
906        let scalar_query = self.scalar_query?;
907        let other_scalar_query = other.scalar_query?;
908        let scalar_query = Some(ScalarIndexExpr::Or(
909            Box::new(scalar_query),
910            Box::new(other_scalar_query),
911        ));
912
913        let refine_expr = match (self.refine_expr, other.refine_expr) {
914            // TODO
915            //
916            // To handle these cases we need a way of going back from a scalar expression query to a logical DF expression (perhaps
917            // we can store the expression that led to the creation of the query)
918            //
919            // For example, imagine we have something like "(color == 'blue' AND size < 20) OR (color == 'green' AND size < 50)"
920            //
921            // We can do an indexed load of all rows matching "color == 'blue' OR color == 'green'" but then we need to
922            // refine that load with the full original expression which, at the moment, we no longer have.
923            (Some(_), Some(_)) => {
924                return None;
925            }
926            (Some(_), None) => {
927                return None;
928            }
929            (None, Some(_)) => {
930                return None;
931            }
932            (None, None) => None,
933        };
934        Some(Self {
935            scalar_query,
936            refine_expr,
937        })
938    }
939
940    fn refine(self, expr: Expr) -> Self {
941        match self.refine_expr {
942            Some(refine_expr) => Self {
943                scalar_query: self.scalar_query,
944                refine_expr: Some(refine_expr.and(expr)),
945            },
946            None => Self {
947                scalar_query: self.scalar_query,
948                refine_expr: Some(expr),
949            },
950        }
951    }
952}
953
954/// A trait implemented by anything that can load indices by name
955///
956/// This is used during the evaluation of an index expression
957#[async_trait]
958pub trait ScalarIndexLoader: Send + Sync {
959    /// Load the index with the given name
960    async fn load_index(
961        &self,
962        column: &str,
963        index_name: &str,
964        metrics: &dyn MetricsCollector,
965    ) -> Result<Arc<dyn ScalarIndex>>;
966}
967
968/// This represents a search into a scalar index
969#[derive(Debug, Clone)]
970pub struct ScalarIndexSearch {
971    /// The column to search (redundant, used for debugging messages)
972    pub column: String,
973    /// The name of the index to search
974    pub index_name: String,
975    /// The query to search for
976    pub query: Arc<dyn AnyQuery>,
977    /// If true, the query results are inexact and will need a recheck
978    pub needs_recheck: bool,
979}
980
981impl PartialEq for ScalarIndexSearch {
982    fn eq(&self, other: &Self) -> bool {
983        self.column == other.column
984            && self.index_name == other.index_name
985            && self.query.as_ref().eq(other.query.as_ref())
986    }
987}
988
989/// This represents a lookup into one or more scalar indices
990///
991/// This is a tree of operations because we may need to logically combine or
992/// modify the results of scalar lookups
993#[derive(Debug, Clone)]
994pub enum ScalarIndexExpr {
995    Not(Box<Self>),
996    And(Box<Self>, Box<Self>),
997    Or(Box<Self>, Box<Self>),
998    Query(ScalarIndexSearch),
999}
1000
1001impl PartialEq for ScalarIndexExpr {
1002    fn eq(&self, other: &Self) -> bool {
1003        match (self, other) {
1004            (Self::Not(l0), Self::Not(r0)) => l0 == r0,
1005            (Self::And(l0, l1), Self::And(r0, r1)) => l0 == r0 && l1 == r1,
1006            (Self::Or(l0, l1), Self::Or(r0, r1)) => l0 == r0 && l1 == r1,
1007            (Self::Query(l_search), Self::Query(r_search)) => l_search == r_search,
1008            _ => false,
1009        }
1010    }
1011}
1012
1013impl std::fmt::Display for ScalarIndexExpr {
1014    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1015        match self {
1016            Self::Not(inner) => write!(f, "NOT({})", inner),
1017            Self::And(lhs, rhs) => write!(f, "AND({},{})", lhs, rhs),
1018            Self::Or(lhs, rhs) => write!(f, "OR({},{})", lhs, rhs),
1019            Self::Query(search) => write!(
1020                f,
1021                "[{}]@{}",
1022                search.query.format(&search.column),
1023                search.index_name
1024            ),
1025        }
1026    }
1027}
1028
1029/// When we evaluate a scalar index query we return a batch with three columns and two rows
1030///
1031/// The first column has the block list and allow list
1032/// The second column tells if the result is least/exact/more (we repeat the discriminant twice)
1033/// The third column has the fragments covered bitmap in the first row and null in the second row
1034pub static INDEX_EXPR_RESULT_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
1035    Arc::new(Schema::new(vec![
1036        Field::new("result".to_string(), DataType::Binary, true),
1037        Field::new("discriminant".to_string(), DataType::UInt32, true),
1038        Field::new("fragments_covered".to_string(), DataType::Binary, true),
1039    ]))
1040});
1041
1042#[derive(Debug)]
1043enum NullableIndexExprResult {
1044    Exact(NullableRowAddrMask),
1045    AtMost(NullableRowAddrMask),
1046    AtLeast(NullableRowAddrMask),
1047}
1048
1049impl From<SearchResult> for NullableIndexExprResult {
1050    fn from(result: SearchResult) -> Self {
1051        match result {
1052            SearchResult::Exact(mask) => Self::Exact(NullableRowAddrMask::AllowList(mask)),
1053            SearchResult::AtMost(mask) => Self::AtMost(NullableRowAddrMask::AllowList(mask)),
1054            SearchResult::AtLeast(mask) => Self::AtLeast(NullableRowAddrMask::AllowList(mask)),
1055        }
1056    }
1057}
1058
1059impl std::ops::BitAnd<Self> for NullableIndexExprResult {
1060    type Output = Self;
1061
1062    fn bitand(self, rhs: Self) -> Self {
1063        match (self, rhs) {
1064            (Self::Exact(lhs), Self::Exact(rhs)) => Self::Exact(lhs & rhs),
1065            (Self::Exact(lhs), Self::AtMost(rhs)) | (Self::AtMost(lhs), Self::Exact(rhs)) => {
1066                Self::AtMost(lhs & rhs)
1067            }
1068            (Self::Exact(exact), Self::AtLeast(_)) | (Self::AtLeast(_), Self::Exact(exact)) => {
1069                // We could do better here, elements in both lhs and rhs are known
1070                // to be true and don't require a recheck.  We only need to recheck
1071                // elements in lhs that are not in rhs
1072                Self::AtMost(exact)
1073            }
1074            (Self::AtMost(lhs), Self::AtMost(rhs)) => Self::AtMost(lhs & rhs),
1075            (Self::AtLeast(lhs), Self::AtLeast(rhs)) => Self::AtLeast(lhs & rhs),
1076            (Self::AtMost(most), Self::AtLeast(_)) | (Self::AtLeast(_), Self::AtMost(most)) => {
1077                Self::AtMost(most)
1078            }
1079        }
1080    }
1081}
1082
1083impl std::ops::BitOr<Self> for NullableIndexExprResult {
1084    type Output = Self;
1085
1086    fn bitor(self, rhs: Self) -> Self {
1087        match (self, rhs) {
1088            (Self::Exact(lhs), Self::Exact(rhs)) => Self::Exact(lhs | rhs),
1089            (Self::Exact(lhs), Self::AtMost(rhs)) | (Self::AtMost(rhs), Self::Exact(lhs)) => {
1090                // We could do better here, elements in lhs are known to be true
1091                // and don't require a recheck.  We only need to recheck elements
1092                // in rhs that are not in lhs
1093                Self::AtMost(lhs | rhs)
1094            }
1095            (Self::Exact(lhs), Self::AtLeast(rhs)) | (Self::AtLeast(rhs), Self::Exact(lhs)) => {
1096                Self::AtLeast(lhs | rhs)
1097            }
1098            (Self::AtMost(lhs), Self::AtMost(rhs)) => Self::AtMost(lhs | rhs),
1099            (Self::AtLeast(lhs), Self::AtLeast(rhs)) => Self::AtLeast(lhs | rhs),
1100            (Self::AtMost(_), Self::AtLeast(least)) | (Self::AtLeast(least), Self::AtMost(_)) => {
1101                Self::AtLeast(least)
1102            }
1103        }
1104    }
1105}
1106
1107impl NullableIndexExprResult {
1108    pub fn drop_nulls(self) -> IndexExprResult {
1109        match self {
1110            Self::Exact(mask) => IndexExprResult::Exact(mask.drop_nulls()),
1111            Self::AtMost(mask) => IndexExprResult::AtMost(mask.drop_nulls()),
1112            Self::AtLeast(mask) => IndexExprResult::AtLeast(mask.drop_nulls()),
1113        }
1114    }
1115}
1116
1117#[derive(Debug)]
1118pub enum IndexExprResult {
1119    // The answer is exactly the rows in the allow list minus the rows in the block list
1120    Exact(RowAddrMask),
1121    // The answer is at most the rows in the allow list minus the rows in the block list
1122    // Some of the rows in the allow list may not be in the result and will need to be filtered
1123    // by a recheck.  Every row in the block list is definitely not in the result.
1124    AtMost(RowAddrMask),
1125    // The answer is at least the rows in the allow list minus the rows in the block list
1126    // Some of the rows in the block list might be in the result.  Every row in the allow list is
1127    // definitely in the result.
1128    AtLeast(RowAddrMask),
1129}
1130
1131impl IndexExprResult {
1132    pub fn row_addr_mask(&self) -> &RowAddrMask {
1133        match self {
1134            Self::Exact(mask) => mask,
1135            Self::AtMost(mask) => mask,
1136            Self::AtLeast(mask) => mask,
1137        }
1138    }
1139
1140    pub fn discriminant(&self) -> u32 {
1141        match self {
1142            Self::Exact(_) => 0,
1143            Self::AtMost(_) => 1,
1144            Self::AtLeast(_) => 2,
1145        }
1146    }
1147
1148    pub fn from_parts(mask: RowAddrMask, discriminant: u32) -> Result<Self> {
1149        match discriminant {
1150            0 => Ok(Self::Exact(mask)),
1151            1 => Ok(Self::AtMost(mask)),
1152            2 => Ok(Self::AtLeast(mask)),
1153            _ => Err(Error::invalid_input_source(
1154                format!("Invalid IndexExprResult discriminant: {}", discriminant).into(),
1155            )),
1156        }
1157    }
1158
1159    #[instrument(skip_all)]
1160    pub fn serialize_to_arrow(
1161        &self,
1162        fragments_covered_by_result: &RoaringBitmap,
1163    ) -> Result<RecordBatch> {
1164        let row_addr_mask = self.row_addr_mask();
1165        let row_addr_mask_arr = row_addr_mask.into_arrow()?;
1166        let discriminant = self.discriminant();
1167        let discriminant_arr =
1168            Arc::new(UInt32Array::from(vec![discriminant, discriminant])) as Arc<dyn Array>;
1169        let mut fragments_covered_builder = BinaryBuilder::new();
1170        let fragments_covered_bytes_len = fragments_covered_by_result.serialized_size();
1171        let mut fragments_covered_bytes = Vec::with_capacity(fragments_covered_bytes_len);
1172        fragments_covered_by_result.serialize_into(&mut fragments_covered_bytes)?;
1173        fragments_covered_builder.append_value(fragments_covered_bytes);
1174        fragments_covered_builder.append_null();
1175        let fragments_covered_arr = Arc::new(fragments_covered_builder.finish()) as Arc<dyn Array>;
1176        Ok(RecordBatch::try_new(
1177            INDEX_EXPR_RESULT_SCHEMA.clone(),
1178            vec![
1179                Arc::new(row_addr_mask_arr),
1180                Arc::new(discriminant_arr),
1181                Arc::new(fragments_covered_arr),
1182            ],
1183        )?)
1184    }
1185}
1186
1187impl ScalarIndexExpr {
1188    /// Evaluates the scalar index expression
1189    ///
1190    /// This will result in loading one or more scalar indices and searching them
1191    ///
1192    /// TODO: We could potentially try and be smarter about reusing loaded indices for
1193    /// any situations where the session cache has been disabled.
1194    #[async_recursion]
1195    async fn evaluate_impl(
1196        &self,
1197        index_loader: &dyn ScalarIndexLoader,
1198        metrics: &dyn MetricsCollector,
1199    ) -> Result<NullableIndexExprResult> {
1200        match self {
1201            Self::Not(inner) => {
1202                let result = inner.evaluate_impl(index_loader, metrics).await?;
1203                // Flip certainty: NOT(AtMost) → AtLeast, NOT(AtLeast) → AtMost
1204                Ok(match result {
1205                    NullableIndexExprResult::Exact(mask) => NullableIndexExprResult::Exact(!mask),
1206                    NullableIndexExprResult::AtMost(mask) => {
1207                        NullableIndexExprResult::AtLeast(!mask)
1208                    }
1209                    NullableIndexExprResult::AtLeast(mask) => {
1210                        NullableIndexExprResult::AtMost(!mask)
1211                    }
1212                })
1213            }
1214            Self::And(lhs, rhs) => {
1215                let lhs_result = lhs.evaluate_impl(index_loader, metrics);
1216                let rhs_result = rhs.evaluate_impl(index_loader, metrics);
1217                let (lhs_result, rhs_result) = try_join!(lhs_result, rhs_result)?;
1218                Ok(lhs_result & rhs_result)
1219            }
1220            Self::Or(lhs, rhs) => {
1221                let lhs_result = lhs.evaluate_impl(index_loader, metrics);
1222                let rhs_result = rhs.evaluate_impl(index_loader, metrics);
1223                let (lhs_result, rhs_result) = try_join!(lhs_result, rhs_result)?;
1224                Ok(lhs_result | rhs_result)
1225            }
1226            Self::Query(search) => {
1227                let index = index_loader
1228                    .load_index(&search.column, &search.index_name, metrics)
1229                    .await?;
1230                let search_result = index.search(search.query.as_ref(), metrics).await?;
1231                Ok(search_result.into())
1232            }
1233        }
1234    }
1235
1236    #[instrument(level = "debug", skip_all)]
1237    pub async fn evaluate(
1238        &self,
1239        index_loader: &dyn ScalarIndexLoader,
1240        metrics: &dyn MetricsCollector,
1241    ) -> Result<IndexExprResult> {
1242        Ok(self
1243            .evaluate_impl(index_loader, metrics)
1244            .await?
1245            .drop_nulls())
1246    }
1247
1248    pub fn to_expr(&self) -> Expr {
1249        match self {
1250            Self::Not(inner) => Expr::Not(inner.to_expr().into()),
1251            Self::And(lhs, rhs) => {
1252                let lhs = lhs.to_expr();
1253                let rhs = rhs.to_expr();
1254                lhs.and(rhs)
1255            }
1256            Self::Or(lhs, rhs) => {
1257                let lhs = lhs.to_expr();
1258                let rhs = rhs.to_expr();
1259                lhs.or(rhs)
1260            }
1261            Self::Query(search) => search.query.to_expr(search.column.clone()),
1262        }
1263    }
1264
1265    pub fn needs_recheck(&self) -> bool {
1266        match self {
1267            Self::Not(inner) => inner.needs_recheck(),
1268            Self::And(lhs, rhs) | Self::Or(lhs, rhs) => lhs.needs_recheck() || rhs.needs_recheck(),
1269            Self::Query(search) => search.needs_recheck,
1270        }
1271    }
1272}
1273
1274// Extract a column from the expression, if it is a column, or None
1275fn maybe_column(expr: &Expr) -> Option<&str> {
1276    match expr {
1277        Expr::Column(col) => Some(&col.name),
1278        _ => None,
1279    }
1280}
1281
1282// Extract the full nested column path from a get_field expression chain
1283// For example: get_field(get_field(metadata, "status"), "code") -> "metadata.status.code"
1284fn extract_nested_column_path(expr: &Expr) -> Option<String> {
1285    let mut current_expr = expr;
1286    let mut parts = Vec::new();
1287
1288    // Walk up the get_field chain
1289    loop {
1290        match current_expr {
1291            Expr::ScalarFunction(udf) if udf.name() == "get_field" => {
1292                if udf.args.len() != 2 {
1293                    return None;
1294                }
1295                // Extract the field name from the second argument
1296                // The Literal now has two fields: ScalarValue and Option<FieldMetadata>
1297                if let Expr::Literal(ScalarValue::Utf8(Some(field_name)), _) = &udf.args[1] {
1298                    parts.push(field_name.clone());
1299                } else {
1300                    return None;
1301                }
1302                // Move up to the parent expression
1303                current_expr = &udf.args[0];
1304            }
1305            Expr::Column(col) => {
1306                // We've reached the base column
1307                parts.push(col.name.clone());
1308                break;
1309            }
1310            _ => {
1311                return None;
1312            }
1313        }
1314    }
1315
1316    // Reverse to get the correct order (parent.child.grandchild)
1317    parts.reverse();
1318
1319    // Format the path correctly
1320    let field_refs: Vec<&str> = parts.iter().map(|s| s.as_str()).collect();
1321    Some(lance_core::datatypes::format_field_path(&field_refs))
1322}
1323
1324// Extract a column from the expression, if it is a column, and we have an index for that column, or None
1325//
1326// There's two ways to get a column.  First, the obvious way, is a
1327// simple column reference (e.g. x = 7).  Second, a more complex way,
1328// is some kind of projection into a column (e.g. json_extract(json, '$.name')).
1329// Third way is nested field access (e.g. get_field(metadata, "status.code"))
1330fn maybe_indexed_column<'b>(
1331    expr: &Expr,
1332    index_info: &'b dyn IndexInformationProvider,
1333) -> Option<(String, DataType, &'b dyn ScalarQueryParser)> {
1334    // First try to extract the full nested column path for get_field expressions
1335    if let Some(nested_path) = extract_nested_column_path(expr)
1336        && let Some((data_type, parser)) = index_info.get_index(&nested_path)
1337        && let Some(data_type) = parser.is_valid_reference(expr, data_type)
1338    {
1339        return Some((nested_path, data_type, parser));
1340    }
1341
1342    match expr {
1343        Expr::Column(col) => {
1344            let col = col.name.as_str();
1345            let (data_type, parser) = index_info.get_index(col)?;
1346            if let Some(data_type) = parser.is_valid_reference(expr, data_type) {
1347                Some((col.to_string(), data_type, parser))
1348            } else {
1349                None
1350            }
1351        }
1352        Expr::ScalarFunction(udf) => {
1353            if udf.args.is_empty() {
1354                return None;
1355            }
1356            // For non-get_field functions, fall back to old behavior
1357            let col = maybe_column(&udf.args[0])?;
1358            let (data_type, parser) = index_info.get_index(col)?;
1359            if let Some(data_type) = parser.is_valid_reference(expr, data_type) {
1360                Some((col.to_string(), data_type, parser))
1361            } else {
1362                None
1363            }
1364        }
1365        _ => None,
1366    }
1367}
1368
1369// Extract a literal scalar value from an expression, if it is a literal, or None
1370fn maybe_scalar(expr: &Expr, expected_type: &DataType) -> Option<ScalarValue> {
1371    match expr {
1372        Expr::Literal(value, _) => safe_coerce_scalar(value, expected_type),
1373        // Some literals can't be expressed in datafusion's SQL and can only be expressed with
1374        // a cast.  For example, there is no way to express a fixed-size-binary literal (which is
1375        // commonly used for UUID).  As a result the expression could look like...
1376        //
1377        // col = arrow_cast(value, 'fixed_size_binary(16)')
1378        //
1379        // In this case we need to extract the value, apply the cast, and then test the casted value
1380        Expr::Cast(cast) => match cast.expr.as_ref() {
1381            Expr::Literal(value, _) => {
1382                let casted = value.cast_to(&cast.data_type).ok()?;
1383                safe_coerce_scalar(&casted, expected_type)
1384            }
1385            _ => None,
1386        },
1387        Expr::ScalarFunction(scalar_function) => {
1388            if scalar_function.name() == "arrow_cast" {
1389                if scalar_function.args.len() != 2 {
1390                    return None;
1391                }
1392                match (&scalar_function.args[0], &scalar_function.args[1]) {
1393                    (Expr::Literal(value, _), Expr::Literal(cast_type, _)) => {
1394                        let target_type = scalar_function
1395                            .func
1396                            .return_field_from_args(ReturnFieldArgs {
1397                                arg_fields: &[
1398                                    Arc::new(Field::new("expression", value.data_type(), false)),
1399                                    Arc::new(Field::new("datatype", cast_type.data_type(), false)),
1400                                ],
1401                                scalar_arguments: &[Some(value), Some(cast_type)],
1402                            })
1403                            .ok()?;
1404                        let casted = value.cast_to(target_type.data_type()).ok()?;
1405                        safe_coerce_scalar(&casted, expected_type)
1406                    }
1407                    _ => None,
1408                }
1409            } else {
1410                None
1411            }
1412        }
1413        _ => None,
1414    }
1415}
1416
1417// Extract a list of scalar values from an expression, if it is a list of scalar values, or None
1418fn maybe_scalar_list(exprs: &Vec<Expr>, expected_type: &DataType) -> Option<Vec<ScalarValue>> {
1419    let mut scalar_values = Vec::with_capacity(exprs.len());
1420    for expr in exprs {
1421        match maybe_scalar(expr, expected_type) {
1422            Some(scalar_val) => {
1423                scalar_values.push(scalar_val);
1424            }
1425            None => {
1426                return None;
1427            }
1428        }
1429    }
1430    Some(scalar_values)
1431}
1432
1433fn visit_between(
1434    between: &Between,
1435    index_info: &dyn IndexInformationProvider,
1436) -> Option<IndexedExpression> {
1437    let (column, col_type, query_parser) = maybe_indexed_column(&between.expr, index_info)?;
1438    let low = maybe_scalar(&between.low, &col_type)?;
1439    let high = maybe_scalar(&between.high, &col_type)?;
1440
1441    let indexed_expr =
1442        query_parser.visit_between(&column, &Bound::Included(low), &Bound::Included(high))?;
1443
1444    if between.negated {
1445        indexed_expr.maybe_not()
1446    } else {
1447        Some(indexed_expr)
1448    }
1449}
1450
1451fn visit_in_list(
1452    in_list: &InList,
1453    index_info: &dyn IndexInformationProvider,
1454) -> Option<IndexedExpression> {
1455    let (column, col_type, query_parser) = maybe_indexed_column(&in_list.expr, index_info)?;
1456    let values = maybe_scalar_list(&in_list.list, &col_type)?;
1457
1458    let indexed_expr = query_parser.visit_in_list(&column, &values)?;
1459
1460    if in_list.negated {
1461        indexed_expr.maybe_not()
1462    } else {
1463        Some(indexed_expr)
1464    }
1465}
1466
1467fn visit_is_bool(
1468    expr: &Expr,
1469    index_info: &dyn IndexInformationProvider,
1470    value: bool,
1471) -> Option<IndexedExpression> {
1472    let (column, col_type, query_parser) = maybe_indexed_column(expr, index_info)?;
1473    if col_type != DataType::Boolean {
1474        None
1475    } else {
1476        query_parser.visit_is_bool(&column, value)
1477    }
1478}
1479
1480// A column can be a valid indexed expression if the column is boolean (e.g. 'WHERE on_sale')
1481fn visit_column(
1482    col: &Expr,
1483    index_info: &dyn IndexInformationProvider,
1484) -> Option<IndexedExpression> {
1485    let (column, col_type, query_parser) = maybe_indexed_column(col, index_info)?;
1486    if col_type != DataType::Boolean {
1487        None
1488    } else {
1489        query_parser.visit_is_bool(&column, true)
1490    }
1491}
1492
1493fn visit_is_null(
1494    expr: &Expr,
1495    index_info: &dyn IndexInformationProvider,
1496    negated: bool,
1497) -> Option<IndexedExpression> {
1498    let (column, _, query_parser) = maybe_indexed_column(expr, index_info)?;
1499    let indexed_expr = query_parser.visit_is_null(&column)?;
1500    if negated {
1501        indexed_expr.maybe_not()
1502    } else {
1503        Some(indexed_expr)
1504    }
1505}
1506
1507fn visit_not(
1508    expr: &Expr,
1509    index_info: &dyn IndexInformationProvider,
1510    depth: usize,
1511) -> Result<Option<IndexedExpression>> {
1512    let node = visit_node(expr, index_info, depth + 1)?;
1513    Ok(node.and_then(|node| node.maybe_not()))
1514}
1515
1516fn visit_comparison(
1517    expr: &BinaryExpr,
1518    index_info: &dyn IndexInformationProvider,
1519) -> Option<IndexedExpression> {
1520    let left_col = maybe_indexed_column(&expr.left, index_info);
1521    if let Some((column, col_type, query_parser)) = left_col {
1522        let scalar = maybe_scalar(&expr.right, &col_type)?;
1523        query_parser.visit_comparison(&column, &scalar, &expr.op)
1524    } else {
1525        // Datafusion's query simplifier will canonicalize expressions and so we shouldn't reach this case.  If, for some reason, we
1526        // do reach this case we can handle it in the future by inverting expr.op and swapping the left and right sides
1527        None
1528    }
1529}
1530
1531fn maybe_range(
1532    expr: &BinaryExpr,
1533    index_info: &dyn IndexInformationProvider,
1534) -> Option<IndexedExpression> {
1535    let left_expr = match expr.left.as_ref() {
1536        Expr::BinaryExpr(binary_expr) => Some(binary_expr),
1537        _ => None,
1538    }?;
1539    let right_expr = match expr.right.as_ref() {
1540        Expr::BinaryExpr(binary_expr) => Some(binary_expr),
1541        _ => None,
1542    }?;
1543
1544    let (left_col, dt, parser) = maybe_indexed_column(&left_expr.left, index_info)?;
1545    let right_col = maybe_column(&right_expr.left)?;
1546
1547    if left_col != right_col {
1548        return None;
1549    }
1550
1551    let left_value = maybe_scalar(&left_expr.right, &dt)?;
1552    let right_value = maybe_scalar(&right_expr.right, &dt)?;
1553
1554    let (low, high) = match (left_expr.op, right_expr.op) {
1555        // x >= a && x <= b
1556        (Operator::GtEq, Operator::LtEq) => {
1557            (Bound::Included(left_value), Bound::Included(right_value))
1558        }
1559        // x >= a && x < b
1560        (Operator::GtEq, Operator::Lt) => {
1561            (Bound::Included(left_value), Bound::Excluded(right_value))
1562        }
1563        // x > a && x <= b
1564        (Operator::Gt, Operator::LtEq) => {
1565            (Bound::Excluded(left_value), Bound::Included(right_value))
1566        }
1567        // x > a && x < b
1568        (Operator::Gt, Operator::Lt) => (Bound::Excluded(left_value), Bound::Excluded(right_value)),
1569        // x <= a && x >= b
1570        (Operator::LtEq, Operator::GtEq) => {
1571            (Bound::Included(right_value), Bound::Included(left_value))
1572        }
1573        // x <= a && x > b
1574        (Operator::LtEq, Operator::Gt) => {
1575            (Bound::Included(right_value), Bound::Excluded(left_value))
1576        }
1577        // x < a && x >= b
1578        (Operator::Lt, Operator::GtEq) => {
1579            (Bound::Excluded(right_value), Bound::Included(left_value))
1580        }
1581        // x < a && x > b
1582        (Operator::Lt, Operator::Gt) => (Bound::Excluded(right_value), Bound::Excluded(left_value)),
1583        _ => return None,
1584    };
1585
1586    parser.visit_between(&left_col, &low, &high)
1587}
1588
1589fn visit_and(
1590    expr: &BinaryExpr,
1591    index_info: &dyn IndexInformationProvider,
1592    depth: usize,
1593) -> Result<Option<IndexedExpression>> {
1594    // Many scalar indices can efficiently handle a BETWEEN query as a single search and this
1595    // can be much more efficient than two separate range queries.  As an optimization we check
1596    // to see if this is a between query and, if so, we handle it as a single query
1597    //
1598    // Note: We can't rely on users writing the SQL BETWEEN operator because:
1599    //   * Some users won't realize it's an option or a good idea
1600    //   * Datafusion's simplifier will rewrite the BETWEEN operator into two separate range queries
1601    if let Some(range_expr) = maybe_range(expr, index_info) {
1602        return Ok(Some(range_expr));
1603    }
1604
1605    let left = visit_node(&expr.left, index_info, depth + 1)?;
1606    let right = visit_node(&expr.right, index_info, depth + 1)?;
1607    Ok(match (left, right) {
1608        (Some(left), Some(right)) => Some(left.and(right)),
1609        (Some(left), None) => Some(left.refine((*expr.right).clone())),
1610        (None, Some(right)) => Some(right.refine((*expr.left).clone())),
1611        (None, None) => None,
1612    })
1613}
1614
1615fn visit_or(
1616    expr: &BinaryExpr,
1617    index_info: &dyn IndexInformationProvider,
1618    depth: usize,
1619) -> Result<Option<IndexedExpression>> {
1620    let left = visit_node(&expr.left, index_info, depth + 1)?;
1621    let right = visit_node(&expr.right, index_info, depth + 1)?;
1622    Ok(match (left, right) {
1623        (Some(left), Some(right)) => left.maybe_or(right),
1624        // If one side can use an index and the other side cannot then
1625        // we must abandon the entire thing.  For example, consider the
1626        // query "color == 'blue' or size > 10" where color is indexed but
1627        // size is not.  It's entirely possible that size > 10 matches every
1628        // row in our database.  There is nothing we can do except a full scan
1629        (Some(_), None) => None,
1630        (None, Some(_)) => None,
1631        (None, None) => None,
1632    })
1633}
1634
1635fn visit_binary_expr(
1636    expr: &BinaryExpr,
1637    index_info: &dyn IndexInformationProvider,
1638    depth: usize,
1639) -> Result<Option<IndexedExpression>> {
1640    match &expr.op {
1641        Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq | Operator::Eq => {
1642            Ok(visit_comparison(expr, index_info))
1643        }
1644        // visit_comparison will maybe create an Eq query which we negate
1645        Operator::NotEq => Ok(visit_comparison(expr, index_info).and_then(|node| node.maybe_not())),
1646        Operator::And => visit_and(expr, index_info, depth),
1647        Operator::Or => visit_or(expr, index_info, depth),
1648        _ => Ok(None),
1649    }
1650}
1651
1652fn visit_scalar_fn(
1653    scalar_fn: &ScalarFunction,
1654    index_info: &dyn IndexInformationProvider,
1655) -> Option<IndexedExpression> {
1656    if scalar_fn.args.is_empty() {
1657        return None;
1658    }
1659    let (col, data_type, query_parser) = maybe_indexed_column(&scalar_fn.args[0], index_info)?;
1660    query_parser.visit_scalar_function(&col, &data_type, &scalar_fn.func, &scalar_fn.args)
1661}
1662
1663fn visit_node(
1664    expr: &Expr,
1665    index_info: &dyn IndexInformationProvider,
1666    depth: usize,
1667) -> Result<Option<IndexedExpression>> {
1668    if depth >= MAX_DEPTH {
1669        return Err(Error::invalid_input(format!(
1670            "the filter expression is too long, lance limit the max number of conditions to {}",
1671            MAX_DEPTH
1672        )));
1673    }
1674    match expr {
1675        Expr::Between(between) => Ok(visit_between(between, index_info)),
1676        Expr::Alias(alias) => visit_node(alias.expr.as_ref(), index_info, depth),
1677        Expr::Column(_) => Ok(visit_column(expr, index_info)),
1678        Expr::InList(in_list) => Ok(visit_in_list(in_list, index_info)),
1679        Expr::IsFalse(expr) => Ok(visit_is_bool(expr.as_ref(), index_info, false)),
1680        Expr::IsTrue(expr) => Ok(visit_is_bool(expr.as_ref(), index_info, true)),
1681        Expr::IsNull(expr) => Ok(visit_is_null(expr.as_ref(), index_info, false)),
1682        Expr::IsNotNull(expr) => Ok(visit_is_null(expr.as_ref(), index_info, true)),
1683        Expr::Not(expr) => visit_not(expr.as_ref(), index_info, depth),
1684        Expr::BinaryExpr(binary_expr) => visit_binary_expr(binary_expr, index_info, depth),
1685        Expr::ScalarFunction(scalar_fn) => Ok(visit_scalar_fn(scalar_fn, index_info)),
1686        _ => Ok(None),
1687    }
1688}
1689
1690/// A trait to be used in `apply_scalar_indices` to inform the function which columns are indexeds
1691pub trait IndexInformationProvider {
1692    /// Check if an index exists for `col` and, if so, return the data type of col
1693    /// as well as a query parser that can parse queries for that column
1694    fn get_index(&self, col: &str) -> Option<(&DataType, &dyn ScalarQueryParser)>;
1695}
1696
1697/// Attempt to split a filter expression into a search of scalar indexes and an
1698///   optional post-search refinement query
1699pub fn apply_scalar_indices(
1700    expr: Expr,
1701    index_info: &dyn IndexInformationProvider,
1702) -> Result<IndexedExpression> {
1703    Ok(visit_node(&expr, index_info, 0)?.unwrap_or(IndexedExpression::refine_only(expr)))
1704}
1705
1706#[derive(Clone, Default, Debug)]
1707pub struct FilterPlan {
1708    pub index_query: Option<ScalarIndexExpr>,
1709    /// True if the index query is guaranteed to return exact results
1710    pub skip_recheck: bool,
1711    pub refine_expr: Option<Expr>,
1712    pub full_expr: Option<Expr>,
1713}
1714
1715impl FilterPlan {
1716    pub fn empty() -> Self {
1717        Self {
1718            index_query: None,
1719            skip_recheck: true,
1720            refine_expr: None,
1721            full_expr: None,
1722        }
1723    }
1724
1725    pub fn new_refine_only(expr: Expr) -> Self {
1726        Self {
1727            index_query: None,
1728            skip_recheck: true,
1729            refine_expr: Some(expr.clone()),
1730            full_expr: Some(expr),
1731        }
1732    }
1733
1734    pub fn is_empty(&self) -> bool {
1735        self.refine_expr.is_none() && self.index_query.is_none()
1736    }
1737
1738    pub fn all_columns(&self) -> Vec<String> {
1739        self.full_expr
1740            .as_ref()
1741            .map(Planner::column_names_in_expr)
1742            .unwrap_or_default()
1743    }
1744
1745    pub fn refine_columns(&self) -> Vec<String> {
1746        self.refine_expr
1747            .as_ref()
1748            .map(Planner::column_names_in_expr)
1749            .unwrap_or_default()
1750    }
1751
1752    /// Return true if this has a refine step, regardless of the status of prefilter
1753    pub fn has_refine(&self) -> bool {
1754        self.refine_expr.is_some()
1755    }
1756
1757    /// Return true if this has a scalar index query
1758    pub fn has_index_query(&self) -> bool {
1759        self.index_query.is_some()
1760    }
1761
1762    pub fn has_any_filter(&self) -> bool {
1763        self.refine_expr.is_some() || self.index_query.is_some()
1764    }
1765
1766    pub fn make_refine_only(&mut self) {
1767        self.index_query = None;
1768        self.refine_expr = self.full_expr.clone();
1769    }
1770
1771    /// Return true if there is no refine or recheck of any kind and there is an index query
1772    pub fn is_exact_index_search(&self) -> bool {
1773        self.index_query.is_some() && self.refine_expr.is_none() && self.skip_recheck
1774    }
1775}
1776
1777pub trait PlannerIndexExt {
1778    /// Determine how to apply a provided filter
1779    ///
1780    /// We parse the filter into a logical expression.  We then
1781    /// split the logical expression into a portion that can be
1782    /// satisfied by an index search (of one or more indices) and
1783    /// a refine portion that must be applied after the index search
1784    fn create_filter_plan(
1785        &self,
1786        filter: Expr,
1787        index_info: &dyn IndexInformationProvider,
1788        use_scalar_index: bool,
1789    ) -> Result<FilterPlan>;
1790}
1791
1792impl PlannerIndexExt for Planner {
1793    fn create_filter_plan(
1794        &self,
1795        filter: Expr,
1796        index_info: &dyn IndexInformationProvider,
1797        use_scalar_index: bool,
1798    ) -> Result<FilterPlan> {
1799        let logical_expr = self.optimize_expr(filter)?;
1800        if use_scalar_index {
1801            let indexed_expr = apply_scalar_indices(logical_expr.clone(), index_info)?;
1802            let mut skip_recheck = false;
1803            if let Some(scalar_query) = indexed_expr.scalar_query.as_ref() {
1804                skip_recheck = !scalar_query.needs_recheck();
1805            }
1806            Ok(FilterPlan {
1807                index_query: indexed_expr.scalar_query,
1808                refine_expr: indexed_expr.refine_expr,
1809                full_expr: Some(logical_expr),
1810                skip_recheck,
1811            })
1812        } else {
1813            Ok(FilterPlan {
1814                index_query: None,
1815                skip_recheck: true,
1816                refine_expr: Some(logical_expr.clone()),
1817                full_expr: Some(logical_expr),
1818            })
1819        }
1820    }
1821}
1822
1823#[cfg(test)]
1824mod tests {
1825    use std::collections::HashMap;
1826
1827    use arrow_schema::{Field, Schema};
1828    use chrono::Utc;
1829    use datafusion_common::{Column, DFSchema};
1830    use datafusion_expr::execution_props::ExecutionProps;
1831    use datafusion_expr::simplify::SimplifyContext;
1832    use lance_datafusion::exec::{LanceExecutionOptions, get_session_context};
1833
1834    use crate::scalar::json::{JsonQuery, JsonQueryParser};
1835
1836    use super::*;
1837
1838    struct ColInfo {
1839        data_type: DataType,
1840        parser: Box<dyn ScalarQueryParser>,
1841    }
1842
1843    impl ColInfo {
1844        fn new(data_type: DataType, parser: Box<dyn ScalarQueryParser>) -> Self {
1845            Self { data_type, parser }
1846        }
1847    }
1848
1849    struct MockIndexInfoProvider {
1850        indexed_columns: HashMap<String, ColInfo>,
1851    }
1852
1853    impl MockIndexInfoProvider {
1854        fn new(indexed_columns: Vec<(&str, ColInfo)>) -> Self {
1855            Self {
1856                indexed_columns: HashMap::from_iter(
1857                    indexed_columns
1858                        .into_iter()
1859                        .map(|(s, ty)| (s.to_string(), ty)),
1860                ),
1861            }
1862        }
1863    }
1864
1865    impl IndexInformationProvider for MockIndexInfoProvider {
1866        fn get_index(&self, col: &str) -> Option<(&DataType, &dyn ScalarQueryParser)> {
1867            self.indexed_columns
1868                .get(col)
1869                .map(|col_info| (&col_info.data_type, col_info.parser.as_ref()))
1870        }
1871    }
1872
1873    fn check(
1874        index_info: &dyn IndexInformationProvider,
1875        expr: &str,
1876        expected: Option<IndexedExpression>,
1877        optimize: bool,
1878    ) {
1879        let schema = Schema::new(vec![
1880            Field::new("color", DataType::Utf8, false),
1881            Field::new("size", DataType::Float32, false),
1882            Field::new("aisle", DataType::UInt32, false),
1883            Field::new("on_sale", DataType::Boolean, false),
1884            Field::new("price", DataType::Float32, false),
1885            Field::new("json", DataType::LargeBinary, false),
1886        ]);
1887        let df_schema: DFSchema = schema.try_into().unwrap();
1888
1889        let ctx = get_session_context(&LanceExecutionOptions::default());
1890        let state = ctx.state();
1891        let mut expr = state.create_logical_expr(expr, &df_schema).unwrap();
1892        if optimize {
1893            let props = ExecutionProps::new().with_query_execution_start_time(Utc::now());
1894            let simplify_context = SimplifyContext::new(&props).with_schema(Arc::new(df_schema));
1895            let simplifier =
1896                datafusion::optimizer::simplify_expressions::ExprSimplifier::new(simplify_context);
1897            expr = simplifier.simplify(expr).unwrap();
1898        }
1899
1900        let actual = apply_scalar_indices(expr.clone(), index_info).unwrap();
1901        if let Some(expected) = expected {
1902            assert_eq!(actual, expected);
1903        } else {
1904            assert!(actual.scalar_query.is_none());
1905            assert_eq!(actual.refine_expr.unwrap(), expr);
1906        }
1907    }
1908
1909    fn check_no_index(index_info: &dyn IndexInformationProvider, expr: &str) {
1910        check(index_info, expr, None, false)
1911    }
1912
1913    fn check_simple(
1914        index_info: &dyn IndexInformationProvider,
1915        expr: &str,
1916        col: &str,
1917        query: impl AnyQuery,
1918    ) {
1919        check(
1920            index_info,
1921            expr,
1922            Some(IndexedExpression::index_query(
1923                col.to_string(),
1924                format!("{}_idx", col),
1925                Arc::new(query),
1926            )),
1927            false,
1928        )
1929    }
1930
1931    fn check_range(
1932        index_info: &dyn IndexInformationProvider,
1933        expr: &str,
1934        col: &str,
1935        query: SargableQuery,
1936    ) {
1937        check(
1938            index_info,
1939            expr,
1940            Some(IndexedExpression::index_query(
1941                col.to_string(),
1942                format!("{}_idx", col),
1943                Arc::new(query),
1944            )),
1945            true,
1946        )
1947    }
1948
1949    fn check_simple_negated(
1950        index_info: &dyn IndexInformationProvider,
1951        expr: &str,
1952        col: &str,
1953        query: SargableQuery,
1954    ) {
1955        check(
1956            index_info,
1957            expr,
1958            Some(
1959                IndexedExpression::index_query(
1960                    col.to_string(),
1961                    format!("{}_idx", col),
1962                    Arc::new(query),
1963                )
1964                .maybe_not()
1965                .unwrap(),
1966            ),
1967            false,
1968        )
1969    }
1970
1971    #[test]
1972    fn test_expressions() {
1973        let index_info = MockIndexInfoProvider::new(vec![
1974            (
1975                "color",
1976                ColInfo::new(
1977                    DataType::Utf8,
1978                    Box::new(SargableQueryParser::new("color_idx".to_string(), false)),
1979                ),
1980            ),
1981            (
1982                "aisle",
1983                ColInfo::new(
1984                    DataType::UInt32,
1985                    Box::new(SargableQueryParser::new("aisle_idx".to_string(), false)),
1986                ),
1987            ),
1988            (
1989                "on_sale",
1990                ColInfo::new(
1991                    DataType::Boolean,
1992                    Box::new(SargableQueryParser::new("on_sale_idx".to_string(), false)),
1993                ),
1994            ),
1995            (
1996                "price",
1997                ColInfo::new(
1998                    DataType::Float32,
1999                    Box::new(SargableQueryParser::new("price_idx".to_string(), false)),
2000                ),
2001            ),
2002            (
2003                "json",
2004                ColInfo::new(
2005                    DataType::LargeBinary,
2006                    Box::new(JsonQueryParser::new(
2007                        "$.name".to_string(),
2008                        Box::new(SargableQueryParser::new("json_idx".to_string(), false)),
2009                    )),
2010                ),
2011            ),
2012        ]);
2013
2014        check_simple(
2015            &index_info,
2016            "json_extract(json, '$.name') = 'foo'",
2017            "json",
2018            JsonQuery::new(
2019                Arc::new(SargableQuery::Equals(ScalarValue::Utf8(Some(
2020                    "foo".to_string(),
2021                )))),
2022                "$.name".to_string(),
2023            ),
2024        );
2025
2026        check_no_index(&index_info, "size BETWEEN 5 AND 10");
2027        // Cast case.  We will cast 5 (an int64) to Int16 and then coerce to UInt32
2028        check_simple(
2029            &index_info,
2030            "aisle = arrow_cast(5, 'Int16')",
2031            "aisle",
2032            SargableQuery::Equals(ScalarValue::UInt32(Some(5))),
2033        );
2034        // 5 different ways of writing BETWEEN (all should be recognized)
2035        check_range(
2036            &index_info,
2037            "aisle BETWEEN 5 AND 10",
2038            "aisle",
2039            SargableQuery::Range(
2040                Bound::Included(ScalarValue::UInt32(Some(5))),
2041                Bound::Included(ScalarValue::UInt32(Some(10))),
2042            ),
2043        );
2044        check_range(
2045            &index_info,
2046            "aisle >= 5 AND aisle <= 10",
2047            "aisle",
2048            SargableQuery::Range(
2049                Bound::Included(ScalarValue::UInt32(Some(5))),
2050                Bound::Included(ScalarValue::UInt32(Some(10))),
2051            ),
2052        );
2053
2054        check_range(
2055            &index_info,
2056            "aisle <= 10 AND aisle >= 5",
2057            "aisle",
2058            SargableQuery::Range(
2059                Bound::Included(ScalarValue::UInt32(Some(5))),
2060                Bound::Included(ScalarValue::UInt32(Some(10))),
2061            ),
2062        );
2063
2064        check_range(
2065            &index_info,
2066            "5 <= aisle AND 10 >= aisle",
2067            "aisle",
2068            SargableQuery::Range(
2069                Bound::Included(ScalarValue::UInt32(Some(5))),
2070                Bound::Included(ScalarValue::UInt32(Some(10))),
2071            ),
2072        );
2073
2074        check_range(
2075            &index_info,
2076            "10 >= aisle AND 5 <= aisle",
2077            "aisle",
2078            SargableQuery::Range(
2079                Bound::Included(ScalarValue::UInt32(Some(5))),
2080                Bound::Included(ScalarValue::UInt32(Some(10))),
2081            ),
2082        );
2083        check_simple(
2084            &index_info,
2085            "on_sale IS TRUE",
2086            "on_sale",
2087            SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2088        );
2089        check_simple(
2090            &index_info,
2091            "on_sale",
2092            "on_sale",
2093            SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2094        );
2095        check_simple_negated(
2096            &index_info,
2097            "NOT on_sale",
2098            "on_sale",
2099            SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2100        );
2101        check_simple(
2102            &index_info,
2103            "on_sale IS FALSE",
2104            "on_sale",
2105            SargableQuery::Equals(ScalarValue::Boolean(Some(false))),
2106        );
2107        check_simple_negated(
2108            &index_info,
2109            "aisle NOT BETWEEN 5 AND 10",
2110            "aisle",
2111            SargableQuery::Range(
2112                Bound::Included(ScalarValue::UInt32(Some(5))),
2113                Bound::Included(ScalarValue::UInt32(Some(10))),
2114            ),
2115        );
2116        // Small in-list (in-list with 3 or fewer items optimizes into or-chain)
2117        check_simple(
2118            &index_info,
2119            "aisle IN (5, 6, 7)",
2120            "aisle",
2121            SargableQuery::IsIn(vec![
2122                ScalarValue::UInt32(Some(5)),
2123                ScalarValue::UInt32(Some(6)),
2124                ScalarValue::UInt32(Some(7)),
2125            ]),
2126        );
2127        check_simple_negated(
2128            &index_info,
2129            "NOT aisle IN (5, 6, 7)",
2130            "aisle",
2131            SargableQuery::IsIn(vec![
2132                ScalarValue::UInt32(Some(5)),
2133                ScalarValue::UInt32(Some(6)),
2134                ScalarValue::UInt32(Some(7)),
2135            ]),
2136        );
2137        check_simple_negated(
2138            &index_info,
2139            "aisle NOT IN (5, 6, 7)",
2140            "aisle",
2141            SargableQuery::IsIn(vec![
2142                ScalarValue::UInt32(Some(5)),
2143                ScalarValue::UInt32(Some(6)),
2144                ScalarValue::UInt32(Some(7)),
2145            ]),
2146        );
2147        check_simple(
2148            &index_info,
2149            "aisle IN (5, 6, 7, 8, 9)",
2150            "aisle",
2151            SargableQuery::IsIn(vec![
2152                ScalarValue::UInt32(Some(5)),
2153                ScalarValue::UInt32(Some(6)),
2154                ScalarValue::UInt32(Some(7)),
2155                ScalarValue::UInt32(Some(8)),
2156                ScalarValue::UInt32(Some(9)),
2157            ]),
2158        );
2159        check_simple_negated(
2160            &index_info,
2161            "NOT aisle IN (5, 6, 7, 8, 9)",
2162            "aisle",
2163            SargableQuery::IsIn(vec![
2164                ScalarValue::UInt32(Some(5)),
2165                ScalarValue::UInt32(Some(6)),
2166                ScalarValue::UInt32(Some(7)),
2167                ScalarValue::UInt32(Some(8)),
2168                ScalarValue::UInt32(Some(9)),
2169            ]),
2170        );
2171        check_simple_negated(
2172            &index_info,
2173            "aisle NOT IN (5, 6, 7, 8, 9)",
2174            "aisle",
2175            SargableQuery::IsIn(vec![
2176                ScalarValue::UInt32(Some(5)),
2177                ScalarValue::UInt32(Some(6)),
2178                ScalarValue::UInt32(Some(7)),
2179                ScalarValue::UInt32(Some(8)),
2180                ScalarValue::UInt32(Some(9)),
2181            ]),
2182        );
2183        check_simple(
2184            &index_info,
2185            "on_sale is false",
2186            "on_sale",
2187            SargableQuery::Equals(ScalarValue::Boolean(Some(false))),
2188        );
2189        check_simple(
2190            &index_info,
2191            "on_sale is true",
2192            "on_sale",
2193            SargableQuery::Equals(ScalarValue::Boolean(Some(true))),
2194        );
2195        check_simple(
2196            &index_info,
2197            "aisle < 10",
2198            "aisle",
2199            SargableQuery::Range(
2200                Bound::Unbounded,
2201                Bound::Excluded(ScalarValue::UInt32(Some(10))),
2202            ),
2203        );
2204        check_simple(
2205            &index_info,
2206            "aisle <= 10",
2207            "aisle",
2208            SargableQuery::Range(
2209                Bound::Unbounded,
2210                Bound::Included(ScalarValue::UInt32(Some(10))),
2211            ),
2212        );
2213        check_simple(
2214            &index_info,
2215            "aisle > 10",
2216            "aisle",
2217            SargableQuery::Range(
2218                Bound::Excluded(ScalarValue::UInt32(Some(10))),
2219                Bound::Unbounded,
2220            ),
2221        );
2222        // In the future we can handle this case if we need to.  For
2223        // now let's make sure we don't accidentally do the wrong thing
2224        // (we were getting this backwards in the past)
2225        check_no_index(&index_info, "10 > aisle");
2226        check_simple(
2227            &index_info,
2228            "aisle >= 10",
2229            "aisle",
2230            SargableQuery::Range(
2231                Bound::Included(ScalarValue::UInt32(Some(10))),
2232                Bound::Unbounded,
2233            ),
2234        );
2235        check_simple(
2236            &index_info,
2237            "aisle = 10",
2238            "aisle",
2239            SargableQuery::Equals(ScalarValue::UInt32(Some(10))),
2240        );
2241        check_simple_negated(
2242            &index_info,
2243            "aisle <> 10",
2244            "aisle",
2245            SargableQuery::Equals(ScalarValue::UInt32(Some(10))),
2246        );
2247        // // Common compound case, AND'd clauses
2248        let left = Box::new(ScalarIndexExpr::Query(ScalarIndexSearch {
2249            column: "aisle".to_string(),
2250            index_name: "aisle_idx".to_string(),
2251            query: Arc::new(SargableQuery::Equals(ScalarValue::UInt32(Some(10)))),
2252            needs_recheck: false,
2253        }));
2254        let right = Box::new(ScalarIndexExpr::Query(ScalarIndexSearch {
2255            column: "color".to_string(),
2256            index_name: "color_idx".to_string(),
2257            query: Arc::new(SargableQuery::Equals(ScalarValue::Utf8(Some(
2258                "blue".to_string(),
2259            )))),
2260            needs_recheck: false,
2261        }));
2262        check(
2263            &index_info,
2264            "aisle = 10 AND color = 'blue'",
2265            Some(IndexedExpression {
2266                scalar_query: Some(ScalarIndexExpr::And(left.clone(), right.clone())),
2267                refine_expr: None,
2268            }),
2269            false,
2270        );
2271        // Compound AND's and not all of them are indexed columns
2272        let refine = Expr::Column(Column::new_unqualified("size")).gt(datafusion_expr::lit(30_i64));
2273        check(
2274            &index_info,
2275            "aisle = 10 AND color = 'blue' AND size > 30",
2276            Some(IndexedExpression {
2277                scalar_query: Some(ScalarIndexExpr::And(left.clone(), right.clone())),
2278                refine_expr: Some(refine.clone()),
2279            }),
2280            false,
2281        );
2282        // Compounded OR's where ALL columns are indexed
2283        check(
2284            &index_info,
2285            "aisle = 10 OR color = 'blue'",
2286            Some(IndexedExpression {
2287                scalar_query: Some(ScalarIndexExpr::Or(left.clone(), right.clone())),
2288                refine_expr: None,
2289            }),
2290            false,
2291        );
2292        // Compounded OR's with one or more unindexed columns
2293        check_no_index(&index_info, "aisle = 10 OR color = 'blue' OR size > 30");
2294        // AND'd group of OR
2295        check(
2296            &index_info,
2297            "(aisle = 10 OR color = 'blue') AND size > 30",
2298            Some(IndexedExpression {
2299                scalar_query: Some(ScalarIndexExpr::Or(left, right)),
2300                refine_expr: Some(refine),
2301            }),
2302            false,
2303        );
2304        // Examples of things that are not yet supported but should be supportable someday
2305
2306        // OR'd group of refined index searches (see IndexedExpression::or for details)
2307        check_no_index(
2308            &index_info,
2309            "(aisle = 10 AND size > 30) OR (color = 'blue' AND size > 20)",
2310        );
2311
2312        // Non-normalized arithmetic (can use expression simplification)
2313        check_no_index(&index_info, "aisle + 3 < 10");
2314
2315        // Currently we assume that the return of an index search tells us which rows are
2316        // TRUE and all other rows are FALSE.  This will need to change but for now it is
2317        // safer to not support the following cases because the return value of non-matched
2318        // rows is NULL and not FALSE.
2319        check_no_index(&index_info, "aisle IN (5, 6, NULL)");
2320        // OR-list with NULL (in future DF version this will be optimized repr of
2321        // small in-list with NULL so let's get ready for it)
2322        check_no_index(&index_info, "aisle = 5 OR aisle = 6 OR NULL");
2323        check_no_index(&index_info, "aisle IN (5, 6, 7, 8, NULL)");
2324        check_no_index(&index_info, "aisle = NULL");
2325        check_no_index(&index_info, "aisle BETWEEN 5 AND NULL");
2326        check_no_index(&index_info, "aisle BETWEEN NULL AND 10");
2327    }
2328
2329    #[tokio::test]
2330    async fn test_not_flips_certainty() {
2331        use lance_core::utils::mask::{NullableRowAddrSet, RowAddrTreeMap};
2332
2333        // Test that NOT flips certainty for inexact index results
2334        // This tests the implementation in evaluate_impl for Self::Not
2335
2336        // Helper function that mimics the NOT logic we just fixed
2337        fn apply_not(result: NullableIndexExprResult) -> NullableIndexExprResult {
2338            match result {
2339                NullableIndexExprResult::Exact(mask) => NullableIndexExprResult::Exact(!mask),
2340                NullableIndexExprResult::AtMost(mask) => NullableIndexExprResult::AtLeast(!mask),
2341                NullableIndexExprResult::AtLeast(mask) => NullableIndexExprResult::AtMost(!mask),
2342            }
2343        }
2344
2345        // AtMost: superset of matches (e.g., bloom filter says "might be in [1,2]")
2346        let at_most = NullableIndexExprResult::AtMost(NullableRowAddrMask::AllowList(
2347            NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2348        ));
2349        // NOT(AtMost) should be AtLeast (definitely NOT in [1,2], might be elsewhere)
2350        assert!(matches!(
2351            apply_not(at_most),
2352            NullableIndexExprResult::AtLeast(_)
2353        ));
2354
2355        // AtLeast: subset of matches (e.g., definitely in [1,2], might be more)
2356        let at_least = NullableIndexExprResult::AtLeast(NullableRowAddrMask::AllowList(
2357            NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2358        ));
2359        // NOT(AtLeast) should be AtMost (might NOT be in [1,2], definitely elsewhere)
2360        assert!(matches!(
2361            apply_not(at_least),
2362            NullableIndexExprResult::AtMost(_)
2363        ));
2364
2365        // Exact should stay Exact
2366        let exact = NullableIndexExprResult::Exact(NullableRowAddrMask::AllowList(
2367            NullableRowAddrSet::new(RowAddrTreeMap::from_iter(&[1, 2]), RowAddrTreeMap::new()),
2368        ));
2369        assert!(matches!(
2370            apply_not(exact),
2371            NullableIndexExprResult::Exact(_)
2372        ));
2373    }
2374
2375    #[tokio::test]
2376    async fn test_and_or_preserve_certainty() {
2377        use lance_core::utils::mask::{NullableRowAddrSet, RowAddrTreeMap};
2378
2379        // Test that AND/OR correctly propagate certainty
2380        let make_at_most = || {
2381            NullableIndexExprResult::AtMost(NullableRowAddrMask::AllowList(
2382                NullableRowAddrSet::new(
2383                    RowAddrTreeMap::from_iter(&[1, 2, 3]),
2384                    RowAddrTreeMap::new(),
2385                ),
2386            ))
2387        };
2388
2389        let make_at_least = || {
2390            NullableIndexExprResult::AtLeast(NullableRowAddrMask::AllowList(
2391                NullableRowAddrSet::new(
2392                    RowAddrTreeMap::from_iter(&[2, 3, 4]),
2393                    RowAddrTreeMap::new(),
2394                ),
2395            ))
2396        };
2397
2398        let make_exact = || {
2399            NullableIndexExprResult::Exact(NullableRowAddrMask::AllowList(NullableRowAddrSet::new(
2400                RowAddrTreeMap::from_iter(&[1, 2]),
2401                RowAddrTreeMap::new(),
2402            )))
2403        };
2404
2405        // AtMost & AtMost → AtMost
2406        assert!(matches!(
2407            make_at_most() & make_at_most(),
2408            NullableIndexExprResult::AtMost(_)
2409        ));
2410
2411        // AtLeast & AtLeast → AtLeast
2412        assert!(matches!(
2413            make_at_least() & make_at_least(),
2414            NullableIndexExprResult::AtLeast(_)
2415        ));
2416
2417        // AtMost & AtLeast → AtMost (superset remains superset)
2418        assert!(matches!(
2419            make_at_most() & make_at_least(),
2420            NullableIndexExprResult::AtMost(_)
2421        ));
2422
2423        // AtMost | AtMost → AtMost
2424        assert!(matches!(
2425            make_at_most() | make_at_most(),
2426            NullableIndexExprResult::AtMost(_)
2427        ));
2428
2429        // AtLeast | AtLeast → AtLeast
2430        assert!(matches!(
2431            make_at_least() | make_at_least(),
2432            NullableIndexExprResult::AtLeast(_)
2433        ));
2434
2435        // AtMost | AtLeast → AtLeast (subset coverage guaranteed)
2436        assert!(matches!(
2437            make_at_most() | make_at_least(),
2438            NullableIndexExprResult::AtLeast(_)
2439        ));
2440
2441        // Exact & AtMost → AtMost
2442        assert!(matches!(
2443            make_exact() & make_at_most(),
2444            NullableIndexExprResult::AtMost(_)
2445        ));
2446
2447        // Exact | AtLeast → AtLeast
2448        assert!(matches!(
2449            make_exact() | make_at_least(),
2450            NullableIndexExprResult::AtLeast(_)
2451        ));
2452    }
2453}