mito2/manifest/storage/
utils.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 common_datasource::compression::CompressionType;
16use common_telemetry::debug;
17use object_store::{Entry, ObjectStore};
18use snafu::ResultExt;
19use store_api::ManifestVersion;
20
21use crate::cache::manifest_cache::ManifestCache;
22use crate::error::{OpenDalSnafu, Result};
23
24const CHUNK_SIZE: usize = 64 * 1024 * 1024; // 64MB
25
26/// Gets and decodes the object payload cached under `key`.
27///
28/// Older remote fills cached decoded content. If decoding fails, discard the
29/// entry and let the caller refill it from object storage. Remote decode and
30/// checksum errors are still returned by the caller.
31pub(crate) async fn get_from_cache_decoded(
32    cache: Option<&ManifestCache>,
33    key: &str,
34    compress_type: CompressionType,
35) -> Option<Vec<u8>> {
36    let cache = cache?;
37    let raw = cache.get_file(key).await?;
38    match compress_type.decode(raw.as_slice()).await {
39        Ok(data) => Some(data),
40        Err(error) => {
41            debug!(
42                "Failed to decode cached manifest {}, removing entry: {}",
43                key, error
44            );
45            cache.remove(key).await;
46            None
47        }
48    }
49}
50
51/// Removes a manifest file from cache.
52pub(crate) async fn remove_from_cache(cache: Option<&ManifestCache>, key: &str) {
53    let Some(cache) = cache else {
54        return;
55    };
56    cache.remove(key).await
57}
58
59/// Writes data to object store and caches the same encoded payload.
60pub(crate) async fn write_and_put_cache(
61    object_store: &ObjectStore,
62    cache: Option<&ManifestCache>,
63    path: &str,
64    data: Vec<u8>,
65) -> Result<()> {
66    // Clone only when caching is enabled; retain the owner for the cache write.
67    let cache_data = cache.map(|cache| (cache, data.clone()));
68
69    // Write to object store
70    object_store
71        .write_with(path, data)
72        .chunk(CHUNK_SIZE) // 64MB
73        .await
74        .context(OpenDalSnafu)?;
75
76    // Put to cache if we cloned the data
77    if let Some((cache, data)) = cache_data {
78        cache.put_file(path.to_string(), data).await;
79    }
80
81    Ok(())
82}
83
84/// Sorts the manifest files.
85pub(crate) fn sort_manifests(entries: &mut [(ManifestVersion, Entry)]) {
86    entries.sort_unstable_by_key(|(version, _)| *version);
87}
mito2/manifest/storage/utils.rs

mito2/manifest/storage/
utils.rs
