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, warn};
19use futures::TryStreamExt;
20use futures::future::try_join_all;
21use object_store::{Buffer, 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    ManifestVersionConflictSnafu, OpenDalSnafu, Result,
31};
32use crate::manifest::storage::size_tracker::Tracker;
33use crate::manifest::storage::utils::{
34    get_from_cache_decoded, 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            let compress_type = file_compress_type(entry.name());
210            // Try to get from cache first
211            if let Some(data) = get_from_cache_decoded(self.manifest_cache.as_ref(), cache_key, compress_type).await {
212                let size = entry.metadata().content_length();
213                if size > 0 {
214                    self.delta_tracker.record(*v, size);
215                } else {
216                    match self.object_store.stat(entry.path()).await {
217                        Ok(metadata) => self.delta_tracker.record(*v, metadata.content_length()),
218                        Err(error) => warn!(error; "Failed to account for cached manifest delta {}", entry.path()),
219                    }
220                }
221                return Ok((*v, data));
222            }
223
224            // Fetch from remote object store
225            let bytes = match self.object_store.read(entry.path()).await {
226                Ok(bytes) => bytes,
227                Err(error) if error.kind() == ErrorKind::NotFound => {
228                    return Err(error).context(ManifestDeltaNotFoundSnafu {
229                        version: *v,
230                        path: entry.path(),
231                    });
232                }
233                Err(error) => return Err(error).context(OpenDalSnafu),
234            };
235            self.delta_tracker.record(*v, bytes.len() as u64);
236            let data = compress_type
237                .decode(bytes.clone())
238                .await
239                .context(DecompressObjectSnafu {
240                    compress_type,
241                    path: entry.path(),
242                })?;
243
244            // Add to cache
245            if let Some(cache) = &self.manifest_cache {
246                cache.put_file(cache_key.to_string(), bytes.to_vec()).await;
247            }
248
249            Ok((*v, data))
250        });
251
252        try_join_all(tasks).await
253    }
254
255    /// Fetch all manifests in concurrent, and return the manifests in range [start_version, end_version)
256    ///
257    /// **Notes**: This function is no guarantee to return manifests from the `start_version` strictly.
258    /// Uses [fetch_manifests_strict_from](DeltaStorage::fetch_manifests_strict_from) to get manifests from the `start_version`.
259    pub async fn fetch_manifests(
260        &self,
261        start_version: ManifestVersion,
262        end_version: ManifestVersion,
263    ) -> Result<Vec<(ManifestVersion, Vec<u8>)>> {
264        let manifests = self.scan(start_version, end_version).await?;
265        self.fetch_manifests_from_entries(manifests).await
266    }
267
268    /// Save the delta manifest file.
269    pub async fn save(&mut self, version: ManifestVersion, bytes: &[u8]) -> Result<()> {
270        let path = self.delta_file_path(version);
271        debug!("Save log to manifest storage, version: {}", version);
272        let data = self
273            .compress_type
274            .encode(bytes)
275            .await
276            .context(CompressObjectSnafu {
277                compress_type: self.compress_type,
278                path: &path,
279            })?;
280        let delta_size = data.len();
281
282        write_and_put_cache(
283            &self.object_store,
284            self.manifest_cache.as_ref(),
285            &path,
286            data,
287        )
288        .await?;
289        self.delta_tracker.record(version, delta_size as u64);
290
291        Ok(())
292    }
293
294    /// Creates the delta of `version` under its fixed `.json.gz` name, so that only one writer can
295    /// commit the version.
296    ///
297    /// An existing object with the same decoded content is a retry of an earlier write that
298    /// landed. Any other existing object fails with a version conflict.
299    pub(crate) async fn save_conditional(
300        &mut self,
301        version: ManifestVersion,
302        bytes: &[u8],
303    ) -> Result<()> {
304        let path = gen_path(&self.path, &delta_file(version), CompressionType::Gzip);
305        // Every writer has to use one object name, so `compress_manifest = false` is honoured
306        // with gzip level 0, which stores the bytes without compressing them.
307        let data = if self.compress_type.is_compressed() {
308            CompressionType::Gzip.encode(bytes).await
309        } else {
310            CompressionType::encode_gzip_stored(bytes).await
311        }
312        .context(CompressObjectSnafu {
313            compress_type: CompressionType::Gzip,
314            path: &path,
315        })?;
316        let data = Buffer::from(data);
317
318        let stored_size = match self
319            .object_store
320            .write_with(&path, data.clone())
321            .if_not_exists(true)
322            .await
323        {
324            Ok(_) => data.len(),
325            Err(error)
326                if matches!(
327                    error.kind(),
328                    ErrorKind::AlreadyExists | ErrorKind::ConditionNotMatch
329                ) =>
330            {
331                self.ensure_same_delta(version, &path, bytes).await?
332            }
333            Err(error) => return Err(error).context(OpenDalSnafu),
334        };
335
336        self.delta_tracker.record(version, stored_size as u64);
337        if let Some(cache) = &self.manifest_cache {
338            cache.put_file(path, data.to_vec()).await;
339        }
340
341        Ok(())
342    }
343
344    /// Checks that the object at `path` decodes to `bytes`, reading it directly from storage.
345    /// Returns the size of the stored object.
346    async fn ensure_same_delta(
347        &self,
348        version: ManifestVersion,
349        path: &str,
350        bytes: &[u8],
351    ) -> Result<usize> {
352        let conflict = |reason: String| {
353            ManifestVersionConflictSnafu {
354                version,
355                manifest_dir: &self.path,
356                reason,
357            }
358            .build()
359        };
360        let existing = self
361            .object_store
362            .read(path)
363            .await
364            .map_err(|e| conflict(format!("cannot read the existing delta: {e}")))?;
365        let stored_size = existing.len();
366        let existing = CompressionType::Gzip
367            .decode(existing)
368            .await
369            .map_err(|e| conflict(format!("cannot decode the existing delta: {e}")))?;
370        if existing != bytes {
371            return Err(conflict(
372                "the existing delta has different content".to_string(),
373            ));
374        }
375        Ok(stored_size)
376    }
377}
378
379#[cfg(test)]
380impl<T: Tracker> DeltaStorage<T> {
381    pub fn set_compress_type(&mut self, compress_type: CompressionType) {
382        self.compress_type = compress_type;
383    }
384}
mito2/manifest/storage/delta.rs

mito2/manifest/storage/
delta.rs
