mito2/sst/index/fulltext_index/applier/
builder.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeMap;
16
17use common_base::term_token::extract_like_probes;
18use common_telemetry::warn;
19use datafusion_common::ScalarValue;
20use datafusion_expr::expr::{Like, ScalarFunction};
21use datafusion_expr::{BinaryExpr, Expr, Operator};
22use datatypes::schema::FulltextBackend;
23use index::target::IndexTarget;
24use object_store::ObjectStore;
25use puffin::puffin_manager::cache::PuffinMetadataCacheRef;
26use store_api::metadata::RegionMetadata;
27use store_api::region_request::PathType;
28
29use crate::cache::file_cache::FileCacheRef;
30use crate::cache::index::bloom_filter_index::BloomFilterIndexCacheRef;
31use crate::error::Result;
32use crate::sst::index::column::{fulltext_target_options, json_index_target};
33use crate::sst::index::fulltext_index::applier::FulltextIndexApplier;
34use crate::sst::index::puffin_manager::PuffinManagerFactory;
35
36/// A request for fulltext index.
37///
38/// It contains all the queries and terms for a column.
39#[derive(Default, Debug, Clone, PartialEq, Eq, Hash)]
40pub struct FulltextRequest {
41    pub queries: Vec<FulltextQuery>,
42    pub terms: Vec<FulltextTerm>,
43    pub likes: Vec<FulltextLike>,
44}
45
46/// A query to be matched in fulltext index.
47///
48/// `query` is the query to be matched, e.g. "+foo -bar" in `SELECT * FROM t WHERE matches(text, "+foo -bar")`.
49#[derive(Debug, Clone, PartialEq, Eq, Hash)]
50pub struct FulltextQuery(pub String);
51
52/// A term to be matched in fulltext index.
53///
54/// `term` is the term to be matched, e.g. "foo" in `SELECT * FROM t WHERE matches_term(text, "foo")`.
55/// `col_lowered` indicates whether the column is lowercased, e.g. `col_lowered = true` when `matches_term(lower(text), "foo")`.
56#[derive(Debug, Clone, PartialEq, Eq, Hash)]
57pub struct FulltextTerm {
58    pub col_lowered: bool,
59    pub term: String,
60}
61
62/// A case-sensitive `LIKE` pattern to be matched in fulltext index.
63///
64/// `pattern` is e.g. "%foo bar%" in `SELECT * FROM t WHERE text LIKE '%foo bar%'`.
65/// `col_lowered` indicates whether the column is lowercased, e.g. `lower(text) LIKE '%foo bar%'`.
66#[derive(Debug, Clone, PartialEq, Eq, Hash)]
67pub struct FulltextLike {
68    pub col_lowered: bool,
69    pub pattern: String,
70}
71
72/// `FulltextIndexApplierBuilder` is a builder for `FulltextIndexApplier`.
73pub struct FulltextIndexApplierBuilder<'a> {
74    table_dir: String,
75    path_type: PathType,
76    store: ObjectStore,
77    puffin_manager_factory: PuffinManagerFactory,
78    metadata: &'a RegionMetadata,
79    file_cache: Option<FileCacheRef>,
80    puffin_metadata_cache: Option<PuffinMetadataCacheRef>,
81    bloom_filter_cache: Option<BloomFilterIndexCacheRef>,
82}
83
84impl<'a> FulltextIndexApplierBuilder<'a> {
85    /// Creates a new `FulltextIndexApplierBuilder`.
86    pub fn new(
87        table_dir: String,
88        path_type: PathType,
89        store: ObjectStore,
90        puffin_manager_factory: PuffinManagerFactory,
91        metadata: &'a RegionMetadata,
92    ) -> Self {
93        Self {
94            table_dir,
95            path_type,
96            store,
97            puffin_manager_factory,
98            metadata,
99            file_cache: None,
100            puffin_metadata_cache: None,
101            bloom_filter_cache: None,
102        }
103    }
104
105    /// Sets the file cache to be used by the `FulltextIndexApplier`.
106    pub fn with_file_cache(mut self, file_cache: Option<FileCacheRef>) -> Self {
107        self.file_cache = file_cache;
108        self
109    }
110
111    /// Sets the puffin metadata cache to be used by the `FulltextIndexApplier`.
112    pub fn with_puffin_metadata_cache(
113        mut self,
114        puffin_metadata_cache: Option<PuffinMetadataCacheRef>,
115    ) -> Self {
116        self.puffin_metadata_cache = puffin_metadata_cache;
117        self
118    }
119
120    /// Sets the bloom filter cache to be used by the `FulltextIndexApplier`.
121    pub fn with_bloom_filter_cache(
122        mut self,
123        bloom_filter_cache: Option<BloomFilterIndexCacheRef>,
124    ) -> Self {
125        self.bloom_filter_cache = bloom_filter_cache;
126        self
127    }
128
129    /// Builds `SstIndexApplier` from the given expressions.
130    pub fn build(self, exprs: &[Expr]) -> Result<Option<FulltextIndexApplier>> {
131        let mut requests = BTreeMap::new();
132        for expr in exprs {
133            Self::extract_requests(expr, self.metadata, &mut requests);
134        }
135
136        // Check if any requests have queries or terms
137        let has_requests = requests.iter().any(|(_, request)| {
138            !request.queries.is_empty() || !request.terms.is_empty() || !request.likes.is_empty()
139        });
140
141        Ok(has_requests.then(|| {
142            FulltextIndexApplier::new(
143                self.table_dir,
144                self.path_type,
145                self.store,
146                requests,
147                self.puffin_manager_factory,
148                self.metadata,
149            )
150            .with_file_cache(self.file_cache)
151            .with_puffin_metadata_cache(self.puffin_metadata_cache)
152            .with_bloom_filter_cache(self.bloom_filter_cache)
153        }))
154    }
155
156    fn extract_requests(
157        expr: &Expr,
158        metadata: &'a RegionMetadata,
159        requests: &mut BTreeMap<IndexTarget, FulltextRequest>,
160    ) {
161        if let Expr::BinaryExpr(BinaryExpr {
162            left,
163            op: Operator::And,
164            right,
165        }) = expr
166        {
167            Self::extract_requests(left, metadata, requests);
168            Self::extract_requests(right, metadata, requests);
169            return;
170        }
171        if let Err(err) = Self::extract_request(expr, metadata, requests) {
172            warn!(err; "Failed to extract fulltext index predicate from {expr}");
173        }
174    }
175
176    fn extract_request(
177        expr: &Expr,
178        metadata: &RegionMetadata,
179        requests: &mut BTreeMap<IndexTarget, FulltextRequest>,
180    ) -> Result<()> {
181        match expr {
182            Expr::Like(like) => {
183                if let Some((target, like)) = Self::expr_to_like(metadata, like)? {
184                    requests.entry(target).or_default().likes.push(like);
185                }
186            }
187            Expr::ScalarFunction(func) => {
188                if let Some((target, query)) = Self::expr_to_query(metadata, func)? {
189                    requests.entry(target).or_default().queries.push(query);
190                } else if let Some((target, term)) = Self::expr_to_term(metadata, func)? {
191                    requests.entry(target).or_default().terms.push(term);
192                }
193            }
194            _ => {}
195        }
196        Ok(())
197    }
198
199    fn expr_to_query(
200        metadata: &RegionMetadata,
201        f: &ScalarFunction,
202    ) -> Result<Option<(IndexTarget, FulltextQuery)>> {
203        if f.name() != "matches" {
204            return Ok(None);
205        }
206        if f.args.len() != 2 {
207            return Ok(None);
208        }
209
210        let Some(target) = Self::index_target_from_expr(metadata, &f.args[0])? else {
211            return Ok(None);
212        };
213
214        let Expr::Literal(
215            ScalarValue::Utf8(Some(query))
216            | ScalarValue::LargeUtf8(Some(query))
217            | ScalarValue::Utf8View(Some(query)),
218            _,
219        ) = &f.args[1]
220        else {
221            return Ok(None);
222        };
223
224        Ok(Some((target, FulltextQuery(query.clone()))))
225    }
226
227    fn expr_to_term(
228        metadata: &RegionMetadata,
229        f: &ScalarFunction,
230    ) -> Result<Option<(IndexTarget, FulltextTerm)>> {
231        if f.name() != "matches_term" {
232            return Ok(None);
233        }
234        if f.args.len() != 2 {
235            return Ok(None);
236        }
237
238        let (expr, lowered) = Self::unwrap_lower(&f.args[0]);
239        let Some(target) = Self::index_target_from_expr(metadata, expr)? else {
240            return Ok(None);
241        };
242
243        let Expr::Literal(
244            ScalarValue::Utf8(Some(term))
245            | ScalarValue::LargeUtf8(Some(term))
246            | ScalarValue::Utf8View(Some(term)),
247            _,
248        ) = &f.args[1]
249        else {
250            return Ok(None);
251        };
252
253        Ok(Some((
254            target,
255            FulltextTerm {
256                col_lowered: lowered,
257                term: term.clone(),
258            },
259        )))
260    }
261
262    fn expr_to_like(
263        metadata: &RegionMetadata,
264        like: &Like,
265    ) -> Result<Option<(IndexTarget, FulltextLike)>> {
266        // `ILIKE` uses Unicode case folding, which disagrees with the index's
267        // `to_lowercase` on characters like 'ſ', so its probes could miss rows.
268        if like.negated || like.case_insensitive {
269            return Ok(None);
270        }
271        // Probes assume arrow's default `\` escape.
272        if !matches!(like.escape_char, None | Some('\\')) {
273            return Ok(None);
274        }
275
276        let (expr, col_lowered) = Self::unwrap_lower(&like.expr);
277        let Some(target) = Self::index_target_from_expr(metadata, expr)? else {
278            return Ok(None);
279        };
280        let Some(options) = fulltext_target_options(metadata, &target)? else {
281            return Ok(None);
282        };
283        if options.backend != FulltextBackend::Bloom {
284            return Ok(None);
285        }
286
287        let Expr::Literal(
288            ScalarValue::Utf8(Some(pattern))
289            | ScalarValue::LargeUtf8(Some(pattern))
290            | ScalarValue::Utf8View(Some(pattern)),
291            _,
292        ) = like.pattern.as_ref()
293        else {
294            return Ok(None);
295        };
296        if extract_like_probes(pattern).is_empty() {
297            return Ok(None);
298        }
299
300        Ok(Some((
301            target,
302            FulltextLike {
303                col_lowered,
304                pattern: pattern.clone(),
305            },
306        )))
307    }
308
309    /// Resolves a column reference or JSON path expression to a single index target.
310    /// Returns `None` for unsupported expressions without traversing their children.
311    fn index_target_from_expr(
312        metadata: &RegionMetadata,
313        expr: &Expr,
314    ) -> Result<Option<IndexTarget>> {
315        if let Expr::Column(col) = expr {
316            let Some(col) = metadata.column_by_name(&col.name) else {
317                return Ok(None);
318            };
319            if !col.column_schema.data_type.is_string() {
320                return Ok(None);
321            }
322            let target = IndexTarget::ColumnId(col.column_id);
323            return Ok(fulltext_target_options(metadata, &target)?.map(|_| target));
324        }
325        let Some((target, ty)) = json_index_target(metadata, expr, |hint| {
326            hint.fulltext_index
327                .as_ref()
328                .is_some_and(|options| options.enable)
329        })?
330        else {
331            return Ok(None);
332        };
333        Ok(ty.is_string().then_some(target))
334    }
335
336    /// Returns the inner expression and `true` for `lower(expr)`, or the original
337    /// expression and `false` otherwise.
338    fn unwrap_lower(expr: &Expr) -> (&Expr, bool) {
339        if let Expr::ScalarFunction(func) = expr
340            && let Some(inner) = Self::extract_lower_arg(func)
341        {
342            return (inner, true);
343        }
344        (expr, false)
345    }
346
347    fn extract_lower_arg(lower_func: &ScalarFunction) -> Option<&Expr> {
348        if lower_func.args.len() != 1 {
349            return None;
350        }
351
352        if lower_func.name() != "lower" {
353            return None;
354        }
355
356        Some(&lower_func.args[0])
357    }
358}
359
360#[cfg(test)]
361mod tests {
362    use std::sync::Arc;
363
364    use common_function::function::FunctionRef;
365    use common_function::function_factory::ScalarFunctionFactory;
366    use common_function::scalars::json::json_get::JsonGetWithType;
367    use common_function::scalars::matches::MatchesFunction;
368    use common_function::scalars::matches_term::MatchesTermFunction;
369    use common_function::scalars::udf::create_udf;
370    use datafusion::functions::string::lower;
371    use datafusion_common::Column;
372    use datafusion_expr::expr::{Like, ScalarFunction};
373    use datafusion_expr::{Literal, ScalarUDF};
374    use datatypes::schema::{FULLTEXT_KEY, FulltextOptions};
375    use store_api::metadata::RegionMetadataBuilder;
376
377    use super::*;
378    use crate::sst::index::fulltext_index::applier::tests::{
379        json_metadata, mock_metadata_with_fulltext, mock_metadata_without_fulltext,
380    };
381
382    #[test]
383    fn test_json_fulltext_predicate_extraction() {
384        let options = FulltextOptions {
385            enable: true,
386            ..Default::default()
387        };
388        let metadata = json_metadata(Some(options));
389        let target = IndexTarget::new_json_path(3, vec!["message".into()]).unwrap();
390        for literal in [
391            ScalarValue::Utf8(Some("error timeout".into())),
392            ScalarValue::LargeUtf8(Some("error timeout".into())),
393            ScalarValue::Utf8View(Some("error timeout".into())),
394        ] {
395            let expr = json_expr("$.message", Some(ScalarValue::Utf8View(None)));
396            let query = ScalarFunction::new_udf(
397                matches_func(),
398                vec![expr.clone(), Expr::Literal(literal.clone(), None)],
399            );
400            assert_eq!(
401                FulltextIndexApplierBuilder::expr_to_query(&metadata, &query).unwrap(),
402                Some((target.clone(), FulltextQuery("error timeout".into())))
403            );
404            for lowered in [false, true] {
405                let expr = if lowered {
406                    lower().call(vec![expr.clone()])
407                } else {
408                    expr.clone()
409                };
410                let term = ScalarFunction::new_udf(
411                    matches_term_func(),
412                    vec![expr.clone(), Expr::Literal(literal.clone(), None)],
413                );
414                let (actual, term) = FulltextIndexApplierBuilder::expr_to_term(&metadata, &term)
415                    .unwrap()
416                    .unwrap();
417                assert_eq!(actual, target);
418                assert_eq!(
419                    term,
420                    FulltextTerm {
421                        col_lowered: lowered,
422                        term: "error timeout".into()
423                    }
424                );
425                let like = Like::new(
426                    false,
427                    Box::new(expr),
428                    Box::new(Expr::Literal(literal.clone(), None)),
429                    None,
430                    false,
431                );
432                let (actual, like) = FulltextIndexApplierBuilder::expr_to_like(&metadata, &like)
433                    .unwrap()
434                    .unwrap();
435                assert_eq!(actual, target);
436                assert_eq!(
437                    like,
438                    FulltextLike {
439                        col_lowered: lowered,
440                        pattern: "error timeout".into()
441                    }
442                );
443            }
444        }
445        for expr in [
446            json_expr("$.missing", None),
447            json_expr("$.message", Some(ScalarValue::Int64(None))),
448        ] {
449            assert!(
450                FulltextIndexApplierBuilder::index_target_from_expr(&metadata, &expr)
451                    .unwrap()
452                    .is_none()
453            );
454        }
455        for options in [None, Some(FulltextOptions::default())] {
456            let metadata = json_metadata(options);
457            let path = json_expr("$.message", None);
458            for expr in [
459                matches_func().call(vec![path.clone(), "error".lit()]),
460                matches_term_func().call(vec![path.clone(), "error".lit()]),
461                path.like("%error timeout%".lit()),
462            ] {
463                let mut requests = BTreeMap::new();
464                FulltextIndexApplierBuilder::extract_requests(&expr, &metadata, &mut requests);
465                assert!(requests.is_empty(), "{expr}");
466            }
467        }
468    }
469
470    fn json_expr(path: &str, ty: Option<ScalarValue>) -> Expr {
471        let mut args = vec![datafusion_expr::col("attrs"), datafusion_expr::lit(path)];
472        if let Some(ty) = ty {
473            args.push(Expr::Literal(ty, None));
474        }
475        Expr::ScalarFunction(ScalarFunction::new_udf(
476            Arc::new(create_udf(Arc::new(JsonGetWithType::default()))),
477            args,
478        ))
479    }
480
481    fn mock_metadata() -> RegionMetadata {
482        mock_metadata_with_fulltext(FulltextBackend::Bloom)
483    }
484
485    fn matches_func() -> Arc<ScalarUDF> {
486        Arc::new(
487            ScalarFunctionFactory::from(Arc::new(MatchesFunction::default()) as FunctionRef)
488                .provide(Default::default()),
489        )
490    }
491
492    fn matches_term_func() -> Arc<ScalarUDF> {
493        Arc::new(
494            ScalarFunctionFactory::from(Arc::new(MatchesTermFunction::default()) as FunctionRef)
495                .provide(Default::default()),
496        )
497    }
498
499    #[test]
500    fn test_ordinary_fulltext_requests_require_enabled_index() {
501        for options in [
502            None,
503            Some(FulltextOptions::default()),
504            Some(FulltextOptions {
505                enable: true,
506                ..Default::default()
507            }),
508            Some(FulltextOptions {
509                enable: true,
510                backend: FulltextBackend::Tantivy,
511                ..Default::default()
512            }),
513        ] {
514            let enabled = options.as_ref().is_some_and(|options| options.enable);
515            let bloom_enabled = options
516                .as_ref()
517                .is_some_and(|options| options.enable && options.backend == FulltextBackend::Bloom);
518            let mut metadata = mock_metadata_without_fulltext();
519            if let Some(options) = options {
520                metadata.column_metadatas[0]
521                    .column_schema
522                    .set_fulltext_options(&options)
523                    .unwrap();
524            }
525            let column = datafusion_expr::col("text");
526            let expressions = [
527                (
528                    matches_func().call(vec![column.clone(), "error".lit()]),
529                    enabled,
530                ),
531                (
532                    matches_term_func().call(vec![column.clone(), "error".lit()]),
533                    enabled,
534                ),
535                (
536                    matches_term_func()
537                        .call(vec![lower().call(vec![column.clone()]), "error".lit()]),
538                    enabled,
539                ),
540                (column.like("%error timeout retry%".lit()), bloom_enabled),
541            ];
542            for (expr, expected) in expressions {
543                let mut requests = BTreeMap::new();
544                FulltextIndexApplierBuilder::extract_requests(&expr, &metadata, &mut requests);
545                assert_eq!(!requests.is_empty(), expected, "{expr}");
546            }
547        }
548    }
549
550    #[test]
551    fn test_expr_to_query_basic() {
552        let metadata = mock_metadata();
553        for literal in [
554            ScalarValue::Utf8(Some("foo".into())),
555            ScalarValue::LargeUtf8(Some("foo".into())),
556            ScalarValue::Utf8View(Some("foo".into())),
557        ] {
558            let func = ScalarFunction::new_udf(
559                matches_func(),
560                vec![datafusion_expr::col("text"), Expr::Literal(literal, None)],
561            );
562            assert_eq!(
563                FulltextIndexApplierBuilder::expr_to_query(&metadata, &func).unwrap(),
564                Some((IndexTarget::ColumnId(1), FulltextQuery("foo".into())))
565            );
566        }
567    }
568
569    #[test]
570    fn test_expr_to_query_wrong_num_args() {
571        let metadata = mock_metadata();
572
573        let func = ScalarFunction {
574            args: vec![Expr::Column(Column::from_name("text"))],
575            func: matches_func(),
576        };
577
578        assert!(
579            FulltextIndexApplierBuilder::expr_to_query(&metadata, &func)
580                .unwrap()
581                .is_none()
582        );
583    }
584
585    #[test]
586    fn test_expr_to_query_not_found_column() {
587        let metadata = mock_metadata();
588
589        let func = ScalarFunction {
590            args: vec![Expr::Column(Column::from_name("not_found")), "foo".lit()],
591            func: matches_func(),
592        };
593
594        assert!(
595            FulltextIndexApplierBuilder::expr_to_query(&metadata, &func)
596                .unwrap()
597                .is_none()
598        );
599    }
600
601    #[test]
602    fn test_expr_to_query_column_wrong_data_type() {
603        let metadata = mock_metadata();
604
605        let func = ScalarFunction {
606            args: vec![Expr::Column(Column::from_name("ts")), "foo".lit()],
607            func: matches_func(),
608        };
609
610        assert!(
611            FulltextIndexApplierBuilder::expr_to_query(&metadata, &func)
612                .unwrap()
613                .is_none()
614        );
615    }
616
617    #[test]
618    fn test_expr_to_query_pattern_not_string() {
619        let metadata = mock_metadata();
620
621        let func = ScalarFunction {
622            args: vec![Expr::Column(Column::from_name("text")), 42.lit()],
623            func: matches_func(),
624        };
625
626        assert!(
627            FulltextIndexApplierBuilder::expr_to_query(&metadata, &func)
628                .unwrap()
629                .is_none()
630        );
631    }
632
633    #[test]
634    fn test_expr_to_term_basic() {
635        let metadata = mock_metadata();
636        for literal in [
637            ScalarValue::Utf8(Some("foo".into())),
638            ScalarValue::LargeUtf8(Some("foo".into())),
639            ScalarValue::Utf8View(Some("foo".into())),
640        ] {
641            let func = ScalarFunction::new_udf(
642                matches_term_func(),
643                vec![datafusion_expr::col("text"), Expr::Literal(literal, None)],
644            );
645            assert_eq!(
646                FulltextIndexApplierBuilder::expr_to_term(&metadata, &func).unwrap(),
647                Some((
648                    IndexTarget::ColumnId(1),
649                    FulltextTerm {
650                        col_lowered: false,
651                        term: "foo".into()
652                    }
653                ))
654            );
655        }
656    }
657
658    #[test]
659    fn test_expr_to_term_with_lower() {
660        let metadata = mock_metadata();
661
662        let lower_func_expr = ScalarFunction {
663            args: vec![Expr::Column(Column::from_name("text"))],
664            func: lower(),
665        };
666
667        let func = ScalarFunction {
668            args: vec![Expr::ScalarFunction(lower_func_expr), "foo".lit()],
669            func: matches_term_func(),
670        };
671
672        let (target, term) = FulltextIndexApplierBuilder::expr_to_term(&metadata, &func)
673            .unwrap()
674            .unwrap();
675        assert_eq!(target, IndexTarget::ColumnId(1));
676        assert_eq!(
677            term,
678            FulltextTerm {
679                col_lowered: true,
680                term: "foo".to_string(),
681            }
682        );
683    }
684
685    #[test]
686    fn test_expr_to_term_wrong_num_args() {
687        let metadata = mock_metadata();
688
689        let func = ScalarFunction {
690            args: vec![Expr::Column(Column::from_name("text"))],
691            func: matches_term_func(),
692        };
693
694        assert!(
695            FulltextIndexApplierBuilder::expr_to_term(&metadata, &func)
696                .unwrap()
697                .is_none()
698        );
699    }
700
701    #[test]
702    fn test_expr_to_term_wrong_function_name() {
703        let metadata = mock_metadata();
704
705        let func = ScalarFunction {
706            args: vec![Expr::Column(Column::from_name("text")), "foo".lit()],
707            func: matches_func(), // Using 'matches' instead of 'matches_term'
708        };
709
710        assert!(
711            FulltextIndexApplierBuilder::expr_to_term(&metadata, &func)
712                .unwrap()
713                .is_none()
714        );
715    }
716
717    #[test]
718    fn test_extract_lower_arg() {
719        let func = ScalarFunction {
720            args: vec![Expr::Column(Column::from_name("text"))],
721            func: lower(),
722        };
723
724        let arg = FulltextIndexApplierBuilder::extract_lower_arg(&func).unwrap();
725        match arg {
726            Expr::Column(c) => {
727                assert_eq!(c.name, "text");
728            }
729            _ => panic!("Expected Column expression"),
730        }
731    }
732
733    #[test]
734    fn test_extract_lower_arg_wrong_function() {
735        let func = ScalarFunction {
736            args: vec![Expr::Column(Column::from_name("text"))],
737            func: matches_func(), // Not 'lower'
738        };
739
740        assert!(FulltextIndexApplierBuilder::extract_lower_arg(&func).is_none());
741    }
742
743    #[test]
744    fn test_extract_requests() {
745        let metadata = mock_metadata();
746
747        // Create a matches expression
748        let matches_expr = Expr::ScalarFunction(ScalarFunction {
749            args: vec![Expr::Column(Column::from_name("text")), "foo".lit()],
750            func: matches_func(),
751        });
752
753        let mut requests = BTreeMap::new();
754        FulltextIndexApplierBuilder::extract_requests(&matches_expr, &metadata, &mut requests);
755
756        assert_eq!(requests.len(), 1);
757        let request = requests.get(&IndexTarget::ColumnId(1)).unwrap();
758        assert_eq!(request.queries.len(), 1);
759        assert_eq!(request.terms.len(), 0);
760        assert_eq!(request.queries[0], FulltextQuery("foo".to_string()));
761    }
762
763    #[test]
764    fn test_expr_to_like() {
765        let like = |negated, case_insensitive, escape_char, pattern: &str| Like {
766            negated,
767            expr: Box::new(Expr::Column(Column::from_name("text"))),
768            pattern: Box::new(pattern.lit()),
769            escape_char,
770            case_insensitive,
771        };
772        let bloom = mock_metadata_with_fulltext(FulltextBackend::Bloom);
773
774        for escape_char in [None, Some('\\')] {
775            assert_eq!(
776                FulltextIndexApplierBuilder::expr_to_like(
777                    &bloom,
778                    &like(false, false, escape_char, "%foo bar baz%")
779                )
780                .unwrap(),
781                Some((
782                    IndexTarget::ColumnId(1),
783                    FulltextLike {
784                        col_lowered: false,
785                        pattern: "%foo bar baz%".to_string(),
786                    }
787                ))
788            );
789        }
790        let lowered = Like {
791            expr: Box::new(Expr::ScalarFunction(ScalarFunction {
792                args: vec![Expr::Column(Column::from_name("text"))],
793                func: lower(),
794            })),
795            ..like(false, false, None, "%foo bar baz%")
796        };
797        assert_eq!(
798            FulltextIndexApplierBuilder::expr_to_like(&bloom, &lowered).unwrap(),
799            Some((
800                IndexTarget::ColumnId(1),
801                FulltextLike {
802                    col_lowered: true,
803                    pattern: "%foo bar baz%".to_string(),
804                }
805            ))
806        );
807        // NOT LIKE, ILIKE and other escape characters can't use the probes, and
808        // '%foo%' has none.
809        for like in [
810            like(true, false, None, "%foo bar baz%"),
811            like(false, true, None, "%foo bar baz%"),
812            like(false, false, Some('!'), "%foo bar baz%"),
813            like(false, false, None, "%foo%"),
814        ] {
815            assert_eq!(
816                FulltextIndexApplierBuilder::expr_to_like(&bloom, &like).unwrap(),
817                None
818            );
819        }
820        // Only columns with a bloom fulltext index.
821        for metadata in [
822            mock_metadata_without_fulltext(),
823            mock_metadata_with_fulltext(FulltextBackend::Tantivy),
824        ] {
825            assert_eq!(
826                FulltextIndexApplierBuilder::expr_to_like(
827                    &metadata,
828                    &like(false, false, None, "%foo bar baz%")
829                )
830                .unwrap(),
831                None
832            );
833        }
834    }
835
836    #[test]
837    fn test_extract_requests_boolean_boundaries() {
838        let metadata = json_metadata(Some(FulltextOptions {
839            enable: true,
840            ..Default::default()
841        }));
842        let ordinary = matches_func().call(vec![datafusion_expr::col("text"), "error".lit()]);
843        let json = matches_term_func().call(vec![json_expr("$.message", None), "error".lit()]);
844        let ordinary_target = IndexTarget::ColumnId(1);
845        let json_target = IndexTarget::new_json_path(3, vec!["message".into()]).unwrap();
846        for (expr, expected) in [
847            (
848                ordinary.clone().and(json.clone()),
849                vec![ordinary_target.clone(), json_target],
850            ),
851            (ordinary.clone().or(json.clone()), vec![]),
852            (Expr::Not(Box::new(json.clone())), vec![]),
853            (
854                ordinary.clone().and(json.or(ordinary.clone())),
855                vec![ordinary_target],
856            ),
857        ] {
858            let mut requests = BTreeMap::new();
859            FulltextIndexApplierBuilder::extract_requests(&expr, &metadata, &mut requests);
860            assert_eq!(
861                requests.keys().cloned().collect::<Vec<_>>(),
862                expected,
863                "{expr}"
864            );
865        }
866    }
867
868    #[test]
869    fn test_extract_requests_keeps_valid_conjunct_on_error() {
870        let mut metadata = json_metadata(Some(FulltextOptions {
871            enable: true,
872            ..Default::default()
873        }));
874        let column = &mut metadata.column_metadatas[0].column_schema;
875        let mut options = column.metadata().clone();
876        options.insert(FULLTEXT_KEY.into(), "invalid json".into());
877        *column = column.clone().with_metadata(options);
878        let metadata = RegionMetadataBuilder::from_existing(metadata)
879            .build()
880            .unwrap();
881        let column = datafusion_expr::col("text");
882        let valid = matches_term_func().call(vec![json_expr("$.message", None), "error".lit()]);
883        let expected = BTreeMap::from([(
884            IndexTarget::new_json_path(3, vec!["message".into()]).unwrap(),
885            FulltextRequest {
886                terms: vec![FulltextTerm {
887                    col_lowered: false,
888                    term: "error".into(),
889                }],
890                ..Default::default()
891            },
892        )]);
893        for invalid in [
894            matches_func().call(vec![column.clone(), "error".lit()]),
895            matches_term_func().call(vec![column.clone(), "error".lit()]),
896            column.like("%error timeout%".lit()),
897        ] {
898            let mut requests = BTreeMap::new();
899            assert!(
900                FulltextIndexApplierBuilder::extract_request(&invalid, &metadata, &mut requests)
901                    .is_err()
902            );
903            assert!(requests.is_empty());
904            for expr in [
905                invalid.clone().and(valid.clone()),
906                valid.clone().and(invalid),
907            ] {
908                FulltextIndexApplierBuilder::extract_requests(&expr, &metadata, &mut requests);
909                assert_eq!(requests, expected, "{expr}");
910                requests.clear();
911            }
912        }
913    }
914
915    #[test]
916    fn test_extract_multiple_requests() {
917        let metadata = mock_metadata();
918
919        // Create a matches expression
920        let matches_expr = Expr::ScalarFunction(ScalarFunction {
921            args: vec![Expr::Column(Column::from_name("text")), "foo".lit()],
922            func: matches_func(),
923        });
924
925        // Create a matches_term expression
926        let matches_term_expr = Expr::ScalarFunction(ScalarFunction {
927            args: vec![Expr::Column(Column::from_name("text")), "bar".lit()],
928            func: matches_term_func(),
929        });
930
931        // Create a binary expression combining both
932        let binary_expr = Expr::BinaryExpr(BinaryExpr {
933            left: Box::new(matches_expr),
934            op: Operator::And,
935            right: Box::new(matches_term_expr),
936        });
937
938        let mut requests = BTreeMap::new();
939        FulltextIndexApplierBuilder::extract_requests(&binary_expr, &metadata, &mut requests);
940
941        assert_eq!(requests.len(), 1);
942        let request = requests.get(&IndexTarget::ColumnId(1)).unwrap();
943        assert_eq!(request.queries.len(), 1);
944        assert_eq!(request.terms.len(), 1);
945        assert_eq!(request.queries[0], FulltextQuery("foo".to_string()));
946        assert_eq!(
947            request.terms[0],
948            FulltextTerm {
949                col_lowered: false,
950                term: "bar".to_string(),
951            }
952        );
953    }
954}
mito2/sst/index/fulltext_index/applier/builder.rs

mito2/sst/index/fulltext_index/applier/
builder.rs
