feat(parquet): introduce inverted index applier to reader

src/index/src/inverted_index/search/index_apply.rs

@@ -28,7 +28,7 @@ use crate::inverted_index::format::reader::InvertedIndexReader;
 /// avoiding repeated compilation of fixed predicates such as regex patterns.
 #[mockall::automock]
 #[async_trait]
-pub trait IndexApplier {
+pub trait IndexApplier: Send + Sync {
     /// Applies the predefined predicates to the data read by the given index reader, returning
     /// a list of relevant indices (e.g., post IDs, group IDs, row IDs).
     async fn apply<'a>(

src/mito2/src/access_layer.rs

@@ -22,9 +22,9 @@ use store_api::metadata::RegionMetadataRef;
 
 use crate::cache::write_cache::SstUploadRequest;
 use crate::cache::CacheManagerRef;
-use crate::error::{CleanDirSnafu, DeleteSstSnafu, OpenDalSnafu, Result};
+use crate::error::{CleanDirSnafu, DeleteIndexSnafu, DeleteSstSnafu, OpenDalSnafu, Result};
 use crate::read::Source;
-use crate::sst::file::{FileHandle, FileId};
+use crate::sst::file::{FileHandle, FileId, FileMeta};
 use crate::sst::location;
 use crate::sst::parquet::reader::ParquetReaderBuilder;
 use crate::sst::parquet::writer::ParquetWriter;
@@ -67,12 +67,26 @@ impl AccessLayer {
     }
 
     /// Deletes a SST file with given file id.
-    pub(crate) async fn delete_sst(&self, file_id: FileId) -> Result<()> {
-        let path = location::sst_file_path(&self.region_dir, file_id);
+    pub(crate) async fn delete_sst(&self, file_meta: &FileMeta) -> Result<()> {
+        let path = location::sst_file_path(&self.region_dir, file_meta.file_id);
         self.object_store
             .delete(&path)
             .await
-            .context(DeleteSstSnafu { file_id })
+            .context(DeleteSstSnafu {
+                file_id: file_meta.file_id,
+            })?;
+
+        if file_meta.inverted_index_available {
+            let path = location::index_file_path(&self.region_dir, file_meta.file_id);
+            self.object_store
+                .delete(&path)
+                .await
+                .context(DeleteIndexSnafu {
+                    file_id: file_meta.file_id,
+                })?;
+        }
+
+        Ok(())
     }
 
     /// Returns a reader builder for specific `file`.

src/mito2/src/compaction/test_util.rs

@@ -35,6 +35,7 @@ pub fn new_file_handle(
             ),
             level,
             file_size: 0,
+            inverted_index_available: false,
         },
         file_purger,
     )

src/mito2/src/compaction/twcs.rs

@@ -330,6 +330,7 @@ impl TwcsCompactionTask {
                         time_range: sst_info.time_range,
                         level: output.output_level,
                         file_size: sst_info.file_size,
+                        inverted_index_available: sst_info.inverted_index_available,
                     });
                 Ok(file_meta_opt)
             });

src/mito2/src/engine/basic_test.rs

@@ -553,5 +553,5 @@ async fn test_region_usage() {
     assert_eq!(region_stat.sst_usage, 2742);
 
     // region total usage
-    assert_eq!(region_stat.disk_usage(), 3748);
+    assert_eq!(region_stat.disk_usage(), 3781);
 }

src/mito2/src/error.rs

