mito2/engine/
puffin_index.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::convert::TryFrom;
16use std::sync::Arc;
17
18use common_base::range_read::RangeReader;
19use common_telemetry::warn;
20use greptime_proto::v1::index::{BloomFilterMeta, InvertedIndexMeta, InvertedIndexMetas};
21use index::bitmap::BitmapType;
22use index::bloom_filter::reader::{BloomFilterReader, BloomFilterReaderImpl};
23use index::fulltext_index::Config as FulltextConfig;
24use index::inverted_index::format::reader::{InvertedIndexBlobReader, InvertedIndexReader};
25use index::target::IndexTarget;
26use puffin::blob_metadata::BlobMetadata;
27use puffin::puffin_manager::{PuffinManager, PuffinReader};
28use serde_json::{Map, Value, json};
29use store_api::sst_entry::{
30    PUFFIN_INDEX_TYPE_BLOOM_FILTER, PUFFIN_INDEX_TYPE_FULLTEXT_BLOOM,
31    PUFFIN_INDEX_TYPE_FULLTEXT_TANTIVY, PUFFIN_INDEX_TYPE_INVERTED, PuffinIndexMetaEntry,
32};
33use store_api::storage::{ColumnId, RegionGroup, RegionId, RegionNumber, RegionSeq, TableId};
34
35use crate::cache::index::bloom_filter_index::{
36    BloomFilterIndexCacheRef, CachedBloomFilterIndexBlobReader, Tag,
37};
38use crate::cache::index::inverted_index::{CachedInvertedIndexBlobReader, InvertedIndexCacheRef};
39use crate::sst::file::RegionIndexId;
40use crate::sst::index::bloom_filter::INDEX_BLOB_TYPE as BLOOM_BLOB_TYPE;
41use crate::sst::index::fulltext_index::{
42    FULLTEXT_BLOOM_VERSION, INDEX_BLOB_TYPE_BLOOM as FULLTEXT_BLOOM_BLOB_TYPE,
43    INDEX_BLOB_TYPE_BLOOM_V1 as FULLTEXT_BLOOM_V1_BLOB_TYPE,
44    INDEX_BLOB_TYPE_TANTIVY as FULLTEXT_TANTIVY_BLOB_TYPE,
45};
46use crate::sst::index::inverted_index::INDEX_BLOB_TYPE as INVERTED_BLOB_TYPE;
47use crate::sst::index::puffin_manager::{SstPuffinManager, SstPuffinReader};
48
49const TARGET_TYPE_UNKNOWN: &str = "unknown";
50
51const TARGET_TYPE_COLUMN: &str = "column";
52
53const TARGET_TYPE_JSON_PATH: &str = "json_path";
54
55pub(crate) struct IndexEntryContext<'a> {
56    pub(crate) table_dir: &'a str,
57    pub(crate) index_file_path: &'a str,
58    pub(crate) region_id: RegionId,
59    pub(crate) table_id: TableId,
60    pub(crate) region_number: RegionNumber,
61    pub(crate) region_group: RegionGroup,
62    pub(crate) region_sequence: RegionSeq,
63    pub(crate) file_id: &'a str,
64    pub(crate) index_file_size: Option<u64>,
65    pub(crate) node_id: Option<u64>,
66}
67
68/// Collect index metadata entries present in the SST puffin file.
69pub(crate) async fn collect_index_entries_from_puffin(
70    manager: SstPuffinManager,
71    region_index_id: RegionIndexId,
72    context: IndexEntryContext<'_>,
73    bloom_filter_cache: Option<BloomFilterIndexCacheRef>,
74    inverted_index_cache: Option<InvertedIndexCacheRef>,
75) -> Vec<PuffinIndexMetaEntry> {
76    let mut entries = Vec::new();
77
78    let reader = match manager.reader(&region_index_id).await {
79        Ok(reader) => reader,
80        Err(err) => {
81            warn!(
82                err;
83                "Failed to open puffin index file, table_dir: {}, file_id: {}",
84                context.table_dir,
85                context.file_id
86            );
87            return entries;
88        }
89    };
90
91    let file_metadata = match reader.metadata().await {
92        Ok(metadata) => metadata,
93        Err(err) => {
94            warn!(
95                err;
96                "Failed to read puffin file metadata, table_dir: {}, file_id: {}",
97                context.table_dir,
98                context.file_id
99            );
100            return entries;
101        }
102    };
103
104    for blob in &file_metadata.blobs {
105        match BlobIndexTypeTargetKey::from_blob_type(&blob.blob_type) {
106            Some(BlobIndexTypeTargetKey::BloomFilter(target_key)) => {
107                let bloom_meta = try_read_bloom_meta(
108                    &reader,
109                    region_index_id,
110                    blob.blob_type.as_str(),
111                    target_key,
112                    bloom_filter_cache.as_ref(),
113                    Tag::Skipping,
114                    &context,
115                )
116                .await;
117
118                let bloom_value = bloom_meta.as_deref().map(bloom_meta_value);
119                let (target_type, target_json) = decode_target_info(target_key);
120                let meta_json = build_meta_json(bloom_value, None, None);
121                let entry = build_index_entry(
122                    &context,
123                    PUFFIN_INDEX_TYPE_BLOOM_FILTER,
124                    target_type,
125                    target_key.to_string(),
126                    target_json,
127                    blob.length as u64,
128                    meta_json,
129                );
130                entries.push(entry);
131            }
132            Some(BlobIndexTypeTargetKey::FulltextBloom(target_key, bloom_version)) => {
133                let bloom_meta = try_read_bloom_meta(
134                    &reader,
135                    region_index_id,
136                    blob.blob_type.as_str(),
137                    target_key,
138                    bloom_filter_cache.as_ref(),
139                    Tag::Fulltext,
140                    &context,
141                )
142                .await;
143
144                let bloom_value = bloom_meta.as_deref().map(bloom_meta_value);
145                let fulltext_value = Some(fulltext_meta_value(blob, Some(bloom_version)));
146                let (target_type, target_json) = decode_target_info(target_key);
147                let meta_json = build_meta_json(bloom_value, fulltext_value, None);
148                let entry = build_index_entry(
149                    &context,
150                    PUFFIN_INDEX_TYPE_FULLTEXT_BLOOM,
151                    target_type,
152                    target_key.to_string(),
153                    target_json,
154                    blob.length as u64,
155                    meta_json,
156                );
157                entries.push(entry);
158            }
159            Some(BlobIndexTypeTargetKey::FulltextTantivy(target_key)) => {
160                let fulltext_value = Some(fulltext_meta_value(blob, None));
161                let (target_type, target_json) = decode_target_info(target_key);
162                let meta_json = build_meta_json(None, fulltext_value, None);
163                let entry = build_index_entry(
164                    &context,
165                    PUFFIN_INDEX_TYPE_FULLTEXT_TANTIVY,
166                    target_type,
167                    target_key.to_string(),
168                    target_json,
169                    blob.length as u64,
170                    meta_json,
171                );
172                entries.push(entry);
173            }
174            Some(BlobIndexTypeTargetKey::Inverted) => {
175                let mut inverted_entries = collect_inverted_entries(
176                    &reader,
177                    region_index_id,
178                    inverted_index_cache.as_ref(),
179                    &context,
180                )
181                .await;
182                entries.append(&mut inverted_entries);
183            }
184            None => {}
185        }
186    }
187
188    entries
189}
190
191async fn collect_inverted_entries(
192    reader: &SstPuffinReader,
193    region_index_id: RegionIndexId,
194    cache: Option<&InvertedIndexCacheRef>,
195    context: &IndexEntryContext<'_>,
196) -> Vec<PuffinIndexMetaEntry> {
197    // Read the inverted index blob and surface its per-column metadata entries.
198    let file_id = region_index_id.file_id();
199
200    let guard = match reader.blob(INVERTED_BLOB_TYPE).await {
201        Ok(guard) => guard,
202        Err(err) => {
203            warn!(
204                err;
205                "Failed to open inverted index blob, table_dir: {}, file_id: {}",
206                context.table_dir,
207                context.file_id
208            );
209            return Vec::new();
210        }
211    };
212
213    let blob_reader = match guard.reader().await {
214        Ok(reader) => reader,
215        Err(err) => {
216            warn!(
217                err;
218                "Failed to build inverted index blob reader, table_dir: {}, file_id: {}",
219                context.table_dir,
220                context.file_id
221            );
222            return Vec::new();
223        }
224    };
225
226    let blob_size = blob_reader
227        .metadata()
228        .await
229        .ok()
230        .map(|meta| meta.content_length);
231    let metas = if let (Some(cache), Some(blob_size)) = (cache, blob_size) {
232        let reader = CachedInvertedIndexBlobReader::new(
233            file_id,
234            region_index_id.version,
235            blob_size,
236            InvertedIndexBlobReader::new(blob_reader),
237            cache.clone(),
238        );
239        match reader.metadata(None).await {
240            Ok(metas) => metas,
241            Err(err) => {
242                warn!(
243                    err;
244                    "Failed to read inverted index metadata, table_dir: {}, file_id: {}",
245                    context.table_dir,
246                    context.file_id
247                );
248                return Vec::new();
249            }
250        }
251    } else {
252        let reader = InvertedIndexBlobReader::new(blob_reader);
253        match reader.metadata(None).await {
254            Ok(metas) => metas,
255            Err(err) => {
256                warn!(
257                    err;
258                    "Failed to read inverted index metadata, table_dir: {}, file_id: {}",
259                    context.table_dir,
260                    context.file_id
261                );
262                return Vec::new();
263            }
264        }
265    };
266
267    build_inverted_entries(context, metas.as_ref())
268}
269
270fn build_inverted_entries(
271    context: &IndexEntryContext<'_>,
272    metas: &InvertedIndexMetas,
273) -> Vec<PuffinIndexMetaEntry> {
274    let mut entries = Vec::new();
275    for (name, meta) in &metas.metas {
276        let (target_type, target_json) = decode_target_info(name);
277        let inverted_value = inverted_meta_value(meta, metas);
278        let meta_json = build_meta_json(None, None, Some(inverted_value));
279        let entry = build_index_entry(
280            context,
281            PUFFIN_INDEX_TYPE_INVERTED,
282            target_type,
283            name.clone(),
284            target_json,
285            meta.inverted_index_size,
286            meta_json,
287        );
288        entries.push(entry);
289    }
290    entries
291}
292
293async fn try_read_bloom_meta(
294    reader: &SstPuffinReader,
295    region_index_id: RegionIndexId,
296    blob_type: &str,
297    target_key: &str,
298    cache: Option<&BloomFilterIndexCacheRef>,
299    tag: Tag,
300    context: &IndexEntryContext<'_>,
301) -> Option<Arc<BloomFilterMeta>> {
302    let column_id = decode_column_id(target_key);
303
304    // Failures are logged but do not abort the overall metadata collection.
305    match reader.blob(blob_type).await {
306        Ok(guard) => match guard.reader().await {
307            Ok(blob_reader) => {
308                let blob_size = blob_reader
309                    .metadata()
310                    .await
311                    .ok()
312                    .map(|meta| meta.content_length);
313                let bloom_reader = BloomFilterReaderImpl::new(blob_reader);
314                let result = match (cache, column_id, blob_size) {
315                    (Some(cache), Some(column_id), Some(blob_size)) => {
316                        CachedBloomFilterIndexBlobReader::new(
317                            region_index_id.file_id(),
318                            region_index_id.version,
319                            IndexTarget::ColumnId(column_id),
320                            tag,
321                            blob_size,
322                            bloom_reader,
323                            cache.clone(),
324                        )
325                        .metadata(None)
326                        .await
327                    }
328                    _ => bloom_reader.metadata(None).await,
329                };
330
331                match result {
332                    Ok(meta) => Some(meta),
333                    Err(err) => {
334                        warn!(
335                            err;
336                            "Failed to read index metadata, table_dir: {}, file_id: {}, blob: {}",
337                            context.table_dir,
338                            context.file_id,
339                            blob_type
340                        );
341                        None
342                    }
343                }
344            }
345            Err(err) => {
346                warn!(
347                    err;
348                    "Failed to open index blob reader, table_dir: {}, file_id: {}, blob: {}",
349                    context.table_dir,
350                    context.file_id,
351                    blob_type
352                );
353                None
354            }
355        },
356        Err(err) => {
357            warn!(
358                err;
359                "Failed to open index blob, table_dir: {}, file_id: {}, blob: {}",
360                context.table_dir,
361                context.file_id,
362                blob_type
363            );
364            None
365        }
366    }
367}
368
369fn decode_target_info(target_key: &str) -> (String, String) {
370    match IndexTarget::decode(target_key) {
371        Ok(IndexTarget::ColumnId(id)) => (
372            TARGET_TYPE_COLUMN.to_string(),
373            json!({ "column": id }).to_string(),
374        ),
375        Ok(IndexTarget::JsonPath(target)) => (
376            TARGET_TYPE_JSON_PATH.to_string(),
377            json!({ "json_path": { "column": target.column_id(), "path": target.path() } })
378                .to_string(),
379        ),
380        _ => (
381            TARGET_TYPE_UNKNOWN.to_string(),
382            json!({ "error": "failed_to_decode" }).to_string(),
383        ),
384    }
385}
386
387fn decode_column_id(target_key: &str) -> Option<ColumnId> {
388    match IndexTarget::decode(target_key) {
389        Ok(IndexTarget::ColumnId(id)) => Some(id),
390        _ => None,
391    }
392}
393
394fn bloom_meta_value(meta: &BloomFilterMeta) -> Value {
395    json!({
396        "rows_per_segment": meta.rows_per_segment,
397        "segment_count": meta.segment_count,
398        "row_count": meta.row_count,
399        "bloom_filter_size": meta.bloom_filter_size,
400    })
401}
402
403fn fulltext_meta_value(blob: &BlobMetadata, bloom_version: Option<u8>) -> Value {
404    let config = FulltextConfig::from_blob_metadata(blob).unwrap_or_default();
405    let mut value = json!({
406        "analyzer": config.analyzer.to_str(),
407        "case_sensitive": config.case_sensitive,
408    });
409    if let Some(version) = bloom_version {
410        value["bloom_version"] = json!(version);
411    }
412    value
413}
414
415fn inverted_meta_value(meta: &InvertedIndexMeta, metas: &InvertedIndexMetas) -> Value {
416    let bitmap_type = BitmapType::try_from(meta.bitmap_type)
417        .map(|bt| format!("{:?}", bt))
418        .unwrap_or_else(|_| meta.bitmap_type.to_string());
419    json!({
420        "bitmap_type": bitmap_type,
421        "base_offset": meta.base_offset,
422        "inverted_index_size": meta.inverted_index_size,
423        "relative_fst_offset": meta.relative_fst_offset,
424        "fst_size": meta.fst_size,
425        "relative_null_bitmap_offset": meta.relative_null_bitmap_offset,
426        "null_bitmap_size": meta.null_bitmap_size,
427        "segment_row_count": metas.segment_row_count,
428        "total_row_count": metas.total_row_count,
429    })
430}
431
432fn build_meta_json(
433    bloom: Option<Value>,
434    fulltext: Option<Value>,
435    inverted: Option<Value>,
436) -> Option<String> {
437    let mut map = Map::new();
438    if let Some(value) = bloom {
439        map.insert("bloom".to_string(), value);
440    }
441    if let Some(value) = fulltext {
442        map.insert("fulltext".to_string(), value);
443    }
444    if let Some(value) = inverted {
445        map.insert("inverted".to_string(), value);
446    }
447    if map.is_empty() {
448        None
449    } else {
450        Some(Value::Object(map).to_string())
451    }
452}
453
454enum BlobIndexTypeTargetKey<'a> {
455    BloomFilter(&'a str),
456    /// Target key and version of a fulltext bloom blob.
457    FulltextBloom(&'a str, u8),
458    FulltextTantivy(&'a str),
459    Inverted,
460}
461
462impl<'a> BlobIndexTypeTargetKey<'a> {
463    fn from_blob_type(blob_type: &'a str) -> Option<Self> {
464        if let Some(target_key) = Self::target_key_from_blob(blob_type, BLOOM_BLOB_TYPE) {
465            Some(BlobIndexTypeTargetKey::BloomFilter(target_key))
466        } else if let Some(target_key) =
467            // v2 goes first: the v1 type is a prefix of it.
468            Self::target_key_from_blob(blob_type, FULLTEXT_BLOOM_BLOB_TYPE)
469        {
470            Some(BlobIndexTypeTargetKey::FulltextBloom(
471                target_key,
472                FULLTEXT_BLOOM_VERSION,
473            ))
474        } else if let Some(target_key) =
475            Self::target_key_from_blob(blob_type, FULLTEXT_BLOOM_V1_BLOB_TYPE)
476        {
477            Some(BlobIndexTypeTargetKey::FulltextBloom(target_key, 1))
478        } else if let Some(target_key) =
479            Self::target_key_from_blob(blob_type, FULLTEXT_TANTIVY_BLOB_TYPE)
480        {
481            Some(BlobIndexTypeTargetKey::FulltextTantivy(target_key))
482        } else if blob_type == INVERTED_BLOB_TYPE {
483            Some(BlobIndexTypeTargetKey::Inverted)
484        } else {
485            None
486        }
487    }
488
489    fn target_key_from_blob(blob_type: &'a str, prefix: &str) -> Option<&'a str> {
490        // Blob types encode their target as "<prefix>-<target>".
491        blob_type
492            .strip_prefix(prefix)
493            .and_then(|suffix| suffix.strip_prefix('-'))
494    }
495}
496
497fn build_index_entry(
498    context: &IndexEntryContext<'_>,
499    index_type: &str,
500    target_type: String,
501    target_key: String,
502    target_json: String,
503    blob_size: u64,
504    meta_json: Option<String>,
505) -> PuffinIndexMetaEntry {
506    PuffinIndexMetaEntry {
507        table_dir: context.table_dir.to_string(),
508        index_file_path: context.index_file_path.to_string(),
509        region_id: context.region_id,
510        table_id: context.table_id,
511        region_number: context.region_number,
512        region_group: context.region_group,
513        region_sequence: context.region_sequence,
514        file_id: context.file_id.to_string(),
515        index_file_size: context.index_file_size,
516        index_type: index_type.to_string(),
517        target_type,
518        target_key,
519        target_json,
520        blob_size,
521        meta_json,
522        node_id: context.node_id,
523    }
524}
mito2/engine/puffin_index.rs

mito2/engine/
puffin_index.rs
