risingwave_object_store/object/opendal_engine/
mod.rs

1// Copyright 2023 RisingWave Labs
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
15pub mod opendal_object_store;
16
17use opendal::Operator;
18use opendal::layers::ConcurrentLimitLayer;
19pub use opendal_object_store::*;
20use risingwave_common::config::ObjectStoreConfig;
21
22#[cfg(feature = "hdfs-backend")]
23pub mod hdfs;
24pub mod webhdfs;
25
26pub mod gcs;
27
28pub mod obs;
29
30pub mod azblob;
31pub mod oss;
32pub mod s3;
33
34pub mod fs;
35
36// To make sure the the operation is consistent, we should specially set `atomic_write_dir` for fs, hdfs and webhdfs services.
37const ATOMIC_WRITE_DIR: &str = "atomic_write_dir/";
38
39fn new_operator(config: &ObjectStoreConfig, op: Operator) -> Operator {
40    // Tokio semaphore rejects values above `usize::MAX >> 3`.
41    const UNLIMITED_OPERATION_CONCURRENCY: usize = usize::MAX >> 3;
42
43    if config.req_concurrency_limit > 0 || config.http_concurrent_limit > 0 {
44        let operation_concurrency_limit = if config.req_concurrency_limit > 0 {
45            config.req_concurrency_limit
46        } else {
47            UNLIMITED_OPERATION_CONCURRENCY
48        };
49        let mut concurrent_limit_layer = ConcurrentLimitLayer::new(operation_concurrency_limit);
50        if config.http_concurrent_limit > 0 {
51            concurrent_limit_layer =
52                concurrent_limit_layer.with_http_concurrent_limit(config.http_concurrent_limit);
53        }
54        op.layer(concurrent_limit_layer)
55    } else {
56        op
57    }
58}
risingwave_object_store/object/opendal_engine/mod.rs

risingwave_object_store/object/opendal_engine/
mod.rs
