chore: purge inverted index (#15354)

* purge inverted index

* purge inverted index info files

* add code comments

* add ee sql logic test

* revert default query node config

* chore: cleanup

* resolve merge conflicts

* evict inverted index cache

Author: dantengsky

src/query/storages/common/cache_manager/src/caches.rs

@@ -155,6 +155,20 @@ impl CachedObject<InvertedIndexFile, DefaultHashBuilder, InvertedIndexFileMeter>
     }
 }
 
+impl CachedObject<IndexInfo> for IndexInfo {
+    type Cache = InvertedIndexInfoCache;
+    fn cache() -> Option<Self::Cache> {
+        CacheManager::instance().get_inverted_index_info_cache()
+    }
+}
+
+impl CachedObject<InvertedIndexMeta> for InvertedIndexMeta {
+    type Cache = InvertedIndexMetaCache;
+    fn cache() -> Option<Self::Cache> {
+        CacheManager::instance().get_inverted_index_meta_cache()
+    }
+}
+
 pub struct ColumnArrayMeter;
 
 impl<K, V> Meter<K, Arc<(V, usize)>> for ColumnArrayMeter {

src/query/storages/fuse/src/io/locations.rs

@@ -169,14 +169,14 @@ impl TableMetaLocationGenerator {
     }
 
     pub fn gen_inverted_index_info_location(&self) -> String {
+        let prefix = self.inverted_index_info_prefix();
         let uuid = Uuid::new_v4().simple().to_string();
-        format!(
-            "{}/{}/{}_v{}.mpk",
-            &self.prefix,
-            FUSE_TBL_INVERTED_INDEX_INFO_PREFIX,
-            uuid,
-            IndexInfo::VERSION,
-        )
+        format!("{}{}_v{}.mpk", prefix, uuid, IndexInfo::VERSION,)
+    }
+
+    // inverted index info path, trailing slash "/" included.
+    pub fn inverted_index_info_prefix(&self) -> String {
+        format!("{}/{}/", &self.prefix, FUSE_TBL_INVERTED_INDEX_INFO_PREFIX)
     }
 
     pub fn gen_inverted_index_location_from_block_location(

src/query/storages/fuse/src/io/mod.rs

@@ -25,6 +25,7 @@ pub use read::AggIndexReader;
 pub use read::BlockReader;
 pub use read::BloomBlockFilterReader;
 pub use read::CompactSegmentInfoReader;
+pub use read::InvertedIndexReader;
 pub use read::MergeIOReadResult;
 pub use read::MetaReaders;
 pub use read::NativeReaderExt;

src/query/storages/fuse/src/io/read/inverted_index/inverted_index_loader.rs

@@ -41,6 +41,17 @@ use crate::io::MetaReaders;
 type CachedReader =
     InMemoryCacheReader<InvertedIndexFile, InvertedIndexFileLoader, InvertedIndexFileMeter>;
 
+const INDEX_COLUMN_NAMES: [&str; 8] = [
+    "fast",
+    "store",
+    "fieldnorm",
+    "pos",
+    "idx",
+    "term",
+    "meta.json",
+    ".managed.json",
+];
+
 /// Loads inverted index info data
 /// read data from cache, or populate cache items if possible
 #[minitrace::trace]
@@ -188,7 +199,7 @@ impl InvertedIndexFileReader {
         column_meta: &SingleColumnMeta,
         operator: Operator,
     ) -> Self {
-        let cache_key = format!("{index_path}-{name}");
+        let cache_key = Self::cache_key_of_column(&index_path, &name);
 
         let loader = InvertedIndexFileLoader {
             offset: column_meta.offset,
@@ -217,6 +228,17 @@ impl InvertedIndexFileReader {
     pub async fn read(&self) -> Result<Arc<InvertedIndexFile>> {
         self.cached_reader.read(&self.param).await
     }
+
+    fn cache_key_of_column(index_path: &str, index_column_name: &str) -> String {
+        format!("{index_path}-{index_column_name}")
+    }
+
+    pub(crate) fn cache_key_of_index_columns(index_path: &str) -> Vec<String> {
+        INDEX_COLUMN_NAMES
+            .iter()
+            .map(|column_name| Self::cache_key_of_column(index_path, column_name))
+            .collect()
+    }
 }
 
 /// Loader that fetch range of the target object with customized cache key

src/query/storages/fuse/src/io/read/inverted_index/inverted_index_reader.rs

@@ -25,6 +25,7 @@ use tantivy::Index;
 use tantivy::Score;
 
 use crate::io::read::inverted_index::inverted_index_loader::load_inverted_index_directory;
+use crate::io::read::inverted_index::inverted_index_loader::InvertedIndexFileReader;
 
 #[derive(Clone)]
 pub struct InvertedIndexReader {
@@ -107,4 +108,9 @@ impl InvertedIndexReader {
 
         Ok(Some(matched_rows))
     }
+
+    // delegation of [InvertedIndexFileReader::cache_key_of_index_columns]
+    pub fn cache_key_of_index_columns(index_path: &str) -> Vec<String> {
+        InvertedIndexFileReader::cache_key_of_index_columns(index_path)
+    }
 }

src/query/storages/fuse/src/io/snapshots.rs

@@ -375,7 +375,7 @@ impl SnapshotsIO {
                     }
                 },
                 _ => {
-                    warn!("found not snapshot file in {:}, found: {:?}", prefix, de);
+                    warn!("non-file entry found in {:}, the entry: {:?}", prefix, de);
                     continue;
                 }
             }

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -381,8 +381,8 @@ where F: SnapshotGenerator + Send + 'static
                             let tbl = FuseTable::try_from_table(latest.as_ref())?;
 
                             warn!(
-                                "table detected, purging historical data. ({})",
-                                tbl.table_info.ident
+                                "purging historical data. table: {}, ident: {}",
+                                tbl.table_info.name, tbl.table_info.ident
                             );
 
                             let keep_last_snapshot = true;