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

mito2/engine/
puffin_index.rs
