mito2/series_index/
version.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
15//! Immutable index snapshots and aggregate series-file handles.
16
17use std::collections::{HashMap, HashSet};
18use std::fmt::{self, Debug, Formatter};
19use std::sync::atomic::{AtomicBool, Ordering};
20use std::sync::{Arc, RwLock};
21
22use store_api::storage::{FileId, RegionId};
23
24use crate::series_index::catalog::SeriesIndexEntry;
25use crate::series_index::purger::{IndexFilePurger, PurgeRequest};
26use crate::sst::file::RegionFileId;
27
28/// A reference-counted series-index file with deferred deletion semantics.
29#[derive(Clone)]
30pub(crate) struct SeriesIndexFileHandle {
31    inner: Arc<SeriesIndexFileHandleInner>,
32}
33
34impl Debug for SeriesIndexFileHandle {
35    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
36        f.debug_struct("SeriesIndexFileHandle")
37            .field("file_id", &self.inner.file_id)
38            .field("deleted", &self.inner.deleted.load(Ordering::Relaxed))
39            .finish()
40    }
41}
42
43impl SeriesIndexFileHandle {
44    pub(crate) fn new(
45        region_id: RegionId,
46        entry: SeriesIndexEntry,
47        purger: IndexFilePurger,
48    ) -> Self {
49        Self {
50            inner: Arc::new(SeriesIndexFileHandleInner {
51                file_id: RegionFileId::new(region_id, entry.index_uuid),
52                entry,
53                deleted: AtomicBool::new(false),
54                purger,
55            }),
56        }
57    }
58
59    pub(crate) fn entry(&self) -> &SeriesIndexEntry {
60        &self.inner.entry
61    }
62
63    pub(crate) fn mark_deleted(&self) {
64        self.inner.deleted.store(true, Ordering::Release);
65    }
66}
67
68struct SeriesIndexFileHandleInner {
69    file_id: RegionFileId,
70    entry: SeriesIndexEntry,
71    deleted: AtomicBool,
72    purger: IndexFilePurger,
73}
74
75impl Drop for SeriesIndexFileHandleInner {
76    fn drop(&mut self) {
77        if self.deleted.load(Ordering::Acquire) {
78            self.purger.purge(PurgeRequest {
79                file_id: self.file_id,
80            });
81        }
82    }
83}
84
85/// Immutable series-index snapshot for one region.
86#[derive(Debug, Default)]
87pub(crate) struct SeriesIndexVersion {
88    pub(crate) range_indexes: HashSet<FileId>,
89    pub(crate) series_indexes: HashMap<FileId, SeriesIndexFileHandle>,
90}
91
92impl SeriesIndexVersion {
93    fn mark_all_deleted(&self) {
94        self.series_indexes
95            .values()
96            .for_each(SeriesIndexFileHandle::mark_deleted);
97    }
98}
99
100/// Copy-on-write series-index snapshots owned by a region.
101#[derive(Debug, Default)]
102pub(crate) struct SeriesIndexVersionControl {
103    current: RwLock<Arc<SeriesIndexVersion>>,
104}
105
106impl SeriesIndexVersionControl {
107    pub(crate) fn current(&self) -> Arc<SeriesIndexVersion> {
108        self.current.read().unwrap().clone()
109    }
110
111    pub(crate) fn publish(&self, next: Arc<SeriesIndexVersion>) -> Arc<SeriesIndexVersion> {
112        std::mem::replace(&mut *self.current.write().unwrap(), next)
113    }
114
115    pub(crate) fn mark_dropped(&self) {
116        self.publish(Arc::new(SeriesIndexVersion::default()))
117            .mark_all_deleted();
118    }
119}
mito2/series_index/version.rs

mito2/series_index/
version.rs
