mito2/
access_layer.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::sync::Arc;
16use std::time::{Duration, Instant};
17
18use async_stream::try_stream;
19use common_telemetry::warn;
20use common_time::Timestamp;
21use futures::{Stream, TryStreamExt};
22use object_store::services::Fs;
23use object_store::util::{join_dir, with_instrument_layers};
24use object_store::{ATOMIC_WRITE_DIR, ErrorKind, OLD_ATOMIC_WRITE_DIR, ObjectStore};
25use parquet::file::metadata::PageIndexPolicy;
26use smallvec::SmallVec;
27use snafu::ResultExt;
28use store_api::metadata::RegionMetadataRef;
29use store_api::region_request::PathType;
30use store_api::sst_entry::StorageSstEntry;
31use store_api::storage::{FileId, RegionId, SequenceNumber};
32
33use crate::cache::file_cache::{FileCacheRef, FileType, IndexKey};
34use crate::cache::write_cache::SstUploadRequest;
35use crate::cache::{CacheManagerRef, SstMetaPreparation, prepare_sst_meta_sync};
36use crate::config::{BloomFilterConfig, FulltextIndexConfig, IndexConfig, InvertedIndexConfig};
37use crate::error::{
38    CleanDirSnafu, DeleteIndexSnafu, DeleteIndexesSnafu, DeleteSstsSnafu, OpenDalSnafu, Result,
39};
40use crate::metrics::{COMPACTION_STAGE_ELAPSED, FLUSH_ELAPSED};
41use crate::read::FlatSource;
42use crate::region::options::IndexOptions;
43use crate::sst::file::{FileHandle, RegionFileId, RegionIndexId};
44use crate::sst::index::IndexerBuilderImpl;
45use crate::sst::index::intermediate::IntermediateManager;
46use crate::sst::index::puffin_manager::{PuffinManagerFactory, SstPuffinManager};
47use crate::sst::location::{self, region_dir_from_table_dir};
48use crate::sst::parquet::reader::ParquetReaderBuilder;
49use crate::sst::parquet::writer::ParquetWriter;
50use crate::sst::parquet::{SstInfo, WriteOptions};
51use crate::sst::{DEFAULT_WRITE_BUFFER_SIZE, DEFAULT_WRITE_CONCURRENCY, FormatType};
52
53pub type AccessLayerRef = Arc<AccessLayer>;
54/// SST write results.
55pub type SstInfoArray = SmallVec<[SstInfo; 2]>;
56
57/// Write operation type.
58#[derive(Eq, PartialEq, Debug)]
59pub enum WriteType {
60    /// Writes from flush
61    Flush,
62    /// Writes from compaction.
63    Compaction,
64}
65
66#[derive(Debug)]
67pub struct Metrics {
68    pub(crate) write_type: WriteType,
69    pub(crate) iter_source: Duration,
70    pub(crate) write_batch: Duration,
71    pub(crate) update_index: Duration,
72    pub(crate) upload_parquet: Duration,
73    pub(crate) upload_puffin: Duration,
74    pub(crate) compact_memtable: Duration,
75}
76
77impl Metrics {
78    pub fn new(write_type: WriteType) -> Self {
79        Self {
80            write_type,
81            iter_source: Default::default(),
82            write_batch: Default::default(),
83            update_index: Default::default(),
84            upload_parquet: Default::default(),
85            upload_puffin: Default::default(),
86            compact_memtable: Default::default(),
87        }
88    }
89
90    pub(crate) fn merge(mut self, other: Self) -> Self {
91        assert_eq!(self.write_type, other.write_type);
92        self.iter_source += other.iter_source;
93        self.write_batch += other.write_batch;
94        self.update_index += other.update_index;
95        self.upload_parquet += other.upload_parquet;
96        self.upload_puffin += other.upload_puffin;
97        self.compact_memtable += other.compact_memtable;
98        self
99    }
100
101    pub(crate) fn observe(self) {
102        match self.write_type {
103            WriteType::Flush => {
104                FLUSH_ELAPSED
105                    .with_label_values(&["iter_source"])
106                    .observe(self.iter_source.as_secs_f64());
107                FLUSH_ELAPSED
108                    .with_label_values(&["write_batch"])
109                    .observe(self.write_batch.as_secs_f64());
110                FLUSH_ELAPSED
111                    .with_label_values(&["update_index"])
112                    .observe(self.update_index.as_secs_f64());
113                FLUSH_ELAPSED
114                    .with_label_values(&["upload_parquet"])
115                    .observe(self.upload_parquet.as_secs_f64());
116                FLUSH_ELAPSED
117                    .with_label_values(&["upload_puffin"])
118                    .observe(self.upload_puffin.as_secs_f64());
119                if !self.compact_memtable.is_zero() {
120                    FLUSH_ELAPSED
121                        .with_label_values(&["compact_memtable"])
122                        .observe(self.upload_puffin.as_secs_f64());
123                }
124            }
125            WriteType::Compaction => {
126                COMPACTION_STAGE_ELAPSED
127                    .with_label_values(&["iter_source"])
128                    .observe(self.iter_source.as_secs_f64());
129                COMPACTION_STAGE_ELAPSED
130                    .with_label_values(&["write_batch"])
131                    .observe(self.write_batch.as_secs_f64());
132                COMPACTION_STAGE_ELAPSED
133                    .with_label_values(&["update_index"])
134                    .observe(self.update_index.as_secs_f64());
135                COMPACTION_STAGE_ELAPSED
136                    .with_label_values(&["upload_parquet"])
137                    .observe(self.upload_parquet.as_secs_f64());
138                COMPACTION_STAGE_ELAPSED
139                    .with_label_values(&["upload_puffin"])
140                    .observe(self.upload_puffin.as_secs_f64());
141            }
142        };
143    }
144}
145
146/// A layer to access SST files under the same directory.
147pub struct AccessLayer {
148    table_dir: String,
149    /// Path type for generating file paths.
150    path_type: PathType,
151    /// Target object store.
152    object_store: ObjectStore,
153    /// Puffin manager factory for index.
154    puffin_manager_factory: PuffinManagerFactory,
155    /// Intermediate manager for inverted index.
156    intermediate_manager: IntermediateManager,
157}
158
159impl std::fmt::Debug for AccessLayer {
160    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
161        f.debug_struct("AccessLayer")
162            .field("table_dir", &self.table_dir)
163            .finish()
164    }
165}
166
167impl AccessLayer {
168    /// Returns a new [AccessLayer] for specific `table_dir`.
169    pub fn new(
170        table_dir: impl Into<String>,
171        path_type: PathType,
172        object_store: ObjectStore,
173        puffin_manager_factory: PuffinManagerFactory,
174        intermediate_manager: IntermediateManager,
175    ) -> AccessLayer {
176        AccessLayer {
177            table_dir: table_dir.into(),
178            path_type,
179            object_store,
180            puffin_manager_factory,
181            intermediate_manager,
182        }
183    }
184
185    /// Returns the directory of the table.
186    pub fn table_dir(&self) -> &str {
187        &self.table_dir
188    }
189
190    /// Returns the object store of the layer.
191    pub fn object_store(&self) -> &ObjectStore {
192        &self.object_store
193    }
194
195    /// Returns the path type of the layer.
196    pub fn path_type(&self) -> PathType {
197        self.path_type
198    }
199
200    /// Returns the puffin manager factory.
201    pub fn puffin_manager_factory(&self) -> &PuffinManagerFactory {
202        &self.puffin_manager_factory
203    }
204
205    /// Returns the intermediate manager.
206    pub fn intermediate_manager(&self) -> &IntermediateManager {
207        &self.intermediate_manager
208    }
209
210    /// Build the puffin manager.
211    pub(crate) fn build_puffin_manager(&self) -> SstPuffinManager {
212        let store = self.object_store.clone();
213        let path_provider =
214            RegionFilePathFactory::new(self.table_dir().to_string(), self.path_type());
215        self.puffin_manager_factory.build(store, path_provider)
216    }
217
218    pub(crate) async fn delete_index(
219        &self,
220        index_file_id: RegionIndexId,
221    ) -> Result<(), crate::error::Error> {
222        let path = location::index_file_path(
223            &self.table_dir,
224            RegionIndexId::new(index_file_id.file_id, index_file_id.version),
225            self.path_type,
226        );
227        self.object_store
228            .delete(&path)
229            .await
230            .context(DeleteIndexSnafu {
231                file_id: index_file_id.file_id(),
232            })?;
233        Ok(())
234    }
235
236    pub(crate) async fn delete_ssts(
237        &self,
238        region_id: RegionId,
239        file_ids: &[FileId],
240    ) -> Result<(), crate::error::Error> {
241        if file_ids.is_empty() {
242            return Ok(());
243        }
244
245        let attempted_files = file_ids.to_vec();
246        let paths: Vec<_> = file_ids
247            .iter()
248            .map(|file_id| {
249                location::sst_file_path(
250                    &self.table_dir,
251                    RegionFileId::new(region_id, *file_id),
252                    self.path_type,
253                )
254            })
255            .collect();
256
257        let mut deleter = self
258            .object_store
259            .deleter()
260            .await
261            .with_context(|_| DeleteSstsSnafu {
262                region_id,
263                file_ids: attempted_files.clone(),
264            })?;
265        deleter
266            .delete_iter(paths.iter().map(String::as_str))
267            .await
268            .with_context(|_| DeleteSstsSnafu {
269                region_id,
270                file_ids: attempted_files.clone(),
271            })?;
272        deleter.close().await.with_context(|_| DeleteSstsSnafu {
273            region_id,
274            file_ids: attempted_files,
275        })?;
276
277        Ok(())
278    }
279
280    pub(crate) async fn delete_indexes(
281        &self,
282        index_ids: &[RegionIndexId],
283    ) -> Result<(), crate::error::Error> {
284        if index_ids.is_empty() {
285            return Ok(());
286        }
287
288        let file_ids: Vec<_> = index_ids
289            .iter()
290            .map(|index_id| index_id.file_id())
291            .collect();
292        let paths: Vec<_> = index_ids
293            .iter()
294            .map(|index_id| location::index_file_path(&self.table_dir, *index_id, self.path_type))
295            .collect();
296
297        let mut deleter = self
298            .object_store
299            .deleter()
300            .await
301            .context(DeleteIndexesSnafu {
302                file_ids: file_ids.clone(),
303            })?;
304        deleter
305            .delete_iter(paths.iter().map(String::as_str))
306            .await
307            .context(DeleteIndexesSnafu {
308                file_ids: file_ids.clone(),
309            })?;
310        deleter
311            .close()
312            .await
313            .context(DeleteIndexesSnafu { file_ids })?;
314
315        Ok(())
316    }
317
318    /// Returns the directory of the region in the table.
319    pub fn build_region_dir(&self, region_id: RegionId) -> String {
320        region_dir_from_table_dir(&self.table_dir, region_id, self.path_type)
321    }
322
323    /// Returns a reader builder for specific `file`.
324    pub(crate) fn read_sst(&self, file: FileHandle) -> ParquetReaderBuilder {
325        ParquetReaderBuilder::new(
326            self.table_dir.clone(),
327            self.path_type,
328            file,
329            self.object_store.clone(),
330        )
331    }
332
333    /// Writes a SST with specific `file_id` and `metadata` to the layer.
334    ///
335    /// Returns the info of the SST. If no data written, returns None.
336    pub async fn write_sst(
337        &self,
338        request: SstWriteRequest,
339        write_opts: &WriteOptions,
340        metrics: &mut Metrics,
341    ) -> Result<SstInfoArray> {
342        let region_id = request.metadata.region_id;
343        let region_metadata = request.metadata.clone();
344        let cache_manager = request.cache_manager.clone();
345        let override_sequence = if request.preserve_row_sequence {
346            None
347        } else {
348            request.max_sequence
349        };
350
351        let sst_info = if let Some(write_cache) = cache_manager.write_cache() {
352            // Write to the write cache.
353            write_cache
354                .write_and_upload_sst(
355                    request,
356                    SstUploadRequest {
357                        dest_path_provider: RegionFilePathFactory::new(
358                            self.table_dir.clone(),
359                            self.path_type,
360                        ),
361                        remote_store: self.object_store.clone(),
362                    },
363                    write_opts,
364                    metrics,
365                )
366                .await?
367        } else {
368            // Write cache is disabled.
369            let store = self.object_store.clone();
370            let path_provider = RegionFilePathFactory::new(self.table_dir.clone(), self.path_type);
371            let indexer_builder = IndexerBuilderImpl {
372                build_type: request.op_type.into(),
373                metadata: request.metadata.clone(),
374                puffin_manager: self
375                    .puffin_manager_factory
376                    .build(store, path_provider.clone()),
377                write_cache_enabled: false,
378                intermediate_manager: self.intermediate_manager.clone(),
379                index_options: request.index_options,
380                inverted_index_config: request.inverted_index_config,
381                fulltext_index_config: request.fulltext_index_config,
382                bloom_filter_index_config: request.bloom_filter_index_config,
383                #[cfg(feature = "vector_index")]
384                vector_index_config: request.vector_index_config,
385            };
386            // We disable write cache on file system but we still use atomic write.
387            // TODO(yingwen): If we support other non-fs stores without the write cache, then
388            // we may have find a way to check whether we need the cleaner.
389            let cleaner = TempFileCleaner::new(region_id, self.object_store.clone());
390            let mut writer = ParquetWriter::new_with_object_store(
391                self.object_store.clone(),
392                request.metadata,
393                request.index_config,
394                indexer_builder,
395                path_provider,
396                metrics,
397            )
398            .await
399            .with_file_cleaner(cleaner);
400            match request.sst_write_format {
401                FormatType::PrimaryKey => {
402                    writer
403                        .write_all_flat_as_primary_key(
404                            request.source,
405                            override_sequence,
406                            write_opts,
407                        )
408                        .await?
409                }
410                FormatType::Flat => {
411                    writer
412                        .write_all_flat(request.source, override_sequence, write_opts)
413                        .await?
414                }
415            }
416        };
417
418        // Put parquet metadata to cache manager.
419        if !sst_info.is_empty() && cache_manager.sst_meta_cache_enabled() {
420            for sst in &sst_info {
421                if let Some(parquet_metadata) = &sst.file_metadata {
422                    let file_id = RegionFileId::new(region_id, sst.file_id);
423                    let file_path = format!(
424                        "region_id={}, file_id={}",
425                        file_id.region_id(),
426                        file_id.file_id()
427                    );
428                    let page_index_policy = if parquet_metadata.offset_index().is_some() {
429                        PageIndexPolicy::Optional
430                    } else {
431                        PageIndexPolicy::Skip
432                    };
433                    let parquet_metadata = parquet_metadata.clone();
434                    let region_metadata = region_metadata.clone();
435                    let cache_manager = cache_manager.clone();
436                    // Compact cache preparation is best-effort. Run the entire operation in one
437                    // detached blocking task so it neither blocks an async worker nor delays the
438                    // SST write.
439                    common_runtime::spawn_blocking_global(move || {
440                        match prepare_sst_meta_sync(
441                            &file_path,
442                            Arc::unwrap_or_clone(parquet_metadata),
443                            Some(region_metadata),
444                            page_index_policy,
445                        ) {
446                            Ok(SstMetaPreparation::Prepared(metadata)) => {
447                                cache_manager.put_prepared_sst_meta(file_id, metadata, true);
448                            }
449                            Ok(SstMetaPreparation::DecodedOnly { encoding_error, .. }) => warn!(
450                                encoding_error;
451                                "Failed to encode parquet metadata for cache, file: {}",
452                                file_path
453                            ),
454                            Err(err) => {
455                                warn!(err; "Failed to cache parquet metadata for {}", file_path);
456                            }
457                        }
458                    });
459                }
460            }
461        }
462
463        Ok(sst_info)
464    }
465
466    /// Puts encoded SST bytes to the write cache (if enabled) and uploads it to the object store.
467    pub(crate) async fn put_sst(
468        &self,
469        data: &bytes::Bytes,
470        region_id: RegionId,
471        sst_info: &SstInfo,
472        cache_manager: &CacheManagerRef,
473    ) -> Result<Metrics> {
474        if let Some(write_cache) = cache_manager.write_cache() {
475            // Write to cache and upload to remote store
476            let upload_request = SstUploadRequest {
477                dest_path_provider: RegionFilePathFactory::new(
478                    self.table_dir.clone(),
479                    self.path_type,
480                ),
481                remote_store: self.object_store.clone(),
482            };
483            write_cache
484                .put_and_upload_sst(data, region_id, sst_info, upload_request)
485                .await
486        } else {
487            let start = Instant::now();
488            let cleaner = TempFileCleaner::new(region_id, self.object_store.clone());
489            let path_provider = RegionFilePathFactory::new(self.table_dir.clone(), self.path_type);
490            let sst_file_path =
491                path_provider.build_sst_file_path(RegionFileId::new(region_id, sst_info.file_id));
492            let mut writer = self
493                .object_store
494                .writer_with(&sst_file_path)
495                .chunk(DEFAULT_WRITE_BUFFER_SIZE.as_bytes() as usize)
496                .concurrent(DEFAULT_WRITE_CONCURRENCY)
497                .await
498                .context(OpenDalSnafu)?;
499            if let Err(err) = writer.write(data.clone()).await.context(OpenDalSnafu) {
500                cleaner.clean_by_file_id(sst_info.file_id).await;
501                return Err(err);
502            }
503            if let Err(err) = writer.close().await.context(OpenDalSnafu) {
504                cleaner.clean_by_file_id(sst_info.file_id).await;
505                return Err(err);
506            }
507            let mut metrics = Metrics::new(WriteType::Flush);
508            metrics.write_batch = start.elapsed();
509            Ok(metrics)
510        }
511    }
512
513    /// Lists the SST entries from the storage layer in the table directory.
514    pub fn storage_sst_entries(&self) -> impl Stream<Item = Result<StorageSstEntry>> + use<> {
515        let object_store = self.object_store.clone();
516        let table_dir = self.table_dir.clone();
517
518        try_stream! {
519            let mut lister = object_store
520                .lister_with(table_dir.as_str())
521                .recursive(true)
522                .await
523                .context(OpenDalSnafu)?;
524
525            while let Some(entry) = lister.try_next().await.context(OpenDalSnafu)? {
526                let metadata = entry.metadata();
527                if metadata.is_dir() {
528                    continue;
529                }
530
531                let path = entry.path();
532                if !path.ends_with(".parquet") && !path.ends_with(".puffin") {
533                    continue;
534                }
535
536                let file_size = metadata.content_length();
537                let file_size = if file_size == 0 { None } else { Some(file_size) };
538                let last_modified_ms = metadata
539                    .last_modified()
540                    .map(|ts| Timestamp::new_millisecond(ts.into_inner().as_millisecond()));
541
542                let entry = StorageSstEntry {
543                    file_path: path.to_string(),
544                    file_size,
545                    last_modified_ms,
546                    node_id: None,
547                };
548
549                yield entry;
550            }
551        }
552    }
553}
554
555/// `OperationType` represents the origin of the `SstWriteRequest`.
556#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
557pub enum OperationType {
558    Flush,
559    Compact,
560}
561
562/// Contents to build a SST.
563pub struct SstWriteRequest {
564    pub op_type: OperationType,
565    pub metadata: RegionMetadataRef,
566    pub source: FlatSource,
567    pub cache_manager: CacheManagerRef,
568    #[allow(dead_code)]
569    pub storage: Option<String>,
570    pub max_sequence: Option<SequenceNumber>,
571    pub sst_write_format: FormatType,
572
573    pub preserve_row_sequence: bool,
574
575    /// Configs for index
576    pub index_options: IndexOptions,
577    pub index_config: IndexConfig,
578    pub inverted_index_config: InvertedIndexConfig,
579    pub fulltext_index_config: FulltextIndexConfig,
580    pub bloom_filter_index_config: BloomFilterConfig,
581    #[cfg(feature = "vector_index")]
582    pub vector_index_config: crate::config::VectorIndexConfig,
583}
584
585/// Cleaner to remove temp files on the atomic write dir.
586pub(crate) struct TempFileCleaner {
587    region_id: RegionId,
588    object_store: ObjectStore,
589}
590
591impl TempFileCleaner {
592    /// Constructs the cleaner for the region and store.
593    pub(crate) fn new(region_id: RegionId, object_store: ObjectStore) -> Self {
594        Self {
595            region_id,
596            object_store,
597        }
598    }
599
600    /// Removes the SST and index file from the local atomic dir by the file id.
601    /// This only removes the initial index, since the index version is always 0 for a new SST, this method should be safe to pass 0.
602    pub(crate) async fn clean_by_file_id(&self, file_id: FileId) {
603        let sst_key = IndexKey::new(self.region_id, file_id, FileType::Parquet).to_string();
604        let index_key = IndexKey::new(self.region_id, file_id, FileType::Puffin(0)).to_string();
605
606        Self::clean_atomic_dir_files(&self.object_store, &[&sst_key, &index_key]).await;
607    }
608
609    /// Removes the files from the local atomic dir by their names.
610    pub(crate) async fn clean_atomic_dir_files(
611        local_store: &ObjectStore,
612        names_to_remove: &[&str],
613    ) {
614        // We don't know the actual suffix of the file under atomic dir, so we have
615        // to list the dir. The cost should be acceptable as there won't be to many files.
616        let Ok(entries) = local_store.list(ATOMIC_WRITE_DIR).await.inspect_err(|e| {
617            if e.kind() != ErrorKind::NotFound {
618                common_telemetry::error!(e; "Failed to list tmp files for {:?}", names_to_remove)
619            }
620        }) else {
621            return;
622        };
623
624        // In our case, we can ensure the file id is unique so it is safe to remove all files
625        // with the same file id under the atomic write dir.
626        let actual_files: Vec<_> = entries
627            .into_iter()
628            .filter_map(|entry| {
629                if entry.metadata().is_dir() {
630                    return None;
631                }
632
633                // Remove name that matches files_to_remove.
634                let should_remove = names_to_remove
635                    .iter()
636                    .any(|file| entry.name().starts_with(file));
637                if should_remove {
638                    Some(entry.path().to_string())
639                } else {
640                    None
641                }
642            })
643            .collect();
644
645        common_telemetry::warn!(
646            "Clean files {:?} under atomic write dir for {:?}",
647            actual_files,
648            names_to_remove
649        );
650
651        if let Err(e) = local_store.delete_iter(actual_files).await {
652            common_telemetry::error!(e; "Failed to delete tmp file for {:?}", names_to_remove);
653        }
654    }
655}
656
657pub(crate) async fn new_fs_cache_store(root: &str) -> Result<ObjectStore> {
658    let atomic_write_dir = join_dir(root, ATOMIC_WRITE_DIR);
659    clean_dir(&atomic_write_dir).await?;
660
661    // Compatible code. Remove this after a major release.
662    let old_atomic_temp_dir = join_dir(root, OLD_ATOMIC_WRITE_DIR);
663    clean_dir(&old_atomic_temp_dir).await?;
664
665    let builder = Fs::default().root(root).atomic_write_dir(&atomic_write_dir);
666    let store = ObjectStore::new(builder).context(OpenDalSnafu)?;
667
668    Ok(with_instrument_layers(store, false))
669}
670
671/// Clean the directory.
672async fn clean_dir(dir: &str) -> Result<()> {
673    if tokio::fs::try_exists(dir)
674        .await
675        .context(CleanDirSnafu { dir })?
676    {
677        tokio::fs::remove_dir_all(dir)
678            .await
679            .context(CleanDirSnafu { dir })?;
680    }
681
682    Ok(())
683}
684
685/// Path provider for SST file and index file.
686pub trait FilePathProvider: Send + Sync {
687    /// Creates index file path of given file id. Version default to 0, and not shown in the path.
688    fn build_index_file_path(&self, file_id: RegionFileId) -> String;
689
690    /// Creates index file path of given index id (with version support).
691    fn build_index_file_path_with_version(&self, index_id: RegionIndexId) -> String;
692
693    /// Creates SST file path of given file id.
694    fn build_sst_file_path(&self, file_id: RegionFileId) -> String;
695}
696
697/// Path provider that builds paths in local write cache.
698#[derive(Clone)]
699pub(crate) struct WriteCachePathProvider {
700    file_cache: FileCacheRef,
701}
702
703impl WriteCachePathProvider {
704    /// Creates a new `WriteCachePathProvider` instance.
705    pub fn new(file_cache: FileCacheRef) -> Self {
706        Self { file_cache }
707    }
708}
709
710impl FilePathProvider for WriteCachePathProvider {
711    fn build_index_file_path(&self, file_id: RegionFileId) -> String {
712        let puffin_key = IndexKey::new(file_id.region_id(), file_id.file_id(), FileType::Puffin(0));
713        self.file_cache.cache_file_path(puffin_key)
714    }
715
716    fn build_index_file_path_with_version(&self, index_id: RegionIndexId) -> String {
717        let puffin_key = IndexKey::new(
718            index_id.region_id(),
719            index_id.file_id(),
720            FileType::Puffin(index_id.version),
721        );
722        self.file_cache.cache_file_path(puffin_key)
723    }
724
725    fn build_sst_file_path(&self, file_id: RegionFileId) -> String {
726        let parquet_file_key =
727            IndexKey::new(file_id.region_id(), file_id.file_id(), FileType::Parquet);
728        self.file_cache.cache_file_path(parquet_file_key)
729    }
730}
731
732/// Path provider that builds paths in region storage path.
733#[derive(Clone, Debug)]
734pub(crate) struct RegionFilePathFactory {
735    pub(crate) table_dir: String,
736    pub(crate) path_type: PathType,
737}
738
739impl RegionFilePathFactory {
740    /// Creates a new `RegionFilePathFactory` instance.
741    pub fn new(table_dir: String, path_type: PathType) -> Self {
742        Self {
743            table_dir,
744            path_type,
745        }
746    }
747}
748
749impl FilePathProvider for RegionFilePathFactory {
750    fn build_index_file_path(&self, file_id: RegionFileId) -> String {
751        location::index_file_path_legacy(&self.table_dir, file_id, self.path_type)
752    }
753
754    fn build_index_file_path_with_version(&self, index_id: RegionIndexId) -> String {
755        location::index_file_path(&self.table_dir, index_id, self.path_type)
756    }
757
758    fn build_sst_file_path(&self, file_id: RegionFileId) -> String {
759        location::sst_file_path(&self.table_dir, file_id, self.path_type)
760    }
761}
mito2/access_layer.rs

mito2/
access_layer.rs
