mito2/manifest/storage/
delta.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;
16
17use common_datasource::compression::CompressionType;
18use common_telemetry::debug;
19use futures::TryStreamExt;
20use futures::future::try_join_all;
21use object_store::{Entry, ErrorKind, Lister, ObjectStore};
22use snafu::{ResultExt, ensure};
23use store_api::ManifestVersion;
24use store_api::storage::RegionId;
25use tokio::sync::Semaphore;
26
27use crate::cache::manifest_cache::ManifestCache;
28use crate::error::{
29    CompressObjectSnafu, DecompressObjectSnafu, InvalidScanIndexSnafu, ManifestDeltaNotFoundSnafu,
30    OpenDalSnafu, Result,
31};
32use crate::manifest::storage::size_tracker::Tracker;
33use crate::manifest::storage::utils::{
34    get_from_cache, put_to_cache, sort_manifests, write_and_put_cache,
35};
36use crate::manifest::storage::{
37    FETCH_MANIFEST_PARALLELISM, delta_file, file_compress_type, file_version, gen_path,
38    is_delta_file, list_start_after,
39};
40
41#[derive(Debug, Clone)]
42pub(crate) struct DeltaStorage<T: Tracker> {
43    object_store: ObjectStore,
44    compress_type: CompressionType,
45    path: String,
46    delta_tracker: Arc<T>,
47    manifest_cache: Option<ManifestCache>,
48}
49
50impl<T: Tracker> DeltaStorage<T> {
51    pub(crate) fn new(
52        path: String,
53        object_store: ObjectStore,
54        compress_type: CompressionType,
55        manifest_cache: Option<ManifestCache>,
56        delta_tracker: Arc<T>,
57    ) -> Self {
58        Self {
59            object_store,
60            compress_type,
61            path,
62            delta_tracker,
63            manifest_cache,
64        }
65    }
66
67    pub(crate) fn path(&self) -> &str {
68        &self.path
69    }
70
71    pub(crate) fn object_store(&self) -> &ObjectStore {
72        &self.object_store
73    }
74
75    fn delta_file_path(&self, version: ManifestVersion) -> String {
76        gen_path(&self.path, &delta_file(version), self.compress_type)
77    }
78
79    /// Returns an iterator of manifests from path directory.
80    ///
81    /// If `start_after` is `Some`, the lister will skip entries whose name is
82    /// lexicographically less than or equal to it (see OpenDAL's `start_after`).
83    pub(crate) async fn manifest_lister(
84        &self,
85        start_after: Option<&str>,
86    ) -> Result<Option<Lister>> {
87        let mut builder = self.object_store.lister_with(&self.path);
88        if let Some(s) = start_after {
89            builder = builder.start_after(s);
90        }
91        match builder.await {
92            Ok(streamer) => Ok(Some(streamer)),
93            Err(e) if e.kind() == ErrorKind::NotFound => {
94                debug!("Manifest directory does not exist: {}", self.path);
95                Ok(None)
96            }
97            Err(e) => Err(e).context(OpenDalSnafu)?,
98        }
99    }
100
101    /// Return all `R`s in the directory that meet the `filter` conditions (that is, the `filter` closure returns `Some(R)`),
102    /// and discard `R` that does not meet the conditions (that is, the `filter` closure returns `None`)
103    /// Return an empty vector when directory is not found.
104    ///
105    /// `start_after` is forwarded to the underlying lister to skip entries
106    /// whose name is lexicographically less than or equal to it.
107    pub async fn get_paths<F, R>(&self, start_after: Option<&str>, mut filter: F) -> Result<Vec<R>>
108    where
109        F: FnMut(Entry) -> Option<R>,
110    {
111        let Some(streamer) = self.manifest_lister(start_after).await? else {
112            return Ok(vec![]);
113        };
114
115        streamer
116            .try_filter_map(|e| {
117                let result = filter(e);
118                async { Ok(result) }
119            })
120            .try_collect::<Vec<_>>()
121            .await
122            .context(OpenDalSnafu)
123    }
124
125    /// Scans the manifest files in the range of [start, end) and return all manifest entries.
126    pub async fn scan(
127        &self,
128        start: ManifestVersion,
129        end: ManifestVersion,
130    ) -> Result<Vec<(ManifestVersion, Entry)>> {
131        ensure!(start <= end, InvalidScanIndexSnafu { start, end });
132
133        // Push the version lower bound into the list request via
134        // `list_start_after`; skip the hint when `start == 0` (nothing to skip).
135        let start_after = (start > 0).then(|| list_start_after(&self.path, start));
136        let mut total_paths = 0;
137        let mut entries: Vec<(ManifestVersion, Entry)> = self
138            .get_paths(start_after.as_deref(), |entry| {
139                total_paths += 1;
140                let file_name = entry.name();
141                if is_delta_file(file_name) {
142                    let version = file_version(file_name);
143                    if start <= version && version < end {
144                        return Some((version, entry));
145                    }
146                }
147                None
148            })
149            .await?;
150
151        sort_manifests(&mut entries);
152
153        common_telemetry::debug!(
154            "DeltaStorage get paths for {}, start: {}, end: {}, start_after: {:?}, total_paths: {}, entries: {}",
155            self.path,
156            start,
157            end,
158            start_after,
159            total_paths,
160            entries.len()
161        );
162
163        Ok(entries)
164    }
165
166    /// Fetches manifests in range [start_version, end_version).
167    ///
168    /// This functions is guaranteed to return manifests from the `start_version` strictly (must contain `start_version`).
169    pub async fn fetch_manifests_strict_from(
170        &self,
171        start_version: ManifestVersion,
172        end_version: ManifestVersion,
173        region_id: RegionId,
174    ) -> Result<Vec<(ManifestVersion, Vec<u8>)>> {
175        let mut manifests = self.fetch_manifests(start_version, end_version).await?;
176        let start_index = manifests.iter().position(|(v, _)| *v == start_version);
177        debug!(
178            "Fetches manifests in range [{},{}), start_index: {:?}, region_id: {}, manifests: {:?}",
179            start_version,
180            end_version,
181            start_index,
182            region_id,
183            manifests.iter().map(|(v, _)| *v).collect::<Vec<_>>()
184        );
185        if let Some(start_index) = start_index {
186            Ok(manifests.split_off(start_index))
187        } else {
188            Ok(vec![])
189        }
190    }
191
192    /// Common implementation for fetching manifests from entries in parallel.
193    pub(crate) async fn fetch_manifests_from_entries(
194        &self,
195        entries: Vec<(ManifestVersion, Entry)>,
196    ) -> Result<Vec<(ManifestVersion, Vec<u8>)>> {
197        if entries.is_empty() {
198            return Ok(vec![]);
199        }
200
201        // TODO(weny): Make it configurable.
202        let semaphore = Semaphore::new(FETCH_MANIFEST_PARALLELISM);
203
204        let tasks = entries.iter().map(|(v, entry)| async {
205            // Safety: semaphore must exist.
206            let _permit = semaphore.acquire().await.unwrap();
207
208            let cache_key = entry.path();
209            // Try to get from cache first
210            if let Some(data) = get_from_cache(self.manifest_cache.as_ref(), cache_key).await {
211                return Ok((*v, data));
212            }
213
214            // Fetch from remote object store
215            let compress_type = file_compress_type(entry.name());
216            let bytes = match self.object_store.read(entry.path()).await {
217                Ok(bytes) => bytes,
218                Err(error) if error.kind() == ErrorKind::NotFound => {
219                    return Err(error).context(ManifestDeltaNotFoundSnafu {
220                        version: *v,
221                        path: entry.path(),
222                    });
223                }
224                Err(error) => return Err(error).context(OpenDalSnafu),
225            };
226            let data = compress_type
227                .decode(bytes)
228                .await
229                .context(DecompressObjectSnafu {
230                    compress_type,
231                    path: entry.path(),
232                })?;
233
234            // Add to cache
235            put_to_cache(self.manifest_cache.as_ref(), cache_key.to_string(), &data).await;
236
237            Ok((*v, data))
238        });
239
240        try_join_all(tasks).await
241    }
242
243    /// Fetch all manifests in concurrent, and return the manifests in range [start_version, end_version)
244    ///
245    /// **Notes**: This function is no guarantee to return manifests from the `start_version` strictly.
246    /// Uses [fetch_manifests_strict_from](DeltaStorage::fetch_manifests_strict_from) to get manifests from the `start_version`.
247    pub async fn fetch_manifests(
248        &self,
249        start_version: ManifestVersion,
250        end_version: ManifestVersion,
251    ) -> Result<Vec<(ManifestVersion, Vec<u8>)>> {
252        let manifests = self.scan(start_version, end_version).await?;
253        self.fetch_manifests_from_entries(manifests).await
254    }
255
256    /// Save the delta manifest file.
257    pub async fn save(&mut self, version: ManifestVersion, bytes: &[u8]) -> Result<()> {
258        let path = self.delta_file_path(version);
259        debug!("Save log to manifest storage, version: {}", version);
260        let data = self
261            .compress_type
262            .encode(bytes)
263            .await
264            .context(CompressObjectSnafu {
265                compress_type: self.compress_type,
266                path: &path,
267            })?;
268        let delta_size = data.len();
269
270        write_and_put_cache(
271            &self.object_store,
272            self.manifest_cache.as_ref(),
273            &path,
274            data,
275        )
276        .await?;
277        self.delta_tracker.record(version, delta_size as u64);
278
279        Ok(())
280    }
281}
282
283#[cfg(test)]
284impl<T: Tracker> DeltaStorage<T> {
285    pub fn set_compress_type(&mut self, compress_type: CompressionType) {
286        self.compress_type = compress_type;
287    }
288}
mito2/manifest/storage/delta.rs

mito2/manifest/storage/
delta.rs
