mito2/sst/parquet/
helper.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::ops::Range;
16use std::time::Instant;
17
18use bytes::Bytes;
19use common_telemetry::trace;
20use object_store::ObjectStore;
21
22const FETCH_PARALLELISM: usize = 8;
23pub(crate) const MERGE_GAP: usize = 512 * 1024;
24
25/// Asynchronously fetches byte ranges from an object store.
26///
27/// * `FETCH_PARALLELISM` - The number of concurrent fetch operations.
28/// * `MERGE_GAP` - The maximum gap size (in bytes) to merge small byte ranges for optimized fetching.
29pub async fn fetch_byte_ranges(
30    file_path: &str,
31    object_store: ObjectStore,
32    ranges: &[Range<u64>],
33) -> object_store::Result<Vec<Bytes>> {
34    let total_size = ranges.iter().map(|r| r.end - r.start).sum::<u64>();
35    let start = Instant::now();
36
37    let result = object_store
38        .reader_with(file_path)
39        .concurrent(FETCH_PARALLELISM)
40        .gap(MERGE_GAP)
41        .await?
42        .fetch(ranges.to_vec())
43        .await?
44        .into_iter()
45        .map(|buf| buf.to_bytes())
46        .collect::<Vec<_>>();
47
48    trace!(
49        "Fetch {} bytes from '{}' in object store, cost: {:?}",
50        total_size,
51        file_path,
52        start.elapsed()
53    );
54
55    Ok(result)
56}
mito2/sst/parquet/helper.rs

mito2/sst/parquet/
helper.rs