@@ -321,6 +321,14 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to delete index file, file id: {}", file_id))]
+    DeleteIndex {
+        file_id: FileId,
+        #[snafu(source)]
+        error: object_store::Error,
+        location: Location,
+    },
+
     #[snafu(display("Failed to flush region {}", region_id))]
     FlushRegion {
         region_id: RegionId,
@@ -583,7 +591,7 @@ impl ErrorExt for Error {
             InvalidSender { .. } => StatusCode::InvalidArguments,
             InvalidSchedulerState { .. } => StatusCode::InvalidArguments,
             StopScheduler { .. } => StatusCode::Internal,
-            DeleteSst { .. } => StatusCode::StorageUnavailable,
+            DeleteSst { .. } | DeleteIndex { .. } => StatusCode::StorageUnavailable,
             FlushRegion { source, .. } => source.status_code(),
             RegionDropped { .. } => StatusCode::Cancelled,
             RegionClosed { .. } => StatusCode::Cancelled,

src/mito2/src/flush.rs

@@ -339,6 +339,7 @@ impl RegionFlushTask {
                 time_range: sst_info.time_range,
                 level: 0,
                 file_size: sst_info.file_size,
+                inverted_index_available: sst_info.inverted_index_available,
             };
             file_metas.push(file_meta);
         }

src/mito2/src/manifest/tests/checkpoint.rs

@@ -171,6 +171,7 @@ async fn checkpoint_with_different_compression_types() {
             time_range: (0.into(), 10000000.into()),
             level: 0,
             file_size: 1024000,
+            inverted_index_available: false,
         };
         let action = RegionMetaActionList::new(vec![RegionMetaAction::Edit(RegionEdit {
             files_to_add: vec![file_meta],

src/mito2/src/metrics.rs

@@ -121,6 +121,9 @@ lazy_static! {
     /// Counter of filtered rows during merge.
     pub static ref MERGE_FILTER_ROWS_TOTAL: IntCounterVec =
         register_int_counter_vec!("greptime_mito_merge_filter_rows_total", "mito merge filter rows total", &[TYPE_LABEL]).unwrap();
+    /// Counter of row groups read.
+    pub static ref READ_ROW_GROUPS_TOTAL: IntCounterVec =
+        register_int_counter_vec!("greptime_mito_read_row_groups_total", "mito read row groups total", &[TYPE_LABEL]).unwrap();
     // ------- End of query metrics.
 
     // Cache related metrics.

src/mito2/src/read/scan_region.rs

@@ -14,8 +14,10 @@
 
 //! Scans a region according to the scan request.
 
+use std::sync::Arc;
+
 use common_recordbatch::SendableRecordBatchStream;
-use common_telemetry::debug;
+use common_telemetry::{debug, warn};
 use common_time::range::TimestampRange;
 use store_api::storage::ScanRequest;
 use table::predicate::{Predicate, TimeRangePredicateBuilder};
@@ -27,6 +29,8 @@ use crate::read::projection::ProjectionMapper;
 use crate::read::seq_scan::SeqScan;
 use crate::region::version::VersionRef;
 use crate::sst::file::FileHandle;
+use crate::sst::index::applier::builder::SstIndexApplierBuilder;
+use crate::sst::index::applier::SstIndexApplierRef;
 
 /// A scanner scans a region and returns a [SendableRecordBatchStream].
 pub(crate) enum Scanner {
@@ -194,6 +198,7 @@ impl ScanRegion {
             total_ssts
         );
 
+        let index_applier = self.build_index_applier();
         let predicate = Predicate::new(self.request.filters.clone());
         // The mapper always computes projected column ids as the schema of SSTs may change.
         let mapper = match &self.request.projection {
@@ -207,6 +212,7 @@ impl ScanRegion {
             .with_memtables(memtables)
             .with_files(files)
             .with_cache(self.cache_manager)
+            .with_index_applier(index_applier)
             .with_parallelism(self.parallelism);
 
         Ok(seq_scan)
@@ -224,6 +230,20 @@ impl ScanRegion {
         TimeRangePredicateBuilder::new(&time_index.column_schema.name, unit, &self.request.filters)
             .build()
     }
+
+    /// Use the latest schema to build the index applier.
+    fn build_index_applier(&self) -> Option<SstIndexApplierRef> {
+        SstIndexApplierBuilder::new(
+            self.access_layer.region_dir().to_string(),
+            self.access_layer.object_store().clone(),
+            self.version.metadata.as_ref(),
+        )
+        .build(&self.request.filters)
+        .inspect_err(|err| warn!(err; "Failed to build index applier"))
+        .ok()
+        .flatten()
+        .map(Arc::new)
+    }
 }
 
 /// Config for parallel scan.

src/mito2/src/read/seq_scan.rs

@@ -39,6 +39,7 @@ use crate::read::projection::ProjectionMapper;
 use crate::read::scan_region::ScanParallism;
 use crate::read::{BatchReader, BoxedBatchReader, BoxedBatchStream, Source};
 use crate::sst::file::FileHandle;
+use crate::sst::index::applier::SstIndexApplierRef;
 
 /// Scans a region and returns rows in a sorted sequence.
 ///
@@ -62,6 +63,8 @@ pub struct SeqScan {
     ignore_file_not_found: bool,
     /// Parallelism to scan data.
     parallelism: ScanParallism,
+    /// Index applier.
+    index_applier: Option<SstIndexApplierRef>,
 }
 
 impl SeqScan {
@@ -78,6 +81,7 @@ impl SeqScan {
             cache_manager: None,
             ignore_file_not_found: false,
             parallelism: ScanParallism::default(),
+            index_applier: None,
         }
     }
 
@@ -130,6 +134,13 @@ impl SeqScan {
         self
     }
 
+    /// Sets index applier.
+    #[must_use]
+    pub(crate) fn with_index_applier(mut self, index_applier: Option<SstIndexApplierRef>) -> Self {
+        self.index_applier = index_applier;
+        self
+    }
+
     /// Builds a stream for the query.
     pub async fn build_stream(&self) -> Result<SendableRecordBatchStream> {
         let start = Instant::now();

src/mito2/src/request.rs

@@ -651,8 +651,7 @@ impl OnFailure for FlushFinished {
         // Clean flushed files.
         for file in &self.file_metas {
             self.file_purger.send_request(PurgeRequest {
-                region_id: file.region_id,
-                file_id: file.file_id,
+                file_meta: file.clone(),
             });
         }
     }
@@ -707,14 +706,12 @@ impl OnFailure for CompactionFinished {
             }));
         }
         for file in &self.compacted_files {
-            let file_id = file.file_id;
             warn!(
                 "Cleaning region {} compaction output file: {}",
-                self.region_id, file_id
+                self.region_id, file.file_id
             );
             self.file_purger.send_request(PurgeRequest {
-                region_id: self.region_id,
-                file_id,
+                file_meta: file.clone(),
             });
         }
     }

src/mito2/src/sst/file.rs

@@ -95,6 +95,8 @@ pub struct FileMeta {
     pub level: Level,
     /// Size of the file.
     pub file_size: u64,
+    /// Whether inverted index is available.
+    pub inverted_index_available: bool,
 }
 
 /// Handle to a SST file.
@@ -176,8 +178,7 @@ impl Drop for FileHandleInner {
     fn drop(&mut self) {
         if self.deleted.load(Ordering::Relaxed) {
             self.file_purger.send_request(PurgeRequest {
-                region_id: self.meta.region_id,
-                file_id: self.meta.file_id,
+                file_meta: self.meta.clone(),
             });
         }
     }
@@ -236,6 +237,7 @@ mod tests {
             time_range: FileTimeRange::default(),
             level,
             file_size: 0,
+            inverted_index_available: false,
         }
     }
 

src/mito2/src/sst/file_purger.rs

@@ -16,20 +16,17 @@ use std::fmt;
 use std::sync::Arc;
 
 use common_telemetry::{error, info};
-use store_api::storage::RegionId;
 
 use crate::access_layer::AccessLayerRef;
 use crate::cache::CacheManagerRef;
 use crate::schedule::scheduler::SchedulerRef;
-use crate::sst::file::FileId;
+use crate::sst::file::FileMeta;
 
 /// Request to remove a file.
 #[derive(Debug)]
 pub struct PurgeRequest {
-    /// Region id of the file.
-    pub region_id: RegionId,
-    /// Id of the file.
-    pub file_id: FileId,
+    /// File meta.
+    pub file_meta: FileMeta,
 }
 
 /// A worker to delete files in background.
@@ -72,24 +69,22 @@ impl LocalFilePurger {
 
 impl FilePurger for LocalFilePurger {
     fn send_request(&self, request: PurgeRequest) {
-        let file_id = request.file_id;
-        let region_id = request.region_id;
+        let file_meta = request.file_meta;
         let sst_layer = self.sst_layer.clone();
 
         // Remove meta of the file from cache.
         if let Some(cache) = &self.cache_manager {
-            cache.remove_parquet_meta_data(region_id, file_id);
+            cache.remove_parquet_meta_data(file_meta.region_id, file_meta.file_id);
         }
 
         if let Err(e) = self.scheduler.schedule(Box::pin(async move {
-            if let Err(e) = sst_layer.delete_sst(file_id).await {
-                error!(e; "Failed to delete SST file, file: {}, region: {}", 
-                    file_id.as_parquet(), region_id);
+            if let Err(e) = sst_layer.delete_sst(&file_meta).await {
+                error!(e; "Failed to delete SST file, file_id: {}, region: {}", 
+                    file_meta.file_id, file_meta.region_id);
             } else {
                 info!(
-                    "Successfully deleted SST file: {}, region: {}",
-                    file_id.as_parquet(),
-                    region_id
+                    "Successfully deleted SST file, file_id: {}, region: {}",
+                    file_meta.file_id, file_meta.region_id
                 );
             }
         })) {
@@ -137,6 +132,7 @@ mod tests {
                     time_range: FileTimeRange::default(),
                     level: 0,
                     file_size: 4096,
+                    inverted_index_available: false,
                 },
                 file_purger,
             );
@@ -148,4 +144,51 @@ mod tests {
 
         assert!(!object_store.is_exist(&path).await.unwrap());
     }
+
+    #[tokio::test]
+    async fn test_file_purge_with_index() {
+        common_telemetry::init_default_ut_logging();
+
+        let dir = create_temp_dir("file-purge");
+        let mut builder = Fs::default();
+        builder.root(dir.path().to_str().unwrap());
+        let object_store = ObjectStore::new(builder).unwrap().finish();
+        let sst_file_id = FileId::random();
+        let sst_dir = "table1";
+
+        let path = location::sst_file_path(sst_dir, sst_file_id);
+        object_store.write(&path, vec![0; 4096]).await.unwrap();
+
+        let index_path = location::index_file_path(sst_dir, sst_file_id);
+        object_store
+            .write(&index_path, vec![0; 4096])
+            .await
+            .unwrap();
+
+        let scheduler = Arc::new(LocalScheduler::new(3));
+        let layer = Arc::new(AccessLayer::new(sst_dir, object_store.clone()));
+
+        let file_purger = Arc::new(LocalFilePurger::new(scheduler.clone(), layer, None));
+
+        {
+            let handle = FileHandle::new(
+                FileMeta {
+                    region_id: 0.into(),
+                    file_id: sst_file_id,
+                    time_range: FileTimeRange::default(),
+                    level: 0,
+                    file_size: 4096,
+                    inverted_index_available: true,
+                },
+                file_purger,
+            );
+            // mark file as deleted and drop the handle, we expect the sst file and the index file are deleted.
+            handle.mark_deleted();
+        }
+
+        scheduler.stop(true).await.unwrap();
+
+        assert!(!object_store.is_exist(&path).await.unwrap());
+        assert!(!object_store.is_exist(&index_path).await.unwrap());
+    }
 }