mito2/sst/index/fulltext_index/
creator.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::HashMap;
16use std::sync::Arc;
17use std::sync::atomic::AtomicUsize;
18
19use api::v1::SemanticType;
20use arrow_schema::extension::ExtensionType;
21use common_telemetry::warn;
22use datatypes::arrow::array::{Array, LargeStringArray, StringArray};
23use datatypes::arrow::datatypes::DataType;
24use datatypes::arrow::record_batch::RecordBatch;
25use datatypes::extension::json::Json2ExtensionType;
26use datatypes::schema::{FulltextAnalyzer, FulltextBackend};
27use datatypes::vectors::VectorRef;
28use index::fulltext_index::create::{
29    BloomFilterFulltextIndexCreator, FulltextIndexCreator, TantivyFulltextIndexCreator,
30};
31use index::fulltext_index::{Analyzer, Config};
32use index::target::IndexTarget;
33use puffin::blob_metadata::CompressionCodec;
34use puffin::puffin_manager::PutOptions;
35use snafu::{OptionExt, ResultExt, ensure};
36use store_api::metadata::RegionMetadataRef;
37use store_api::storage::{ConcreteDataType, FileId, RegionId};
38
39use crate::error::{
40    CastVectorSnafu, ComputeArrowSnafu, CreateFulltextCreatorSnafu, DataTypeMismatchSnafu,
41    FulltextFinishSnafu, FulltextPushTextSnafu, IndexOptionsSnafu, InvalidRecordBatchSnafu,
42    OperateAbortedIndexSnafu, Result,
43};
44use crate::read::Batch;
45use crate::sst::index::column::extract_json_type_hint_values;
46use crate::sst::index::fulltext_index::{INDEX_BLOB_TYPE_BLOOM, INDEX_BLOB_TYPE_TANTIVY};
47use crate::sst::index::intermediate::{
48    IntermediateLocation, IntermediateManager, TempFileProvider,
49};
50use crate::sst::index::puffin_manager::SstPuffinWriter;
51use crate::sst::index::statistics::{ByteCount, RowCount, Statistics};
52use crate::sst::index::{IndexBuildType, TYPE_FULLTEXT_INDEX};
53
54/// `FulltextIndexer` is responsible for creating fulltext indexes for SST files.
55pub struct FulltextIndexer {
56    /// Creators for each column.
57    creators: HashMap<IndexTarget, SingleCreator>,
58    /// Whether the index creation was aborted.
59    aborted: bool,
60    /// Statistics of index creation.
61    stats: Statistics,
62}
63
64impl FulltextIndexer {
65    /// Creates a new `FulltextIndexer`.
66    pub async fn new(
67        region_id: &RegionId,
68        sst_file_id: &FileId,
69        intermediate_manager: &IntermediateManager,
70        metadata: &RegionMetadataRef,
71        compress: bool,
72        mem_limit: usize,
73        build_type: IndexBuildType,
74    ) -> Result<Option<Self>> {
75        let mut creators = HashMap::new();
76        let global_memory_usage = Arc::new(AtomicUsize::new(0));
77
78        for column in &metadata.column_metadatas {
79            // Tag columns don't support fulltext index now.
80            // If we need to support fulltext index for tag columns, we also need to parse
81            // the codec and handle sparse encoding for flat format specially.
82            if column.semantic_type == SemanticType::Tag {
83                common_telemetry::debug!(
84                    "Skip creating fulltext index for tag column {}",
85                    column.column_schema.name
86                );
87                continue;
88            }
89
90            let options = column
91                .column_schema
92                .fulltext_options()
93                .context(IndexOptionsSnafu {
94                    column_name: &column.column_schema.name,
95                })?;
96
97            let mut targets = Vec::new();
98            if let Some(options) = options.filter(|options| options.enable) {
99                targets.push((IndexTarget::ColumnId(column.column_id), options));
100            }
101            // Only flush and compaction builds collect JSON targets; manual and schema-change
102            // rebuilds skip them because existing SSTs may not materialize the current hints.
103            // FIXME: Rebuilds replace the whole Puffin file and can drop existing JSON path indexes.
104            if matches!(build_type, IndexBuildType::Flush | IndexBuildType::Compact)
105                && column.column_schema.data_type.is_json2()
106                && let Some(extension) = column
107                    .column_schema
108                    .extension_type::<Json2ExtensionType>()
109                    .context(IndexOptionsSnafu {
110                        column_name: &column.column_schema.name,
111                    })?
112            {
113                for hint in extension.metadata().json_settings().type_hints() {
114                    let Some(options) = hint.fulltext_index.as_ref().filter(|o| o.enable) else {
115                        continue;
116                    };
117                    match IndexTarget::new_json_path(column.column_id, hint.path.clone()) {
118                        Ok(target) => targets.push((target, options.clone())),
119                        Err(err) => {
120                            warn!(
121                                err;
122                                "Skip invalid JSON fulltext index target, column_id: {}, path: {:?}",
123                                column.column_id,
124                                hint.path
125                            );
126                        }
127                    }
128                }
129            }
130            for (target, options) in targets {
131                // fulltext_path generates a fresh UUID for each target, so JSON paths
132                // in the same column use separate intermediate directories.
133                let intm_path =
134                    intermediate_manager.fulltext_path(region_id, sst_file_id, column.column_id);
135
136                let config = Config {
137                    analyzer: match options.analyzer {
138                        FulltextAnalyzer::English => Analyzer::English,
139                        FulltextAnalyzer::Chinese => Analyzer::Chinese,
140                    },
141                    case_sensitive: options.case_sensitive,
142                };
143
144                let inner = match options.backend {
145                    FulltextBackend::Tantivy => {
146                        let creator =
147                            TantivyFulltextIndexCreator::new(&intm_path, config, mem_limit)
148                                .await
149                                .context(CreateFulltextCreatorSnafu)?;
150                        AltFulltextCreator::Tantivy(creator)
151                    }
152                    FulltextBackend::Bloom => {
153                        let temp_file_provider = Arc::new(TempFileProvider::new(
154                            IntermediateLocation::new(&metadata.region_id, sst_file_id),
155                            intermediate_manager.clone(),
156                        ));
157                        let creator = BloomFilterFulltextIndexCreator::new(
158                            config,
159                            options.granularity as _,
160                            options.false_positive_rate(),
161                            temp_file_provider,
162                            global_memory_usage.clone(),
163                            Some(mem_limit),
164                        );
165                        AltFulltextCreator::Bloom(creator)
166                    }
167                };
168
169                creators.insert(
170                    target.clone(),
171                    SingleCreator {
172                        target,
173                        column_name: column.column_schema.name.clone(),
174                        inner,
175                        compress,
176                    },
177                );
178            }
179        }
180        Ok((!creators.is_empty()).then(move || Self {
181            creators,
182            aborted: false,
183            stats: Statistics::new(TYPE_FULLTEXT_INDEX),
184        }))
185    }
186
187    /// Updates the index with the given batch.
188    pub async fn update(&mut self, batch: &mut Batch) -> Result<()> {
189        ensure!(!self.aborted, OperateAbortedIndexSnafu);
190
191        if let Err(update_err) = self.do_update(batch).await {
192            if let Err(err) = self.do_abort().await {
193                if cfg!(any(test, feature = "test")) {
194                    panic!("Failed to abort index creator, err: {err}");
195                } else {
196                    warn!(err; "Failed to abort index creator");
197                }
198            }
199            return Err(update_err);
200        }
201
202        Ok(())
203    }
204
205    /// Updates the fulltext index with the given flat format RecordBatch.
206    pub async fn update_flat(&mut self, batch: &RecordBatch) -> Result<()> {
207        ensure!(!self.aborted, OperateAbortedIndexSnafu);
208
209        if batch.num_rows() == 0 {
210            return Ok(());
211        }
212
213        if let Err(update_err) = self.do_update_flat(batch).await {
214            if let Err(err) = self.do_abort().await {
215                if cfg!(any(test, feature = "test")) {
216                    panic!("Failed to abort index creator, err: {err}");
217                } else {
218                    warn!(err; "Failed to abort index creator");
219                }
220            }
221            return Err(update_err);
222        }
223
224        Ok(())
225    }
226
227    /// Finalizes the index creation.
228    pub async fn finish(
229        &mut self,
230        puffin_writer: &mut SstPuffinWriter,
231    ) -> Result<(RowCount, ByteCount)> {
232        ensure!(!self.aborted, OperateAbortedIndexSnafu);
233
234        match self.do_finish(puffin_writer).await {
235            Ok(()) => Ok((self.stats.row_count(), self.stats.byte_count())),
236            Err(finish_err) => {
237                if let Err(err) = self.do_abort().await {
238                    if cfg!(any(test, feature = "test")) {
239                        panic!("Failed to abort index creator, err: {err}");
240                    } else {
241                        warn!(err; "Failed to abort index creator");
242                    }
243                }
244                Err(finish_err)
245            }
246        }
247    }
248
249    /// Aborts the index creation.
250    pub async fn abort(&mut self) -> Result<()> {
251        if self.aborted {
252            return Ok(());
253        }
254
255        self.do_abort().await
256    }
257
258    /// Returns the memory usage of the index creator.
259    pub fn memory_usage(&self) -> usize {
260        self.creators.values().map(|c| c.inner.memory_usage()).sum()
261    }
262
263    /// Returns the targets indexed by this creator.
264    pub fn indexed_targets(&self) -> impl Iterator<Item = &IndexTarget> {
265        self.creators.keys()
266    }
267
268    /// Returns the targets indexed with the bloom backend.
269    pub fn bloom_indexed_targets(&self) -> impl Iterator<Item = &IndexTarget> {
270        self.creators
271            .values()
272            .filter(|c| matches!(c.inner, AltFulltextCreator::Bloom(_)))
273            .map(|c| &c.target)
274    }
275}
276
277impl FulltextIndexer {
278    async fn do_update(&mut self, batch: &mut Batch) -> Result<()> {
279        let mut guard = self.stats.record_update();
280        guard.inc_row_count(batch.num_rows());
281
282        for creator in self.creators.values_mut() {
283            creator.update(batch).await?;
284        }
285
286        Ok(())
287    }
288
289    async fn do_update_flat(&mut self, batch: &RecordBatch) -> Result<()> {
290        let mut guard = self.stats.record_update();
291        guard.inc_row_count(batch.num_rows());
292
293        for creator in self.creators.values_mut() {
294            creator.update_flat(batch).await?;
295        }
296
297        Ok(())
298    }
299
300    async fn do_finish(&mut self, puffin_writer: &mut SstPuffinWriter) -> Result<()> {
301        let mut guard = self.stats.record_finish();
302
303        let mut written_bytes = 0;
304        for creator in self.creators.values_mut() {
305            written_bytes += creator.finish(puffin_writer).await?;
306        }
307
308        guard.inc_byte_count(written_bytes);
309        Ok(())
310    }
311
312    async fn do_abort(&mut self) -> Result<()> {
313        let _guard = self.stats.record_cleanup();
314
315        self.aborted = true;
316
317        for (_, mut creator) in self.creators.drain() {
318            creator.abort().await?;
319        }
320
321        Ok(())
322    }
323}
324
325/// `SingleCreator` is a creator for a single column.
326struct SingleCreator {
327    /// Column or JSON path to index.
328    target: IndexTarget,
329    /// Column name.
330    column_name: String,
331    /// Inner creator.
332    inner: AltFulltextCreator,
333    /// Whether the index should be compressed.
334    compress: bool,
335}
336
337impl SingleCreator {
338    async fn update(&mut self, batch: &mut Batch) -> Result<()> {
339        let column_id = match &self.target {
340            IndexTarget::ColumnId(id) => *id,
341            IndexTarget::JsonPath(path) => {
342                let root = batch
343                    .field_col_value(path.column_id())
344                    .with_context(|| InvalidRecordBatchSnafu {
345                        reason: format!("Missing JSON fulltext column {}", path.column_id()),
346                    })?
347                    .data
348                    .to_arrow_array();
349                let values = extract_json_type_hint_values(&root, path.path())?;
350                return self.push_json_values(values).await;
351            }
352        };
353        let text_column = batch.fields().iter().find(|c| c.column_id == column_id);
354        match text_column {
355            Some(column) => {
356                let data = column
357                    .data
358                    .cast(&ConcreteDataType::string_datatype())
359                    .context(CastVectorSnafu {
360                        from: column.data.data_type(),
361                        to: ConcreteDataType::string_datatype(),
362                    })?;
363
364                for i in 0..batch.num_rows() {
365                    let data = data.get_ref(i);
366                    let text = data
367                        .try_into_string()
368                        .context(DataTypeMismatchSnafu)?
369                        .unwrap_or_default();
370                    self.inner.push_text(text).await?;
371                }
372            }
373            _ => {
374                // If the column is not found in the batch, push empty text.
375                // Ensure that the number of texts pushed is the same as the number of rows in the SST,
376                // so that the texts are aligned with the row ids.
377                for _ in 0..batch.num_rows() {
378                    self.inner.push_text("").await?;
379                }
380            }
381        }
382
383        Ok(())
384    }
385
386    async fn update_flat(&mut self, batch: &RecordBatch) -> Result<()> {
387        if let IndexTarget::JsonPath(path) = &self.target {
388            let root = batch.column_by_name(&self.column_name).with_context(|| {
389                InvalidRecordBatchSnafu {
390                    reason: format!("Missing JSON fulltext column {}", self.column_name),
391                }
392            })?;
393            let values = extract_json_type_hint_values(root, path.path())?;
394            return self.push_json_values(values).await;
395        }
396        // Find the column in the RecordBatch by name
397        if let Some(column_array) = batch.column_by_name(&self.column_name) {
398            // Convert Arrow array to string array.
399            // TODO(yingwen): Use Utf8View later if possible.
400            match column_array.data_type() {
401                DataType::Utf8 => {
402                    let string_array = column_array.as_any().downcast_ref::<StringArray>().unwrap();
403                    for text_opt in string_array.iter() {
404                        let text = text_opt.unwrap_or_default();
405                        self.inner.push_text(text).await?;
406                    }
407                }
408                DataType::LargeUtf8 => {
409                    let large_string_array = column_array
410                        .as_any()
411                        .downcast_ref::<LargeStringArray>()
412                        .unwrap();
413                    for text_opt in large_string_array.iter() {
414                        let text = text_opt.unwrap_or_default();
415                        self.inner.push_text(text).await?;
416                    }
417                }
418                _ => {
419                    // For other types, cast to Utf8 as before
420                    let array = datatypes::arrow::compute::cast(column_array, &DataType::Utf8)
421                        .context(ComputeArrowSnafu)?;
422                    let string_array = array.as_any().downcast_ref::<StringArray>().unwrap();
423                    for text_opt in string_array.iter() {
424                        let text = text_opt.unwrap_or_default();
425                        self.inner.push_text(text).await?;
426                    }
427                }
428            }
429        } else {
430            // If the column is not found in the batch, push empty text.
431            // Ensure that the number of texts pushed is the same as the number of rows in the SST,
432            // so that the texts are aligned with the row ids.
433            for _ in 0..batch.num_rows() {
434                self.inner.push_text("").await?;
435            }
436        }
437
438        Ok(())
439    }
440
441    async fn push_json_values(&mut self, values: VectorRef) -> Result<()> {
442        for row in 0..values.len() {
443            let text = values
444                .get_ref(row)
445                .try_into_string()
446                .context(DataTypeMismatchSnafu)?
447                .unwrap_or_default();
448            self.inner.push_text(text).await?;
449        }
450        Ok(())
451    }
452
453    async fn finish(&mut self, puffin_writer: &mut SstPuffinWriter) -> Result<ByteCount> {
454        let options = PutOptions {
455            compression: self.compress.then_some(CompressionCodec::Zstd),
456        };
457        self.inner
458            .finish(puffin_writer, &self.target, options)
459            .await
460    }
461
462    async fn abort(&mut self) -> Result<()> {
463        self.inner.abort(&self.target).await;
464        Ok(())
465    }
466}
467
468#[allow(dead_code, clippy::large_enum_variant)]
469/// `AltFulltextCreator` is an alternative fulltext index creator that can be either Tantivy or BloomFilter.
470enum AltFulltextCreator {
471    Tantivy(TantivyFulltextIndexCreator),
472    Bloom(BloomFilterFulltextIndexCreator),
473}
474
475impl AltFulltextCreator {
476    async fn push_text(&mut self, text: &str) -> Result<()> {
477        match self {
478            Self::Tantivy(creator) => creator.push_text(text).await.context(FulltextPushTextSnafu),
479            Self::Bloom(creator) => creator.push_text(text).await.context(FulltextPushTextSnafu),
480        }
481    }
482
483    fn memory_usage(&self) -> usize {
484        match self {
485            Self::Tantivy(creator) => creator.memory_usage(),
486            Self::Bloom(creator) => creator.memory_usage(),
487        }
488    }
489
490    async fn finish(
491        &mut self,
492        puffin_writer: &mut SstPuffinWriter,
493        target: &IndexTarget,
494        put_options: PutOptions,
495    ) -> Result<ByteCount> {
496        let target_key = target.encode();
497        match self {
498            Self::Tantivy(creator) => {
499                let blob_key = format!("{INDEX_BLOB_TYPE_TANTIVY}-{target_key}");
500                creator
501                    .finish(puffin_writer, &blob_key, put_options)
502                    .await
503                    .context(FulltextFinishSnafu)
504            }
505            Self::Bloom(creator) => {
506                let blob_key = format!("{INDEX_BLOB_TYPE_BLOOM}-{target_key}");
507                creator
508                    .finish(puffin_writer, &blob_key, put_options)
509                    .await
510                    .context(FulltextFinishSnafu)
511            }
512        }
513    }
514
515    async fn abort(&mut self, target: &IndexTarget) {
516        match self {
517            Self::Tantivy(creator) => {
518                if let Err(err) = creator.abort().await {
519                    warn!(err; "Failed to abort the fulltext index creator in the Tantivy flavor, target: {:?}", target);
520                }
521            }
522            Self::Bloom(creator) => {
523                if let Err(err) = creator.abort().await {
524                    warn!(err; "Failed to abort the fulltext index creator in the Bloom Filter flavor, target: {:?}", target);
525                }
526            }
527        }
528    }
529}
530
531#[cfg(test)]
532mod tests {
533    use std::collections::{BTreeMap, BTreeSet};
534    use std::sync::Arc;
535
536    use api::v1::SemanticType;
537    use common_base::BitVec;
538    use datatypes::arrow::array::{ArrayRef, StructArray};
539    use datatypes::arrow::buffer::NullBuffer;
540    use datatypes::arrow::datatypes::{Field, Schema};
541    use datatypes::data_type::DataType;
542    use datatypes::extension::json::{Json2ExtensionType, JsonMetadata};
543    use datatypes::json::{JsonSettings, JsonTypeHint};
544    use datatypes::schema::{ColumnSchema, FulltextAnalyzer, FulltextOptions};
545    use datatypes::types::json_type::{JsonNativeType, JsonObjectType};
546    use datatypes::vectors::{Helper, UInt8Vector, UInt64Vector};
547    use futures::FutureExt;
548    use futures::future::BoxFuture;
549    use index::fulltext_index::search::RowId;
550    use object_store::ObjectStore;
551    use object_store::services::Memory;
552    use puffin::puffin_manager::{PuffinManager, PuffinWriter};
553    use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder, RegionMetadataRef};
554    use store_api::region_request::PathType;
555    use store_api::storage::{ColumnId, ConcreteDataType, FileId, RegionId};
556
557    use super::*;
558    use crate::access_layer::RegionFilePathFactory;
559    use crate::read::{Batch, BatchColumn};
560    use crate::sst::file::{RegionFileId, RegionIndexId};
561    use crate::sst::index::fulltext_index::applier::FulltextIndexApplier;
562    use crate::sst::index::fulltext_index::applier::builder::{
563        FulltextQuery, FulltextRequest, FulltextTerm,
564    };
565    use crate::sst::index::puffin_manager::PuffinManagerFactory;
566
567    #[tokio::test]
568    async fn test_json_fulltext_build_and_prune_both_formats() {
569        for backend in [FulltextBackend::Bloom, FulltextBackend::Tantivy] {
570            for (flat, build_type) in [
571                (false, IndexBuildType::Flush),
572                (false, IndexBuildType::Compact),
573                (true, IndexBuildType::Flush),
574                (true, IndexBuildType::Compact),
575            ] {
576                let (dir, factory) =
577                    PuffinManagerFactory::new_for_test_async("json_fulltext").await;
578                let options = FulltextOptions::new(
579                    true,
580                    FulltextAnalyzer::English,
581                    false,
582                    backend.clone(),
583                    1,
584                    0.01,
585                )
586                .unwrap();
587                let hints = ["message", "other/\".field"].map(|name| JsonTypeHint {
588                    path: vec![name.into()],
589                    data_type: ConcreteDataType::string_datatype(),
590                    inverted_index: false,
591                    skipping_index: None,
592                    fulltext_index: Some(options.clone()),
593                });
594                let mut column = ColumnSchema::new(
595                    "attrs",
596                    ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())),
597                    true,
598                );
599                column.with_extension_type(&Json2ExtensionType::new(Arc::new(JsonMetadata::new(
600                    JsonSettings::try_new(hints.to_vec(), Some(0)).unwrap(),
601                ))));
602                let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 2));
603                builder.push_column_metadata(ColumnMetadata {
604                    column_schema: column,
605                    semantic_type: SemanticType::Field,
606                    column_id: 1,
607                });
608                builder.push_column_metadata(ColumnMetadata {
609                    column_schema: ColumnSchema::new(
610                        "ts",
611                        ConcreteDataType::timestamp_millisecond_datatype(),
612                        false,
613                    ),
614                    semantic_type: SemanticType::Timestamp,
615                    column_id: 2,
616                });
617                let metadata = Arc::new(builder.build().unwrap());
618                let file_id = FileId::random();
619                let manager = new_intm_mgr(dir.path().to_string_lossy()).await;
620                for rebuild_type in [IndexBuildType::Manual, IndexBuildType::SchemaChange] {
621                    assert!(
622                        FulltextIndexer::new(
623                            &metadata.region_id,
624                            &file_id,
625                            &manager,
626                            &metadata,
627                            false,
628                            1024,
629                            rebuild_type,
630                        )
631                        .await
632                        .unwrap()
633                        .is_none()
634                    );
635                }
636                let mut creator = FulltextIndexer::new(
637                    &metadata.region_id,
638                    &file_id,
639                    &manager,
640                    &metadata,
641                    false,
642                    1024,
643                    build_type,
644                )
645                .await
646                .unwrap()
647                .unwrap();
648                let messages: ArrayRef = Arc::new(StringArray::from(vec![
649                    Some("alpha"),
650                    None,
651                    Some("beta"),
652                    Some("alpha"),
653                ]));
654                let other: ArrayRef = Arc::new(StringArray::from(vec![
655                    Some("beta"),
656                    None,
657                    Some("alpha"),
658                    Some(""),
659                ]));
660                let root: ArrayRef = Arc::new(StructArray::new(
661                    vec![
662                        Field::new("message", messages.data_type().clone(), true),
663                        Field::new("other/\".field", other.data_type().clone(), true),
664                    ]
665                    .into(),
666                    vec![messages, other],
667                    Some(NullBuffer::from(vec![true, false, true, true])),
668                ));
669                for start in [0, 2] {
670                    let array = root.slice(start, 2);
671                    if flat {
672                        let batch = RecordBatch::try_new(
673                            Arc::new(Schema::new(vec![Field::new(
674                                "attrs",
675                                array.data_type().clone(),
676                                true,
677                            )])),
678                            vec![array],
679                        )
680                        .unwrap();
681                        creator.update_flat(&batch).await.unwrap();
682                    } else {
683                        let mut batch = Batch::new(
684                            vec![],
685                            Arc::new(UInt64Vector::from_vec(vec![0, 1])),
686                            Arc::new(UInt64Vector::from_vec(vec![0, 0])),
687                            Arc::new(UInt8Vector::from_vec(vec![1, 1])),
688                            vec![BatchColumn {
689                                column_id: 1,
690                                data: Helper::try_into_vector(array).unwrap(),
691                            }],
692                        )
693                        .unwrap();
694                        creator.update(&mut batch).await.unwrap();
695                    }
696                }
697                assert_eq!(creator.indexed_targets().count(), hints.len());
698                assert!(
699                    creator
700                        .indexed_targets()
701                        .all(|target| matches!(target, IndexTarget::JsonPath(_)))
702                );
703                let bloom_targets: BTreeSet<_> = creator.bloom_indexed_targets().collect();
704                if backend == FulltextBackend::Bloom {
705                    assert_eq!(bloom_targets, creator.indexed_targets().collect());
706                } else {
707                    assert!(bloom_targets.is_empty());
708                }
709                let store = mock_object_store();
710                let puffin = factory.build(
711                    store.clone(),
712                    RegionFilePathFactory::new("json_test".into(), PathType::Bare),
713                );
714                let index_id =
715                    RegionIndexId::new(RegionFileId::new(metadata.region_id, file_id), 0);
716                let mut writer = puffin.writer(&index_id).await.unwrap();
717                let (rows, bytes) = creator.finish(&mut writer).await.unwrap();
718                assert_eq!(rows, 4);
719                assert!(bytes > 0);
720                writer.finish().await.unwrap();
721                let cache = Arc::new(
722                    crate::cache::index::bloom_filter_index::BloomFilterIndexCache::new(
723                        1024 * 1024,
724                        1024 * 1024,
725                        1024,
726                    ),
727                );
728                for (path, term, expected) in [
729                    ("message", "alpha", vec![0, 3]),
730                    ("other/\".field", "alpha", vec![2]),
731                    ("message", "hidden", vec![]),
732                    ("message", "absent", vec![]),
733                ] {
734                    let target = IndexTarget::new_json_path(1, vec![path.into()]).unwrap();
735                    let request = match backend {
736                        FulltextBackend::Tantivy => FulltextRequest {
737                            queries: vec![FulltextQuery(term.into())],
738                            ..Default::default()
739                        },
740                        FulltextBackend::Bloom => FulltextRequest {
741                            terms: vec![FulltextTerm {
742                                col_lowered: false,
743                                term: term.into(),
744                            }],
745                            ..Default::default()
746                        },
747                    };
748                    let applier = FulltextIndexApplier::new(
749                        "json_test".into(),
750                        PathType::Bare,
751                        store.clone(),
752                        BTreeMap::from([
753                            (target, request.clone()),
754                            (
755                                IndexTarget::new_json_path(1, vec!["missing".into()]).unwrap(),
756                                request,
757                            ),
758                        ]),
759                        factory.clone(),
760                        &metadata,
761                    )
762                    .with_bloom_filter_cache(Some(cache.clone()));
763                    let requests = applier.compatible_requests_for_sst(&metadata).unwrap();
764                    assert_eq!(requests.len(), 1);
765                    for change in 0..3 {
766                        let mut changed_hints = hints.to_vec();
767                        for hint in &mut changed_hints {
768                            let options = hint.fulltext_index.as_mut().unwrap();
769                            match change {
770                                0 => options.analyzer = FulltextAnalyzer::Chinese,
771                                1 => options.case_sensitive = true,
772                                _ => {
773                                    options.backend = match backend {
774                                        FulltextBackend::Bloom => FulltextBackend::Tantivy,
775                                        FulltextBackend::Tantivy => FulltextBackend::Bloom,
776                                    }
777                                }
778                            }
779                        }
780                        let mut changed = (*metadata).clone();
781                        changed.column_metadatas[0]
782                            .column_schema
783                            .with_extension_type(&Json2ExtensionType::new(Arc::new(
784                                JsonMetadata::new(
785                                    JsonSettings::try_new(changed_hints, Some(0)).unwrap(),
786                                ),
787                            )));
788                        assert_eq!(
789                            applier.compatible_requests_for_sst(&changed).is_some(),
790                            change == 0 && backend == FulltextBackend::Bloom,
791                        );
792                    }
793                    assert!(
794                        applier
795                            .compatible_requests_for_sst(&mock_region_metadata(backend.clone()))
796                            .is_none()
797                    );
798                    for _ in 0..2 {
799                        match backend {
800                            FulltextBackend::Tantivy => {
801                                let rows = applier
802                                    .apply_fine(index_id, None, &requests, None)
803                                    .await
804                                    .unwrap()
805                                    .unwrap();
806                                assert_eq!(
807                                    rows.into_iter().collect::<Vec<_>>(),
808                                    expected.iter().map(|v| *v as RowId).collect::<Vec<_>>()
809                                );
810                            }
811                            FulltextBackend::Bloom => {
812                                let ranges = applier
813                                    .apply_coarse(
814                                        index_id,
815                                        None,
816                                        [(4, true)].into_iter(),
817                                        &requests,
818                                        None,
819                                    )
820                                    .await
821                                    .unwrap()
822                                    .unwrap();
823                                let rows: Vec<_> = ranges[0].1.iter().cloned().flatten().collect();
824                                // A Bloom filter may retain extra rows, but must retain every match.
825                                assert!(expected.iter().all(|row| rows.contains(row)));
826                                if expected.is_empty() {
827                                    assert!(
828                                        rows.len() < 4,
829                                        "path={path}, term={term}, flat={flat}, rows={rows:?}"
830                                    );
831                                }
832                            }
833                        }
834                    }
835                }
836            }
837        }
838    }
839
840    fn mock_object_store() -> ObjectStore {
841        ObjectStore::new(Memory::default()).unwrap()
842    }
843
844    async fn new_intm_mgr(path: impl AsRef<str>) -> IntermediateManager {
845        IntermediateManager::init_fs(path).await.unwrap()
846    }
847
848    fn mock_region_metadata(backend: FulltextBackend) -> RegionMetadataRef {
849        let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 2));
850        builder
851            .push_column_metadata(ColumnMetadata {
852                column_schema: ColumnSchema::new(
853                    "text_english_case_sensitive",
854                    ConcreteDataType::string_datatype(),
855                    true,
856                )
857                .with_fulltext_options(FulltextOptions::new_unchecked(
858                    true,
859                    FulltextAnalyzer::English,
860                    true,
861                    backend.clone(),
862                    1,
863                    0.01,
864                ))
865                .unwrap(),
866                semantic_type: SemanticType::Field,
867                column_id: 1,
868            })
869            .push_column_metadata(ColumnMetadata {
870                column_schema: ColumnSchema::new(
871                    "text_english_case_insensitive",
872                    ConcreteDataType::string_datatype(),
873                    true,
874                )
875                .with_fulltext_options(FulltextOptions::new_unchecked(
876                    true,
877                    FulltextAnalyzer::English,
878                    false,
879                    backend.clone(),
880                    1,
881                    0.01,
882                ))
883                .unwrap(),
884                semantic_type: SemanticType::Field,
885                column_id: 2,
886            })
887            .push_column_metadata(ColumnMetadata {
888                column_schema: ColumnSchema::new(
889                    "text_chinese",
890                    ConcreteDataType::string_datatype(),
891                    true,
892                )
893                .with_fulltext_options(FulltextOptions::new_unchecked(
894                    true,
895                    FulltextAnalyzer::Chinese,
896                    false,
897                    backend.clone(),
898                    1,
899                    0.01,
900                ))
901                .unwrap(),
902                semantic_type: SemanticType::Field,
903                column_id: 3,
904            })
905            .push_column_metadata(ColumnMetadata {
906                column_schema: ColumnSchema::new(
907                    "ts",
908                    ConcreteDataType::timestamp_millisecond_datatype(),
909                    false,
910                ),
911                semantic_type: SemanticType::Timestamp,
912                column_id: 4,
913            });
914
915        Arc::new(builder.build().unwrap())
916    }
917
918    fn new_batch(
919        rows: &[(
920            Option<&str>, // text_english_case_sensitive
921            Option<&str>, // text_english_case_insensitive
922            Option<&str>, // text_chinese
923        )],
924    ) -> Batch {
925        let mut vec_english_sensitive =
926            ConcreteDataType::string_datatype().create_mutable_vector(0);
927        let mut vec_english_insensitive =
928            ConcreteDataType::string_datatype().create_mutable_vector(0);
929        let mut vec_chinese = ConcreteDataType::string_datatype().create_mutable_vector(0);
930
931        for (text_english_case_sensitive, text_english_case_insensitive, text_chinese) in rows {
932            match text_english_case_sensitive {
933                Some(s) => vec_english_sensitive.push_value_ref(&(*s).into()),
934                None => vec_english_sensitive.push_null(),
935            }
936            match text_english_case_insensitive {
937                Some(s) => vec_english_insensitive.push_value_ref(&(*s).into()),
938                None => vec_english_insensitive.push_null(),
939            }
940            match text_chinese {
941                Some(s) => vec_chinese.push_value_ref(&(*s).into()),
942                None => vec_chinese.push_null(),
943            }
944        }
945
946        let num_rows = vec_english_sensitive.len();
947        Batch::new(
948            vec![],
949            Arc::new(UInt64Vector::from_iter_values(
950                (0..num_rows).map(|n| n as u64),
951            )),
952            Arc::new(UInt64Vector::from_iter_values(std::iter::repeat_n(
953                0, num_rows,
954            ))),
955            Arc::new(UInt8Vector::from_iter_values(std::iter::repeat_n(
956                1, num_rows,
957            ))),
958            vec![
959                BatchColumn {
960                    column_id: 1,
961                    data: vec_english_sensitive.to_vector(),
962                },
963                BatchColumn {
964                    column_id: 2,
965                    data: vec_english_insensitive.to_vector(),
966                },
967                BatchColumn {
968                    column_id: 3,
969                    data: vec_chinese.to_vector(),
970                },
971            ],
972        )
973        .unwrap()
974    }
975
976    /// Applier factory that can handle both queries and terms.
977    ///
978    /// It builds a fulltext index with the given data rows, and returns a function
979    /// that can handle both queries and terms in a single request.
980    ///
981    /// The function takes two parameters:
982    /// - `queries`: A list of (ColumnId, query_string) pairs for fulltext queries
983    /// - `terms`: A list of (ColumnId, [(bool, String)]) for fulltext terms, where bool indicates if term is lowercased
984    async fn build_fulltext_applier_factory(
985        prefix: &str,
986        backend: FulltextBackend,
987        rows: &[(
988            Option<&str>, // text_english_case_sensitive
989            Option<&str>, // text_english_case_insensitive
990            Option<&str>, // text_chinese
991        )],
992    ) -> impl Fn(
993        Vec<(ColumnId, &str)>,
994        Vec<(ColumnId, Vec<(bool, &str)>)>,
995        Option<BitVec>,
996    ) -> BoxFuture<'static, Option<BTreeSet<RowId>>> {
997        let (d, factory) = PuffinManagerFactory::new_for_test_async(prefix).await;
998        let table_dir = "table0".to_string();
999        let sst_file_id = FileId::random();
1000        let object_store = mock_object_store();
1001        let region_metadata = mock_region_metadata(backend.clone());
1002        let intm_mgr = new_intm_mgr(d.path().to_string_lossy()).await;
1003
1004        let mut indexer = FulltextIndexer::new(
1005            &region_metadata.region_id,
1006            &sst_file_id,
1007            &intm_mgr,
1008            &region_metadata,
1009            true,
1010            1024,
1011            IndexBuildType::Flush,
1012        )
1013        .await
1014        .unwrap()
1015        .unwrap();
1016
1017        let mut batch = new_batch(rows);
1018        indexer.update(&mut batch).await.unwrap();
1019
1020        let puffin_manager = factory.build(
1021            object_store.clone(),
1022            RegionFilePathFactory::new(table_dir.clone(), PathType::Bare),
1023        );
1024        let region_file_id = RegionFileId::new(region_metadata.region_id, sst_file_id);
1025        let index_id = RegionIndexId::new(region_file_id, 0);
1026        let mut writer = puffin_manager.writer(&index_id).await.unwrap();
1027        let _ = indexer.finish(&mut writer).await.unwrap();
1028        writer.finish().await.unwrap();
1029
1030        move |queries: Vec<(ColumnId, &str)>,
1031              terms_requests: Vec<(ColumnId, Vec<(bool, &str)>)>,
1032              coarse_mask: Option<BitVec>| {
1033            let _d = &d;
1034            let table_dir = table_dir.clone();
1035            let object_store = object_store.clone();
1036            let factory = factory.clone();
1037
1038            let mut requests: BTreeMap<IndexTarget, FulltextRequest> = BTreeMap::new();
1039
1040            // Add queries
1041            for (column_id, query) in queries {
1042                requests
1043                    .entry(IndexTarget::ColumnId(column_id))
1044                    .or_default()
1045                    .queries
1046                    .push(FulltextQuery(query.to_string()));
1047            }
1048
1049            // Add terms
1050            for (column_id, terms) in terms_requests {
1051                let fulltext_terms = terms
1052                    .into_iter()
1053                    .map(|(col_lowered, term)| FulltextTerm {
1054                        col_lowered,
1055                        term: term.to_string(),
1056                    })
1057                    .collect::<Vec<_>>();
1058
1059                requests
1060                    .entry(IndexTarget::ColumnId(column_id))
1061                    .or_default()
1062                    .terms
1063                    .extend(fulltext_terms);
1064            }
1065
1066            let applier = FulltextIndexApplier::new(
1067                table_dir,
1068                PathType::Bare,
1069                object_store,
1070                requests,
1071                factory,
1072                &region_metadata,
1073            );
1074
1075            let requests = applier
1076                .compatible_requests_for_sst(&region_metadata)
1077                .unwrap();
1078            let backend = backend.clone();
1079            async move {
1080                match backend {
1081                    FulltextBackend::Tantivy => applier
1082                        .apply_fine(index_id, None, &requests, None)
1083                        .await
1084                        .unwrap(),
1085                    FulltextBackend::Bloom => {
1086                        let coarse_mask = coarse_mask.unwrap_or_default();
1087                        let row_groups = (0..coarse_mask.len()).map(|i| (1, coarse_mask[i]));
1088                        // row group id == row id
1089                        let resp = applier
1090                            .apply_coarse(index_id, None, row_groups, &requests, None)
1091                            .await
1092                            .unwrap();
1093                        resp.map(|r| {
1094                            r.into_iter()
1095                                .filter(|(_, ranges)| !ranges.is_empty())
1096                                .map(|(row_group_id, _)| row_group_id as RowId)
1097                                .collect()
1098                        })
1099                    }
1100                }
1101            }
1102            .boxed()
1103        }
1104    }
1105
1106    fn rows(row_ids: impl IntoIterator<Item = RowId>) -> BTreeSet<RowId> {
1107        row_ids.into_iter().collect()
1108    }
1109
1110    #[tokio::test]
1111    async fn test_fulltext_bloom_shared_memory_spills() {
1112        let dir = common_test_util::temp_dir::create_temp_dir("shared_fulltext_memory");
1113        let metadata = mock_region_metadata(FulltextBackend::Bloom);
1114        let manager = new_intm_mgr(dir.path().to_string_lossy()).await;
1115        let mut indexer = FulltextIndexer::new(
1116            &metadata.region_id,
1117            &FileId::random(),
1118            &manager,
1119            &metadata,
1120            false,
1121            2 * 1024 * 1024,
1122            IndexBuildType::Flush,
1123        )
1124        .await
1125        .unwrap()
1126        .unwrap();
1127        // Each target stays below 2 MB on its own; together they exceed the
1128        // shared threshold and must spill finalized Bloom filters to disk.
1129        for creator in indexer.creators.values_mut() {
1130            for row in 0..20_000 {
1131                creator.inner.push_text(&format!("row{row}")).await.unwrap();
1132            }
1133        }
1134        let mut directories = vec![dir.path().join("__intm")];
1135        let mut spilled = false;
1136        while let Some(directory) = directories.pop() {
1137            for entry in std::fs::read_dir(directory).unwrap() {
1138                let entry = entry.unwrap();
1139                if entry.file_type().unwrap().is_dir() {
1140                    directories.push(entry.path());
1141                } else {
1142                    spilled = true;
1143                }
1144            }
1145        }
1146        assert!(spilled, "combined memory usage must trigger spilling");
1147        indexer.abort().await.unwrap();
1148    }
1149
1150    #[tokio::test]
1151    async fn test_fulltext_index_basic_case_sensitive_tantivy() {
1152        let applier_factory = build_fulltext_applier_factory(
1153            "test_fulltext_index_basic_case_sensitive_tantivy_",
1154            FulltextBackend::Tantivy,
1155            &[
1156                (Some("hello"), None, None),
1157                (Some("world"), None, None),
1158                (None, None, None),
1159                (Some("Hello, World"), None, None),
1160            ],
1161        )
1162        .await;
1163
1164        let row_ids = applier_factory(vec![(1, "hello")], vec![], None).await;
1165        assert_eq!(row_ids, Some(rows([0])));
1166
1167        let row_ids = applier_factory(vec![(1, "world")], vec![], None).await;
1168        assert_eq!(row_ids, Some(rows([1])));
1169
1170        let row_ids = applier_factory(vec![(1, "Hello")], vec![], None).await;
1171        assert_eq!(row_ids, Some(rows([3])));
1172
1173        let row_ids = applier_factory(vec![(1, "World")], vec![], None).await;
1174        assert_eq!(row_ids, Some(rows([3])));
1175
1176        // Terms are left to the bloom backend.
1177        let row_ids = applier_factory(vec![], vec![(1, vec![(false, "hello")])], None).await;
1178        assert_eq!(row_ids, None);
1179    }
1180
1181    #[tokio::test]
1182    async fn test_fulltext_index_basic_case_sensitive_bloom() {
1183        let applier_factory = build_fulltext_applier_factory(
1184            "test_fulltext_index_basic_case_sensitive_bloom_",
1185            FulltextBackend::Bloom,
1186            &[
1187                (Some("hello"), None, None),
1188                (Some("world"), None, None),
1189                (None, None, None),
1190                (Some("Hello, World"), None, None),
1191            ],
1192        )
1193        .await;
1194
1195        let row_ids = applier_factory(
1196            vec![],
1197            vec![(1, vec![(false, "hello")])],
1198            Some(BitVec::from_slice(&[0b1111])),
1199        )
1200        .await;
1201        assert_eq!(row_ids, Some(rows([0])));
1202
1203        let row_ids = applier_factory(
1204            vec![],
1205            vec![(1, vec![(false, "hello")])],
1206            Some(BitVec::from_slice(&[0b1110])), // row 0 is filtered out
1207        )
1208        .await;
1209        assert_eq!(row_ids, Some(rows([])));
1210
1211        let row_ids = applier_factory(
1212            vec![],
1213            vec![(1, vec![(true, "hello")])],
1214            Some(BitVec::from_slice(&[0b1111])),
1215        )
1216        .await;
1217        assert_eq!(row_ids, None);
1218
1219        let row_ids = applier_factory(
1220            vec![],
1221            vec![(1, vec![(false, "world")])],
1222            Some(BitVec::from_slice(&[0b1111])),
1223        )
1224        .await;
1225        assert_eq!(row_ids, Some(rows([1])));
1226
1227        let row_ids = applier_factory(
1228            vec![],
1229            vec![(1, vec![(false, "world")])],
1230            Some(BitVec::from_slice(&[0b1101])), // row 1 is filtered out
1231        )
1232        .await;
1233        assert_eq!(row_ids, Some(rows([])));
1234
1235        let row_ids = applier_factory(
1236            vec![],
1237            vec![(1, vec![(true, "world")])],
1238            Some(BitVec::from_slice(&[0b1111])),
1239        )
1240        .await;
1241        assert_eq!(row_ids, None);
1242
1243        let row_ids = applier_factory(
1244            vec![],
1245            vec![(1, vec![(false, "Hello")])],
1246            Some(BitVec::from_slice(&[0b1111])),
1247        )
1248        .await;
1249        assert_eq!(row_ids, Some(rows([3])));
1250
1251        let row_ids = applier_factory(
1252            vec![],
1253            vec![(1, vec![(false, "Hello")])],
1254            Some(BitVec::from_slice(&[0b0111])), // row 3 is filtered out
1255        )
1256        .await;
1257        assert_eq!(row_ids, Some(rows([])));
1258
1259        let row_ids = applier_factory(
1260            vec![],
1261            vec![(1, vec![(true, "Hello")])],
1262            Some(BitVec::from_slice(&[0b1111])),
1263        )
1264        .await;
1265        assert_eq!(row_ids, None);
1266
1267        let row_ids = applier_factory(
1268            vec![],
1269            vec![(1, vec![(false, "Hello, World")])],
1270            Some(BitVec::from_slice(&[0b1111])),
1271        )
1272        .await;
1273        assert_eq!(row_ids, Some(rows([3])));
1274
1275        let row_ids = applier_factory(
1276            vec![],
1277            vec![(1, vec![(false, "Hello, World")])],
1278            Some(BitVec::from_slice(&[0b0111])), // row 3 is filtered out
1279        )
1280        .await;
1281        assert_eq!(row_ids, Some(rows([])));
1282
1283        let row_ids = applier_factory(
1284            vec![],
1285            vec![(1, vec![(true, "Hello, World")])],
1286            Some(BitVec::from_slice(&[0b1111])),
1287        )
1288        .await;
1289        assert_eq!(row_ids, None);
1290    }
1291
1292    #[tokio::test]
1293    async fn test_fulltext_index_basic_case_insensitive_tantivy() {
1294        let applier_factory = build_fulltext_applier_factory(
1295            "test_fulltext_index_basic_case_insensitive_tantivy_",
1296            FulltextBackend::Tantivy,
1297            &[
1298                (None, Some("hello"), None),
1299                (None, None, None),
1300                (None, Some("world"), None),
1301                (None, Some("Hello, World"), None),
1302            ],
1303        )
1304        .await;
1305
1306        let row_ids = applier_factory(vec![(2, "hello")], vec![], None).await;
1307        assert_eq!(row_ids, Some(rows([0, 3])));
1308
1309        let row_ids = applier_factory(vec![(2, "world")], vec![], None).await;
1310        assert_eq!(row_ids, Some(rows([2, 3])));
1311
1312        let row_ids = applier_factory(vec![(2, "Hello")], vec![], None).await;
1313        assert_eq!(row_ids, Some(rows([0, 3])));
1314
1315        let row_ids = applier_factory(vec![(2, "World")], vec![], None).await;
1316        assert_eq!(row_ids, Some(rows([2, 3])));
1317    }
1318
1319    #[tokio::test]
1320    async fn test_fulltext_index_basic_case_insensitive_bloom() {
1321        let applier_factory = build_fulltext_applier_factory(
1322            "test_fulltext_index_basic_case_insensitive_bloom_",
1323            FulltextBackend::Bloom,
1324            &[
1325                (None, Some("hello"), None),
1326                (None, None, None),
1327                (None, Some("world"), None),
1328                (None, Some("Hello, World"), None),
1329            ],
1330        )
1331        .await;
1332
1333        let row_ids = applier_factory(
1334            vec![],
1335            vec![(2, vec![(false, "hello")])],
1336            Some(BitVec::from_slice(&[0b1111])),
1337        )
1338        .await;
1339        assert_eq!(row_ids, Some(rows([0, 3])));
1340
1341        let row_ids = applier_factory(
1342            vec![],
1343            vec![(2, vec![(false, "hello")])],
1344            Some(BitVec::from_slice(&[0b1110])), // row 0 is filtered out
1345        )
1346        .await;
1347        assert_eq!(row_ids, Some(rows([3])));
1348
1349        let row_ids = applier_factory(
1350            vec![],
1351            vec![(2, vec![(false, "world")])],
1352            Some(BitVec::from_slice(&[0b1111])),
1353        )
1354        .await;
1355        assert_eq!(row_ids, Some(rows([2, 3])));
1356
1357        let row_ids = applier_factory(
1358            vec![],
1359            vec![(2, vec![(false, "world")])],
1360            Some(BitVec::from_slice(&[0b1011])), // row 2 is filtered out
1361        )
1362        .await;
1363        assert_eq!(row_ids, Some(rows([3])));
1364
1365        let row_ids = applier_factory(
1366            vec![],
1367            vec![(2, vec![(false, "Hello")])],
1368            Some(BitVec::from_slice(&[0b1111])),
1369        )
1370        .await;
1371        assert_eq!(row_ids, Some(rows([0, 3])));
1372
1373        let row_ids = applier_factory(
1374            vec![],
1375            vec![(2, vec![(false, "Hello")])],
1376            Some(BitVec::from_slice(&[0b0111])), // row 3 is filtered out
1377        )
1378        .await;
1379        assert_eq!(row_ids, Some(rows([0])));
1380
1381        let row_ids = applier_factory(
1382            vec![],
1383            vec![(2, vec![(false, "World")])],
1384            Some(BitVec::from_slice(&[0b1111])),
1385        )
1386        .await;
1387        assert_eq!(row_ids, Some(rows([2, 3])));
1388
1389        let row_ids = applier_factory(
1390            vec![],
1391            vec![(2, vec![(false, "World")])],
1392            Some(BitVec::from_slice(&[0b0111])), // row 3 is filtered out
1393        )
1394        .await;
1395        assert_eq!(row_ids, Some(rows([2])));
1396
1397        let row_ids = applier_factory(
1398            vec![],
1399            vec![(2, vec![(true, "hello")])],
1400            Some(BitVec::from_slice(&[0b1111])),
1401        )
1402        .await;
1403        assert_eq!(row_ids, Some(rows([0, 3])));
1404    }
1405
1406    #[tokio::test]
1407    async fn test_fulltext_index_basic_chinese_tantivy() {
1408        let applier_factory = build_fulltext_applier_factory(
1409            "test_fulltext_index_basic_chinese_tantivy_",
1410            FulltextBackend::Tantivy,
1411            &[
1412                (None, None, Some("你好")),
1413                (None, None, None),
1414                (None, None, Some("世界")),
1415                (None, None, Some("你好，世界")),
1416            ],
1417        )
1418        .await;
1419
1420        let row_ids = applier_factory(vec![(3, "你好")], vec![], None).await;
1421        assert_eq!(row_ids, Some(rows([0, 3])));
1422
1423        let row_ids = applier_factory(vec![(3, "世界")], vec![], None).await;
1424        assert_eq!(row_ids, Some(rows([2, 3])));
1425    }
1426
1427    #[tokio::test]
1428    async fn test_fulltext_index_basic_chinese_bloom() {
1429        let applier_factory = build_fulltext_applier_factory(
1430            "test_fulltext_index_basic_chinese_bloom_",
1431            FulltextBackend::Bloom,
1432            &[
1433                (None, None, Some("你好")),
1434                (None, None, None),
1435                (None, None, Some("世界")),
1436                (None, None, Some("你好，世界")),
1437            ],
1438        )
1439        .await;
1440
1441        let row_ids = applier_factory(
1442            vec![],
1443            vec![(3, vec![(false, "你好")])],
1444            Some(BitVec::from_slice(&[0b1111])),
1445        )
1446        .await;
1447        assert_eq!(row_ids, Some(rows([0, 3])));
1448
1449        let row_ids = applier_factory(
1450            vec![],
1451            vec![(3, vec![(false, "你好")])],
1452            Some(BitVec::from_slice(&[0b1110])), // row 0 is filtered out
1453        )
1454        .await;
1455        assert_eq!(row_ids, Some(rows([3])));
1456
1457        let row_ids = applier_factory(
1458            vec![],
1459            vec![(3, vec![(false, "世界")])],
1460            Some(BitVec::from_slice(&[0b1111])),
1461        )
1462        .await;
1463        assert_eq!(row_ids, Some(rows([2, 3])));
1464
1465        let row_ids = applier_factory(
1466            vec![],
1467            vec![(3, vec![(false, "世界")])],
1468            Some(BitVec::from_slice(&[0b1011])), // row 2 is filtered out
1469        )
1470        .await;
1471        assert_eq!(row_ids, Some(rows([3])));
1472    }
1473
1474    #[tokio::test]
1475    async fn test_fulltext_index_multi_terms_case_sensitive_bloom() {
1476        let applier_factory = build_fulltext_applier_factory(
1477            "test_fulltext_index_multi_terms_case_sensitive_bloom_",
1478            FulltextBackend::Bloom,
1479            &[
1480                (Some("Hello"), None, None),
1481                (Some("World"), None, None),
1482                (None, None, None),
1483                (Some("Hello, World"), None, None),
1484            ],
1485        )
1486        .await;
1487
1488        let row_ids = applier_factory(
1489            vec![],
1490            vec![(1, vec![(false, "hello"), (false, "world")])],
1491            Some(BitVec::from_slice(&[0b1111])),
1492        )
1493        .await;
1494        assert_eq!(row_ids, Some(rows([])));
1495
1496        let row_ids = applier_factory(
1497            vec![],
1498            vec![(1, vec![(false, "Hello"), (false, "World")])],
1499            Some(BitVec::from_slice(&[0b1111])),
1500        )
1501        .await;
1502        assert_eq!(row_ids, Some(rows([3])));
1503
1504        let row_ids = applier_factory(
1505            vec![],
1506            vec![(1, vec![(true, "Hello"), (false, "World")])],
1507            Some(BitVec::from_slice(&[0b1111])),
1508        )
1509        .await;
1510        assert_eq!(row_ids, Some(rows([1, 3])));
1511
1512        let row_ids = applier_factory(
1513            vec![],
1514            vec![(1, vec![(false, "Hello"), (true, "World")])],
1515            Some(BitVec::from_slice(&[0b1111])),
1516        )
1517        .await;
1518        assert_eq!(row_ids, Some(rows([0, 3])));
1519
1520        let row_ids = applier_factory(
1521            vec![],
1522            vec![(1, vec![(true, "Hello"), (true, "World")])],
1523            Some(BitVec::from_slice(&[0b1111])),
1524        )
1525        .await;
1526        assert_eq!(row_ids, None);
1527    }
1528
1529    #[tokio::test]
1530    async fn test_fulltext_index_multi_terms_case_insensitive_bloom() {
1531        let applier_factory = build_fulltext_applier_factory(
1532            "test_fulltext_index_multi_terms_case_insensitive_bloom_",
1533            FulltextBackend::Bloom,
1534            &[
1535                (None, Some("hello"), None),
1536                (None, None, None),
1537                (None, Some("world"), None),
1538                (None, Some("Hello, World"), None),
1539            ],
1540        )
1541        .await;
1542
1543        let row_ids = applier_factory(
1544            vec![],
1545            vec![(2, vec![(false, "hello"), (false, "world")])],
1546            Some(BitVec::from_slice(&[0b1111])),
1547        )
1548        .await;
1549        assert_eq!(row_ids, Some(rows([3])));
1550
1551        let row_ids = applier_factory(
1552            vec![],
1553            vec![(2, vec![(true, "hello"), (false, "world")])],
1554            Some(BitVec::from_slice(&[0b1111])),
1555        )
1556        .await;
1557        assert_eq!(row_ids, Some(rows([3])));
1558
1559        let row_ids = applier_factory(
1560            vec![],
1561            vec![(2, vec![(false, "hello"), (true, "world")])],
1562            Some(BitVec::from_slice(&[0b1111])),
1563        )
1564        .await;
1565        assert_eq!(row_ids, Some(rows([3])));
1566
1567        let row_ids = applier_factory(
1568            vec![],
1569            vec![(2, vec![(true, "hello"), (true, "world")])],
1570            Some(BitVec::from_slice(&[0b1111])),
1571        )
1572        .await;
1573        assert_eq!(row_ids, Some(rows([3])));
1574    }
1575
1576    #[tokio::test]
1577    async fn test_fulltext_index_multi_columns_tantivy() {
1578        let applier_factory = build_fulltext_applier_factory(
1579            "test_fulltext_index_multi_columns_tantivy_",
1580            FulltextBackend::Tantivy,
1581            &[
1582                (Some("Hello"), None, Some("你好")),
1583                (Some("World"), Some("world"), None),
1584                (None, Some("World"), Some("世界")),
1585                (
1586                    Some("Hello, World"),
1587                    Some("Hello, World"),
1588                    Some("你好，世界"),
1589                ),
1590            ],
1591        )
1592        .await;
1593
1594        let row_ids = applier_factory(vec![(1, "Hello"), (3, "你好")], vec![], None).await;
1595        assert_eq!(row_ids, Some(rows([0, 3])));
1596
1597        let row_ids = applier_factory(vec![(1, "World"), (2, "World")], vec![], None).await;
1598        assert_eq!(row_ids, Some(rows([1, 3])));
1599    }
1600
1601    #[tokio::test]
1602    async fn test_fulltext_index_multi_columns_bloom() {
1603        let applier_factory = build_fulltext_applier_factory(
1604            "test_fulltext_index_multi_columns_bloom_",
1605            FulltextBackend::Bloom,
1606            &[
1607                (Some("Hello"), None, Some("你好")),
1608                (Some("World"), Some("world"), None),
1609                (None, Some("World"), Some("世界")),
1610                (
1611                    Some("Hello, World"),
1612                    Some("Hello, World"),
1613                    Some("你好，世界"),
1614                ),
1615            ],
1616        )
1617        .await;
1618
1619        let row_ids = applier_factory(
1620            vec![],
1621            vec![
1622                (1, vec![(false, "Hello")]),
1623                (2, vec![(false, "world")]),
1624                (3, vec![(false, "你好")]),
1625            ],
1626            Some(BitVec::from_slice(&[0b1111])),
1627        )
1628        .await;
1629        assert_eq!(row_ids, Some(rows([3])));
1630
1631        let row_ids = applier_factory(
1632            vec![],
1633            vec![(1, vec![(false, "World")]), (2, vec![(false, "World")])],
1634            Some(BitVec::from_slice(&[0b1111])),
1635        )
1636        .await;
1637        assert_eq!(row_ids, Some(rows([1, 3])));
1638    }
1639}
mito2/sst/index/fulltext_index/creator.rs

mito2/sst/index/fulltext_index/
creator.rs
