This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 6555d4c3 feat(io): add optional OSS C++ SDK read backend (#1028)
6555d4c3 is described below
commit 6555d4c3a55a4ec86723faeeee29ecdcc5499c15
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 4 22:20:10 2026 +0800
feat(io): add optional OSS C++ SDK read backend (#1028)
---
.github/workflows/ci.yml | 35 ++
bindings/python/src/error.rs | 18 +
crates/paimon/Cargo.toml | 2 +
crates/paimon/src/error.rs | 48 +-
crates/paimon/src/io/file_io.rs | 73 +--
crates/paimon/src/io/mod.rs | 5 +
crates/paimon/src/io/storage.rs | 40 ++
crates/paimon/src/io/storage_oss_cpp.rs | 780 +++++++++++++++++++++++++++
crates/paimon/src/io/storage_oss_cpp_test.rs | 506 +++++++++++++++++
docs/src/getting-started.md | 7 +
integrations/oss-cpp/CMakeLists.txt | 25 +
integrations/oss-cpp/README.md | 80 +++
integrations/oss-cpp/bridge.cc | 190 +++++++
integrations/oss-cpp/bridge.h | 65 +++
14 files changed, 1819 insertions(+), 55 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 692a4680..3138b024 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -240,6 +240,10 @@ jobs:
needs_docker: false
needs_lumina: false
needs_uv: false
+ - suite: oss-cpp
+ needs_docker: false
+ needs_lumina: false
+ needs_uv: false
env:
RUST_LOG: DEBUG
RUST_BACKTRACE: full
@@ -338,3 +342,34 @@ jobs:
- name: Stop Docker containers
if: ${{ always() && matrix.needs_docker }}
run: make docker-down
+
+ - name: Checkout OSS C++ SDK
+ if: matrix.suite == 'oss-cpp'
+ uses: actions/checkout@v7
+ with:
+ repository: aliyun/alibabacloud-oss-cpp-sdk-v2
+ ref: c13a82038b0696f99c32f4e86b623ffcc9e62140
+ path: .oss-cpp-sdk
+
+ - name: Build OSS C++ bridge
+ if: matrix.suite == 'oss-cpp'
+ run: |
+ sudo apt-get update
+ sudo apt-get install -y libcurl4-openssl-dev libssl-dev
+ cmake -S .oss-cpp-sdk -B build/oss-sdk -DCMAKE_BUILD_TYPE=Release \
+ -DUSE_SYSTEM_CURL=ON -DUSE_SYSTEM_OPENSSL=ON
-DBUILD_SHARED_LIBS=ON \
+ -DCMAKE_CXX_FLAGS=-Wno-error=maybe-uninitialized \
+ -DCMAKE_INSTALL_PREFIX="$PWD/build/oss-install"
+ cmake --build build/oss-sdk -j 2
+ cmake --install build/oss-sdk
+ cmake -S integrations/oss-cpp -B build/oss-cpp \
+ -DCMAKE_PREFIX_PATH="$PWD/build/oss-install"
+ cmake --build build/oss-cpp -j 2
+ echo
"OSS_CPP_BRIDGE_LIBRARY=$PWD/build/oss-cpp/liboss_cpp_bridge.so" >>
"$GITHUB_ENV"
+
+ - name: OSS C++ SDK Integration Test
+ if: matrix.suite == 'oss-cpp'
+ run: |
+ cargo test --locked -p paimon --features
storage-oss-cpp,storage-jindo --lib \
+ io::storage_oss_cpp -- --include-ignored
+ cargo clippy --locked -p paimon --features storage-oss-cpp --lib
--tests -- -D warnings
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index b6ca355a..cb3199db 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -79,6 +79,24 @@ mod tests {
});
}
+ #[test]
+ fn oss_cpp_fork_error_has_distinct_python_type() {
+ Python::attach(|py| {
+ let error = paimon::Error::ProcessForkUnsupported {
+ message: "OSS C++ SDK cannot be used after fork; use spawn
workers".to_string(),
+ };
+ assert!(to_py_err(error).is_instance_of::<ForkSafetyError>(py));
+
+ let error = datafusion::error::DataFusionError::External(Box::new(
+ paimon::Error::ProcessForkUnsupported {
+ message: "OSS C++ SDK cannot be used after fork; use spawn
workers".to_string(),
+ },
+ ))
+ .context("SQL planning failed");
+ assert!(df_to_py_err(error).is_instance_of::<ForkSafetyError>(py));
+ });
+ }
+
#[test]
fn wrapped_process_fork_error_has_distinct_python_type() {
Python::attach(|py| {
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 9ca18244..72eee590 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -42,6 +42,7 @@ storage-all = [
"storage-gcs",
"storage-hdfs",
"storage-jindo",
+ "storage-oss-cpp",
]
fulltext = ["dep:paimon-ftindex-core"]
vortex = ["dep:vortex"]
@@ -63,6 +64,7 @@ storage-obs = ["dep:opendal-http-transport-reqwest",
"dep:opendal-service-obs"]
storage-gcs = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-gcs"]
storage-hdfs = ["dep:opendal-service-hdfs-native"]
storage-jindo = ["storage-oss"]
+storage-oss-cpp = ["storage-oss"]
[dependencies]
rand = "0.8.5"
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index aa53a6dc..0087f16d 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -19,6 +19,8 @@ use snafu::prelude::*;
pub(crate) const JINDO_FORK_ERROR: &str =
"Jindo SDK cannot be reused after process fork; use spawn or avoid
initializing Jindo in the parent process";
+pub(crate) const OSS_CPP_FORK_ERROR: &str =
+ "OSS C++ SDK cannot be used after fork; use spawn workers";
/// Result type used in paimon.
pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -150,8 +152,7 @@ impl Error {
source: opendal::Error,
message: impl Into<String>,
) -> Self {
- if source.kind() == opendal::ErrorKind::Unsupported &&
source.message() == JINDO_FORK_ERROR
- {
+ if is_native_fork_error(&source) {
return Error::ProcessForkUnsupported {
message: source.message().to_string(),
};
@@ -170,7 +171,7 @@ impl Error {
Error::DataInvalid { source, .. } | Error::UnexpectedError {
source, .. } => source
.as_deref()
.is_some_and(|source|
error_chain_contains_process_fork(source)),
- Error::IoUnexpected { source, .. } => is_jindo_fork_error(source),
+ Error::IoUnexpected { source, .. } => is_native_fork_error(source),
Error::DataUnexpected { source, .. } =>
error_chain_contains_process_fork(source),
Error::ParquetDataUnexpected { source, .. } => {
error_chain_contains_process_fork(source)
@@ -211,8 +212,9 @@ fn error_chain_contains_io_unexpected(error: &(dyn
std::error::Error + 'static))
.is_some_and(error_chain_contains_io_unexpected)
}
-fn is_jindo_fork_error(error: &opendal::Error) -> bool {
- error.kind() == opendal::ErrorKind::Unsupported && error.message() ==
JINDO_FORK_ERROR
+fn is_native_fork_error(error: &opendal::Error) -> bool {
+ error.kind() == opendal::ErrorKind::Unsupported
+ && matches!(error.message(), JINDO_FORK_ERROR | OSS_CPP_FORK_ERROR)
}
fn error_chain_contains_process_fork(error: &(dyn std::error::Error +
'static)) -> bool {
@@ -220,7 +222,7 @@ fn error_chain_contains_process_fork(error: &(dyn
std::error::Error + 'static))
return error.is_process_fork_unsupported();
}
if let Some(error) = error.downcast_ref::<opendal::Error>() {
- return is_jindo_fork_error(error);
+ return is_native_fork_error(error);
}
if let Some(error) = error.downcast_ref::<std::io::Error>() {
if error
@@ -280,10 +282,14 @@ mod tests {
use super::*;
#[test]
- fn preserves_jindo_fork_error_kind() {
+ fn preserves_native_fork_error_kind() {
+ for message in [JINDO_FORK_ERROR, OSS_CPP_FORK_ERROR] {
+ let error: Error =
opendal::Error::new(opendal::ErrorKind::Unsupported, message).into();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ }
let error: Error =
- opendal::Error::new(opendal::ErrorKind::Unsupported,
JINDO_FORK_ERROR).into();
- assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ opendal::Error::new(opendal::ErrorKind::Unexpected,
OSS_CPP_FORK_ERROR).into();
+ assert!(!error.is_process_fork_unsupported());
}
#[test]
@@ -309,17 +315,19 @@ mod tests {
#[test]
fn detects_wrapped_process_fork_error() {
- let error = Error::UnexpectedError {
- message: "outer context".to_string(),
- source: Some(Box::new(Error::IoUnexpected {
- message: "storage context".to_string(),
- source: Box::new(opendal::Error::new(
- opendal::ErrorKind::Unsupported,
- JINDO_FORK_ERROR,
- )),
- })),
- };
+ for message in [JINDO_FORK_ERROR, OSS_CPP_FORK_ERROR] {
+ let error = Error::UnexpectedError {
+ message: "outer context".to_string(),
+ source: Some(Box::new(Error::IoUnexpected {
+ message: "storage context".to_string(),
+ source: Box::new(opendal::Error::new(
+ opendal::ErrorKind::Unsupported,
+ message,
+ )),
+ })),
+ };
- assert!(error.is_process_fork_unsupported());
+ assert!(error.is_process_fork_unsupported());
+ }
}
}
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index e81af113..dbcebd60 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -1400,7 +1400,7 @@ mod file_action_test {
async fn create(&self, _path: &str) -> crate::Result<(Operator,
String)> {
let service: Servicer = Arc::new(CountingListService {
pulls: Arc::clone(&self.pulls),
- fail_with_fork_error: false,
+ fork_error: None,
});
Ok((
Operator::from_parts(OperationContext::default(), service),
@@ -1412,7 +1412,7 @@ mod file_action_test {
#[derive(Debug)]
struct CountingListService {
pulls: Arc<AtomicUsize>,
- fail_with_fork_error: bool,
+ fork_error: Option<&'static str>,
}
impl Service for CountingListService {
@@ -1450,8 +1450,8 @@ mod file_action_test {
_path: &str,
_args: OpStat,
) -> opendal::Result<RpStat> {
- if self.fail_with_fork_error {
- return Err(fork_test_error());
+ if let Some(message) = self.fork_error {
+ return Err(fork_test_error(message));
}
Err(unsupported_test_operation())
}
@@ -1487,7 +1487,7 @@ mod file_action_test {
Ok(CountingLister {
pulls: Arc::clone(&self.pulls),
next: 0,
- fail_with_fork_error: self.fail_with_fork_error,
+ fork_error: self.fork_error,
})
}
@@ -1529,21 +1529,21 @@ mod file_action_test {
)
}
- fn fork_test_error() -> opendal::Error {
- opendal::Error::new(opendal::ErrorKind::Unsupported, JINDO_FORK_ERROR)
+ fn fork_test_error(message: &'static str) -> opendal::Error {
+ opendal::Error::new(opendal::ErrorKind::Unsupported, message)
}
struct CountingLister {
pulls: Arc<AtomicUsize>,
next: usize,
- fail_with_fork_error: bool,
+ fork_error: Option<&'static str>,
}
impl oio::List for CountingLister {
async fn next(&mut self) -> opendal::Result<Option<oio::Entry>> {
self.pulls.fetch_add(1, AtomicOrdering::SeqCst);
- if self.fail_with_fork_error {
- return Err(fork_test_error());
+ if let Some(message) = self.fork_error {
+ return Err(fork_test_error(message));
}
if self.next == 0 {
self.next += 1;
@@ -1689,13 +1689,6 @@ mod file_action_test {
#[tokio::test]
async fn file_status_reads_preserve_fork_safety_error() {
- let pulls = Arc::new(AtomicUsize::new(0));
- let service: Servicer = Arc::new(CountingListService {
- pulls,
- fail_with_fork_error: true,
- });
- let operator = Operator::from_parts(OperationContext::default(),
service);
-
#[derive(Debug)]
struct FixedProvider(Operator);
@@ -1709,26 +1702,36 @@ mod file_action_test {
}
}
- let file_io =
setup_memory_file_io().with_provider(Arc::new(FixedProvider(operator)));
- let error = file_io
- .get_status("counting:/objects/file")
- .await
- .unwrap_err();
- assert!(
- matches!(error, Error::ProcessForkUnsupported { .. }),
- "{error:?}"
- );
+ for message in [
+ crate::error::JINDO_FORK_ERROR,
+ crate::error::OSS_CPP_FORK_ERROR,
+ ] {
+ let service: Servicer = Arc::new(CountingListService {
+ pulls: Arc::new(AtomicUsize::new(0)),
+ fork_error: Some(message),
+ });
+ let operator = Operator::from_parts(OperationContext::default(),
service);
+ let file_io =
setup_memory_file_io().with_provider(Arc::new(FixedProvider(operator)));
+ let error = file_io
+ .get_status("counting:/objects/file")
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, Error::ProcessForkUnsupported { .. }),
+ "{error:?}"
+ );
- let error =
file_io.list_status("counting:/objects/").await.unwrap_err();
- assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
+ let error =
file_io.list_status("counting:/objects/").await.unwrap_err();
+ assert!(matches!(error, Error::ProcessForkUnsupported { .. }));
- let snapshots =
- crate::table::SnapshotManager::new(file_io,
"counting:/objects/table".to_string());
- let error = snapshots.get_latest_snapshot_id().await.unwrap_err();
- assert!(
- matches!(error, Error::ProcessForkUnsupported { .. }),
- "{error:?}"
- );
+ let snapshots =
+ crate::table::SnapshotManager::new(file_io,
"counting:/objects/table".to_string());
+ let error = snapshots.get_latest_snapshot_id().await.unwrap_err();
+ assert!(
+ matches!(error, Error::ProcessForkUnsupported { .. }),
+ "{error:?}"
+ );
+ }
}
#[tokio::test]
diff --git a/crates/paimon/src/io/mod.rs b/crates/paimon/src/io/mod.rs
index 2718ab6b..c8dd33de 100644
--- a/crates/paimon/src/io/mod.rs
+++ b/crates/paimon/src/io/mod.rs
@@ -77,6 +77,11 @@ mod storage_jindo;
#[cfg(feature = "storage-jindo")]
use storage_jindo::*;
+#[cfg(feature = "storage-oss-cpp")]
+mod storage_oss_cpp;
+#[cfg(feature = "storage-oss-cpp")]
+use storage_oss_cpp::*;
+
#[cfg(feature = "storage-s3")]
mod storage_s3;
#[cfg(feature = "storage-s3")]
diff --git a/crates/paimon/src/io/storage.rs b/crates/paimon/src/io/storage.rs
index afa9b25a..76ea3624 100644
--- a/crates/paimon/src/io/storage.rs
+++ b/crates/paimon/src/io/storage.rs
@@ -41,6 +41,8 @@ use std::sync::MutexGuard;
use super::AzdlsStorageConfig;
#[cfg(feature = "storage-jindo")]
use super::JindoStorageConfig;
+#[cfg(feature = "storage-oss-cpp")]
+use super::OssCppStorageConfig;
#[cfg(feature = "storage-oss")]
use super::OssStorageConfig;
use opendal::Operator;
@@ -70,6 +72,11 @@ use super::FileIOBuilder;
/// The storage carries all supported storage services in paimon
#[derive(Debug)]
pub enum Storage {
+ #[cfg(feature = "storage-oss-cpp")]
+ OssCpp {
+ config: Box<OssCppStorageConfig>,
+ operators: Mutex<HashMap<String, Operator>>,
+ },
/// A caller-provided opendal operator, explicitly scoped to filesystem
semantics
/// (see `FileIOBuilder::with_fs_operator`).
CustomFs { op: Operator },
@@ -142,6 +149,20 @@ impl Storage {
}),
#[cfg(feature = "storage-oss")]
"oss" => {
+ if props
+ .get("fs.oss.impl")
+ .is_some_and(|v| v.eq_ignore_ascii_case("cpp"))
+ {
+ #[cfg(feature = "storage-oss-cpp")]
+ return Ok(Self::OssCpp {
+ config: Box::new(super::oss_cpp_config_parse(props)?),
+ operators: Mutex::new(HashMap::new()),
+ });
+ #[cfg(not(feature = "storage-oss-cpp"))]
+ return Err(error::Error::IoUnsupported {
+ message: "OSS C++ SDK requires the storage-oss-cpp
feature".to_string(),
+ });
+ }
#[cfg(feature = "storage-jindo")]
if super::use_jindo(&props)? {
let config = super::jindo_config_parse(props)?;
@@ -260,6 +281,15 @@ impl Storage {
let op = Self::cached_oss_operator(config, operators, path,
&bucket)?;
Ok((op, Cow::Borrowed(relative_path)))
}
+ #[cfg(feature = "storage-oss-cpp")]
+ Storage::OssCpp { config, operators } => {
+ let (bucket, relative_path) =
+ Self::bucket_and_relative_path(path, "OSS C++", &["oss"])?;
+ let op = Self::cached_operator(operators, "OSS C++", &bucket,
|| {
+ super::oss_cpp_config_build(config, &bucket)
+ })?;
+ Ok((op, Cow::Borrowed(relative_path)))
+ }
#[cfg(feature = "storage-jindo")]
Storage::Jindo { config, operators } => {
let (bucket, relative_path) =
@@ -693,6 +723,16 @@ mod fs_relative_path_tests {
mod tests {
use super::*;
+ #[cfg(all(feature = "storage-oss", not(feature = "storage-oss-cpp")))]
+ #[test]
+ fn oss_cpp_without_feature_does_not_fall_back() {
+ let result =
Storage::build(FileIOBuilder::new("oss").with_props(HashMap::from([(
+ "fs.oss.impl".to_string(),
+ "cpp".to_string(),
+ )])));
+ assert!(matches!(result, Err(error::Error::IoUnsupported { .. })));
+ }
+
fn memory_operator() -> Operator {
Operator::from_config(opendal::services::MemoryConfig::default()).unwrap()
}
diff --git a/crates/paimon/src/io/storage_oss_cpp.rs
b/crates/paimon/src/io/storage_oss_cpp.rs
new file mode 100644
index 00000000..c0ecb8dd
--- /dev/null
+++ b/crates/paimon/src/io/storage_oss_cpp.rs
@@ -0,0 +1,780 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::ffi::{c_char, c_void, CStr, CString};
+use std::fmt::{Debug, Formatter};
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+
+use libloading::Library;
+use opendal::raw::*;
+use opendal::{Buffer, Builder, BytesRange, Capability, EntryMode, Error,
ErrorKind, Metadata};
+use opendal::{OperationContext, Operator, Result};
+use tokio::sync::{OnceCell, Semaphore};
+
+const PREFIX: &str = "fs.oss.cpp.";
+const DEFAULT_CONCURRENCY: usize = 8;
+
+#[derive(Clone)]
+pub struct OssCppStorageConfig {
+ library: String,
+ strings: Vec<CString>,
+ concurrency: usize,
+ gate: Arc<Semaphore>,
+ connect_timeout: i64,
+ request_timeout: i64,
+ retry_attempts: i64,
+ path_style: u8,
+}
+
+impl Default for OssCppStorageConfig {
+ fn default() -> Self {
+ Self {
+ library: String::new(),
+ strings: Vec::new(),
+ concurrency: 0,
+ gate: Arc::new(Semaphore::new(0)),
+ connect_timeout: 0,
+ request_timeout: 0,
+ retry_attempts: 0,
+ path_style: 0,
+ }
+ }
+}
+
+impl Debug for OssCppStorageConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("OssCppStorageConfig")
+ .field("library", &self.library)
+ .field("concurrency", &self.concurrency)
+ .finish_non_exhaustive()
+ }
+}
+
+pub(crate) fn oss_cpp_config_parse(
+ props: HashMap<String, String>,
+) -> crate::Result<OssCppStorageConfig> {
+ let parse = || -> Result<OssCppStorageConfig> {
+ let required = |key: &str| -> Result<String> {
+ props
+ .get(key)
+ .filter(|s| !s.is_empty())
+ .cloned()
+ .ok_or_else(|| Error::new(ErrorKind::ConfigInvalid,
format!("Missing {key}")))
+ };
+ let number = |name: &str, default: i64| -> Result<i64> {
+ let key = format!("{PREFIX}{name}");
+ match props.get(&key) {
+ None => Ok(default),
+ Some(value) => value.parse::<i64>().ok().filter(|n| *n >
0).ok_or_else(|| {
+ Error::new(ErrorKind::ConfigInvalid, format!("{key} must
be positive"))
+ }),
+ }
+ };
+ let endpoint = required("fs.oss.endpoint")?;
+ let endpoint = if endpoint.contains("://") {
+ endpoint
+ } else {
+ format!("https://{endpoint}";)
+ };
+ let url = url::Url::parse(&endpoint)
+ .map_err(|_| Error::new(ErrorKind::ConfigInvalid, "Invalid OSS
endpoint"))?;
+ if !matches!(url.scheme(), "http" | "https")
+ || url.host_str().is_none()
+ || !url.username().is_empty()
+ || url.password().is_some()
+ || url.query().is_some()
+ || url.fragment().is_some()
+ || url.path() != "/"
+ {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "OSS endpoint must be an HTTP(S) origin",
+ ));
+ }
+ let strings = [
+ endpoint,
+ required("fs.oss.region")?,
+ required("fs.oss.accessKeyId")?,
+ required("fs.oss.accessKeySecret")?,
+ props
+ .get("fs.oss.securityToken")
+ .cloned()
+ .unwrap_or_default(),
+ // The native SDK appends this to its own User-Agent.
+ format!("{} oss-cpp", super::user_agent::oss_user_agent(&props)),
+ ]
+ .into_iter()
+ .map(|s| cstring(&s))
+ .collect::<Result<Vec<_>>>()?;
+ let concurrency = number("max.concurrent.requests",
DEFAULT_CONCURRENCY as i64)? as usize;
+ if concurrency > Semaphore::MAX_PERMITS {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "OSS C++ concurrency exceeds semaphore limit",
+ ));
+ }
+ let path_style = match
props.get("fs.oss.cpp.path-style").map(String::as_str) {
+ None | Some("false") => 0,
+ Some("true") => 1,
+ _ => {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "Invalid OSS C++ path-style",
+ ))
+ }
+ };
+ Ok(OssCppStorageConfig {
+ library: required("fs.oss.cpp.library.path")?,
+ strings,
+ concurrency,
+ gate: Arc::new(Semaphore::new(concurrency)),
+ connect_timeout: number("connect.timeout-ms", 10_000)?,
+ request_timeout: number("request.timeout-ms", 30_000)?,
+ retry_attempts: number("retry.max-attempts", 3)?,
+ path_style,
+ })
+ };
+ parse().map_err(|e| crate::Error::ConfigInvalid {
+ message: e.to_string(),
+ })
+}
+
+pub(crate) fn oss_cpp_config_build(
+ config: &OssCppStorageConfig,
+ bucket: &str,
+) -> crate::Result<Operator> {
+ Operator::new(CppBuilder {
+ config: config.clone(),
+ bucket: bucket.to_string(),
+ })
+ .map_err(|e| crate::Error::IoUnexpected {
+ message: "Cannot build OSS C++ operator".to_string(),
+ source: Box::new(e),
+ })
+}
+
+#[derive(Default)]
+struct CppBuilder {
+ config: OssCppStorageConfig,
+ bucket: String,
+}
+
+impl Builder for CppBuilder {
+ type Config = ();
+ fn build(self) -> Result<impl Service> {
+ if self.config.strings.len() != 6 || self.config.concurrency == 0 {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "Missing OSS C++ configuration",
+ ));
+ }
+ Ok(CppService {
+ info: ServiceInfo::new("oss-cpp", "/", &self.bucket),
+ client: Arc::new(LazyClient {
+ gate: self.config.gate.clone(),
+ config: self.config,
+ bucket: self.bucket,
+ client: OnceCell::new(),
+ }),
+ })
+ }
+}
+
+struct CppService {
+ info: ServiceInfo,
+ client: Arc<LazyClient>,
+}
+
+impl Debug for CppService {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("OssCppService")
+ .field("info", &self.info)
+ .finish()
+ }
+}
+
+impl Service for CppService {
+ type Reader = CppReader;
+ type Writer = ();
+ type Lister = CppLister;
+ type Deleter = ();
+ type Copier = ();
+
+ fn info(&self) -> ServiceInfo {
+ self.info.clone()
+ }
+ fn capability(&self) -> Capability {
+ Capability {
+ stat: true,
+ read: true,
+ read_with_suffix: true,
+ list: true,
+ list_with_recursive: true,
+ shared: true,
+ ..Default::default()
+ }
+ }
+ async fn stat(&self, _: &OperationContext, path: &str, _: OpStat) ->
Result<RpStat> {
+ let path = path.to_string();
+ let metadata = self.client.call(move |c| c.stat(&path)).await?;
+ Ok(RpStat::new(metadata))
+ }
+ fn read(&self, _: &OperationContext, path: &str, _: OpRead) ->
Result<Self::Reader> {
+ Ok(CppReader {
+ client: self.client.clone(),
+ path: path.to_string(),
+ })
+ }
+ fn list(&self, _: &OperationContext, path: &str, args: OpList) ->
Result<Self::Lister> {
+ Ok(CppLister {
+ client: self.client.clone(),
+ path: path.to_string(),
+ recursive: args.recursive(),
+ max_keys: args.limit().unwrap_or(1000).clamp(1, 1000) as u32,
+ token: String::new(),
+ done: false,
+ entries: Vec::new().into_iter(),
+ })
+ }
+ async fn create_dir(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: OpCreateDir,
+ ) -> Result<RpCreateDir> {
+ unsupported()
+ }
+ fn write(&self, _: &OperationContext, _: &str, _: OpWrite) ->
Result<Self::Writer> {
+ unsupported()
+ }
+ fn delete(&self, _: &OperationContext) -> Result<Self::Deleter> {
+ unsupported()
+ }
+ fn copy(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpCopy,
+ _: OpCopier,
+ ) -> Result<Self::Copier> {
+ unsupported()
+ }
+ async fn rename(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpRename,
+ ) -> Result<RpRename> {
+ unsupported()
+ }
+ async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) ->
Result<RpPresign> {
+ unsupported()
+ }
+}
+
+fn unsupported<T>() -> Result<T> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "OSS C++ backend is read-only",
+ ))
+}
+
+struct CppReader {
+ client: Arc<LazyClient>,
+ path: String,
+}
+
+impl oio::Read for CppReader {
+ async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn
oio::ReadStreamDyn>)> {
+ let (_, buffer) = self.read(range).await?;
+ Ok((RpRead::default(), Box::new(buffer)))
+ }
+ async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
+ let path = self.path.clone();
+ let data = self.client.call(move |c| c.read(&path, range)).await?;
+ Ok((RpRead::default(), Buffer::from(data)))
+ }
+}
+
+struct CppLister {
+ client: Arc<LazyClient>,
+ path: String,
+ recursive: bool,
+ max_keys: u32,
+ token: String,
+ done: bool,
+ entries: std::vec::IntoIter<oio::Entry>,
+}
+
+impl oio::List for CppLister {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ loop {
+ if let Some(entry) = self.entries.next() {
+ return Ok(Some(entry));
+ }
+ if self.done {
+ return Ok(None);
+ }
+ let path = self.path.clone();
+ let token = self.token.clone();
+ let recursive = self.recursive;
+ let max_keys = self.max_keys;
+ let (entries, next) = self
+ .client
+ .call(move |c| c.list(&path, &token, recursive, max_keys))
+ .await?;
+ self.done = next.is_empty();
+ self.token = next;
+ self.entries = entries.into_iter();
+ }
+ }
+}
+
+struct LazyClient {
+ config: OssCppStorageConfig,
+ bucket: String,
+ client: OnceCell<Arc<Client>>,
+ gate: Arc<Semaphore>,
+}
+
+// Native clients/connection pools cannot be inherited across fork. Check
before
+// touching OnceCell or Semaphore: either may have been locked by a parent
thread.
+fn ensure_process() -> Result<()> {
+ static PID: AtomicU32 = AtomicU32::new(0);
+ let current = std::process::id();
+ match PID.compare_exchange(0, current, Ordering::AcqRel,
Ordering::Acquire) {
+ Ok(_) => Ok(()),
+ Err(pid) if pid == current => Ok(()),
+ Err(_) => Err(Error::new(
+ ErrorKind::Unsupported,
+ crate::error::OSS_CPP_FORK_ERROR,
+ )),
+ }
+}
+
+impl LazyClient {
+ async fn call<T: Send + 'static>(
+ &self,
+ operation: impl FnOnce(&Client) -> Result<T> + Send + 'static,
+ ) -> Result<T> {
+ ensure_process()?;
+ let client = self
+ .client
+ .get_or_try_init(|| async {
+ let config = self.config.clone();
+ let bucket = self.bucket.clone();
+ blocking(move || Client::new(&config,
&bucket).map(Arc::new)).await
+ })
+ .await?
+ .clone();
+ let permit = self
+ .gate
+ .clone()
+ .acquire_owned()
+ .await
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "OSS C++ request
gate closed"))?;
+ blocking(move || {
+ // Cancellation does not stop a synchronous C++ request. Keep both
+ // the client and permit alive until the actual native call
returns.
+ let _permit = permit;
+ operation(&client)
+ })
+ .await
+ }
+}
+
+async fn blocking<T: Send + 'static>(f: impl FnOnce() -> Result<T> + Send +
'static) -> Result<T> {
+ tokio::task::spawn_blocking(f)
+ .await
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "OSS C++ blocking task
failed"))?
+}
+
+#[repr(C)]
+struct NativeConfig {
+ endpoint: *const c_char,
+ region: *const c_char,
+ access_key: *const c_char,
+ secret_key: *const c_char,
+ token: *const c_char,
+ user_agent: *const c_char,
+ connect_timeout_ms: i64,
+ request_timeout_ms: i64,
+ retry_attempts: i64,
+ path_style: u8,
+}
+
+#[repr(C)]
+struct NativeError {
+ status: i32,
+ code: [c_char; 128],
+ request_id: [c_char; 256],
+}
+impl Default for NativeError {
+ fn default() -> Self {
+ Self {
+ status: 0,
+ code: [0; 128],
+ request_id: [0; 256],
+ }
+ }
+}
+impl NativeError {
+ fn check(&self, status: i32) -> Result<()> {
+ if status == 0 {
+ return Ok(());
+ }
+ let kind = match self.status {
+ 404 => ErrorKind::NotFound,
+ 401 | 403 => ErrorKind::PermissionDenied,
+ 429 | 503 => ErrorKind::RateLimited,
+ 416 => ErrorKind::RangeNotSatisfied,
+ _ => ErrorKind::Unexpected,
+ };
+ // Only expose status, symbolic code and request ID, never request
URLs,
+ // authorization headers or SDK exception text.
+ Err(Error::new(kind, "OSS C++ operation failed")
+ .with_context("status", self.status.to_string())
+ .with_context("code", native_string(self.code.as_ptr()))
+ .with_context("request_id",
native_string(self.request_id.as_ptr())))
+ }
+}
+
+#[repr(C)]
+struct NativeMetadata {
+ size: i64,
+ modified: [c_char; 64],
+}
+type EntryCallback =
+ unsafe extern "C" fn(*mut c_void, *const c_char, usize, i64, *const
c_char, u8);
+type Create = unsafe extern "C" fn(*const NativeConfig, *mut NativeError) ->
*mut c_void;
+type Destroy = unsafe extern "C" fn(*mut c_void);
+type Stat = unsafe extern "C" fn(
+ *mut c_void,
+ *const c_char,
+ *const c_char,
+ *mut NativeMetadata,
+ *mut NativeError,
+) -> i32;
+type Read = unsafe extern "C" fn(
+ *mut c_void,
+ *const c_char,
+ *const c_char,
+ u64,
+ usize,
+ *mut u8,
+ *mut NativeError,
+) -> i32;
+type List = unsafe extern "C" fn(
+ *mut c_void,
+ *const c_char,
+ *const c_char,
+ *const c_char,
+ u8,
+ u32,
+ EntryCallback,
+ *mut c_void,
+ *mut *mut c_char,
+ *mut NativeError,
+) -> i32;
+type FreeString = unsafe extern "C" fn(*mut c_char);
+
+struct Api {
+ create: Create,
+ destroy: Destroy,
+ stat: Stat,
+ read: Read,
+ list: List,
+ free_string: FreeString,
+ _library: Library,
+}
+impl Api {
+ fn load(path: &str) -> Result<Self> {
+ // SAFETY: loading native code is explicit opt-in via a trusted library
+ // path. Symbols follow bridge.h; retain the library for their
lifetime.
+ unsafe {
+ let library = Library::new(path).map_err(|_| {
+ Error::new(
+ ErrorKind::ConfigInvalid,
+ "Cannot load fs.oss.cpp.library.path or its dependencies",
+ )
+ })?;
+ let load_error = || Error::new(ErrorKind::ConfigInvalid, "Invalid
OSS C++ bridge ABI");
+ let version = library
+ .get::<unsafe extern "C" fn() ->
u32>(b"oss_cpp_bridge_abi_version\0")
+ .map_err(|_| load_error())?;
+ if version() != 2 {
+ return Err(load_error());
+ }
+ Ok(Self {
+ create: *library
+ .get(b"oss_cpp_bridge_create\0")
+ .map_err(|_| load_error())?,
+ destroy: *library
+ .get(b"oss_cpp_bridge_destroy\0")
+ .map_err(|_| load_error())?,
+ stat: *library
+ .get(b"oss_cpp_bridge_stat\0")
+ .map_err(|_| load_error())?,
+ read: *library
+ .get(b"oss_cpp_bridge_read\0")
+ .map_err(|_| load_error())?,
+ list: *library
+ .get(b"oss_cpp_bridge_list\0")
+ .map_err(|_| load_error())?,
+ free_string: *library
+ .get(b"oss_cpp_bridge_free_string\0")
+ .map_err(|_| load_error())?,
+ _library: library,
+ })
+ }
+ }
+}
+
+struct Client {
+ api: Arc<Api>,
+ pointer: *mut c_void,
+ bucket: CString,
+ pid: u32,
+}
+// SAFETY: the bridge uses a concurrent OSSClient and per-call outputs. The
+// immutable client is released only after all blocking calls drop their Arc.
+unsafe impl Send for Client {}
+unsafe impl Sync for Client {}
+
+impl Client {
+ fn new(config: &OssCppStorageConfig, bucket: &str) -> Result<Self> {
+ let bucket = cstring(bucket)?;
+ let api = Arc::new(Api::load(&config.library)?);
+ let s = &config.strings;
+ let input = NativeConfig {
+ endpoint: s[0].as_ptr(),
+ region: s[1].as_ptr(),
+ access_key: s[2].as_ptr(),
+ secret_key: s[3].as_ptr(),
+ token: s[4].as_ptr(),
+ user_agent: s[5].as_ptr(),
+ connect_timeout_ms: config.connect_timeout,
+ request_timeout_ms: config.request_timeout,
+ retry_attempts: config.retry_attempts,
+ path_style: config.path_style,
+ };
+ let mut error = NativeError::default();
+ let pointer = unsafe { (api.create)(&input, &mut error) };
+ if pointer.is_null() {
+ error.check(-1)?;
+ unreachable!();
+ }
+ Ok(Self {
+ api,
+ pointer,
+ bucket,
+ pid: std::process::id(),
+ })
+ }
+ fn stat(&self, key: &str) -> Result<Metadata> {
+ if key.is_empty() {
+ return Ok(Metadata::new(EntryMode::DIR));
+ }
+ if key.ends_with('/') {
+ let (entries, _) = self.list(key, "", true, 1)?;
+ return if entries.is_empty() {
+ Err(Error::new(ErrorKind::NotFound, "OSS directory not found"))
+ } else {
+ Ok(Metadata::new(EntryMode::DIR))
+ };
+ }
+ let key = cstring(key)?;
+ let mut out = NativeMetadata {
+ size: 0,
+ modified: [0; 64],
+ };
+ let mut error = NativeError::default();
+ let status = unsafe {
+ (self.api.stat)(
+ self.pointer,
+ self.bucket.as_ptr(),
+ key.as_ptr(),
+ &mut out,
+ &mut error,
+ )
+ };
+ error.check(status)?;
+ metadata(out.size, &native_string(out.modified.as_ptr()), false)
+ }
+ fn read(&self, key: &str, range: BytesRange) -> Result<Vec<u8>> {
+ let (offset, length) = match range {
+ BytesRange::Range {
+ offset,
+ size: Some(length),
+ } => (offset, length),
+ BytesRange::Range { offset, size: None } => {
+ let size = self.stat(key)?.content_length();
+ let length = size.checked_sub(offset).ok_or_else(|| {
+ Error::new(ErrorKind::RangeNotSatisfied, "Range starts
after EOF")
+ })?;
+ (offset, length)
+ }
+ BytesRange::Suffix { size } => {
+ let length = self.stat(key)?.content_length();
+ (length.saturating_sub(size), size.min(length))
+ }
+ };
+ offset
+ .checked_add(length)
+ .ok_or_else(|| Error::new(ErrorKind::RangeNotSatisfied, "Range
overflow"))?;
+ let length = usize::try_from(length)
+ .map_err(|_| Error::new(ErrorKind::RangeNotSatisfied, "Range too
large"))?;
+ if length == 0 {
+ return Ok(Vec::new());
+ }
+ if length > i64::MAX as usize {
+ return Err(Error::new(ErrorKind::RangeNotSatisfied, "Range too
large"));
+ }
+ let key = cstring(key)?;
+ let mut data = Vec::new();
+ data.try_reserve_exact(length)
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "Cannot allocate
read buffer"))?;
+ data.resize(length, 0);
+ let mut error = NativeError::default();
+ let status = unsafe {
+ (self.api.read)(
+ self.pointer,
+ self.bucket.as_ptr(),
+ key.as_ptr(),
+ offset,
+ length,
+ data.as_mut_ptr(),
+ &mut error,
+ )
+ };
+ error.check(status)?;
+ Ok(data)
+ }
+ fn list(
+ &self,
+ prefix: &str,
+ token: &str,
+ recursive: bool,
+ max_keys: u32,
+ ) -> Result<(Vec<oio::Entry>, String)> {
+ let prefix = cstring(prefix)?;
+ let token = cstring(token)?;
+ let mut entries = EntryCollector {
+ entries: Vec::new(),
+ error: None,
+ };
+ let mut next = std::ptr::null_mut();
+ let mut error = NativeError::default();
+ let status = unsafe {
+ (self.api.list)(
+ self.pointer,
+ self.bucket.as_ptr(),
+ prefix.as_ptr(),
+ token.as_ptr(),
+ u8::from(recursive),
+ max_keys,
+ collect_entry,
+ &mut entries as *mut EntryCollector as *mut c_void,
+ &mut next,
+ &mut error,
+ )
+ };
+ let next_token = if next.is_null() {
+ String::new()
+ } else {
+ let value = native_string(next);
+ unsafe { (self.api.free_string)(next) };
+ value
+ };
+ error.check(status)?;
+ if let Some(error) = entries.error {
+ return Err(error);
+ }
+ Ok((entries.entries, next_token))
+ }
+}
+impl Drop for Client {
+ fn drop(&mut self) {
+ if self.pid != std::process::id() {
+ // Do not destroy native handles or unload their library in a
forked child.
+ std::mem::forget(self.api.clone());
+ return;
+ }
+ unsafe { (self.api.destroy)(self.pointer) };
+ }
+}
+
+struct EntryCollector {
+ entries: Vec<oio::Entry>,
+ error: Option<Error>,
+}
+unsafe extern "C" fn collect_entry(
+ ctx: *mut c_void,
+ key: *const c_char,
+ len: usize,
+ size: i64,
+ modified: *const c_char,
+ dir: u8,
+) {
+ // SAFETY: the synchronous bridge only invokes this callback while ctx and
+ // the borrowed native strings are alive; no callback is retained.
+ let collector = unsafe { &mut *(ctx as *mut EntryCollector) };
+ if collector.error.is_some() {
+ return;
+ }
+ let bytes = unsafe { std::slice::from_raw_parts(key.cast::<u8>(), len) };
+ let result = std::str::from_utf8(bytes)
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "Invalid UTF-8 object
key"))
+ .and_then(|path| {
+ metadata(size, &native_string(modified), dir != 0)
+ .map(|meta| oio::Entry::new(path, meta))
+ });
+ match result {
+ Ok(entry) => collector.entries.push(entry),
+ Err(error) => collector.error = Some(error),
+ }
+}
+fn metadata(size: i64, modified: &str, dir: bool) -> Result<Metadata> {
+ let size = u64::try_from(size)
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "Negative object
size"))?;
+ let mut meta = Metadata::new(if dir { EntryMode::DIR } else {
EntryMode::FILE });
+ meta.set_content_length(size);
+ if !modified.is_empty() {
+ let time = chrono::DateTime::parse_from_rfc2822(modified)
+ .or_else(|_| chrono::DateTime::parse_from_rfc3339(modified))
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "Invalid object
modification time"))?;
+
meta.set_last_modified(Timestamp::from_millisecond(time.timestamp_millis())?);
+ }
+ Ok(meta)
+}
+fn cstring(value: &str) -> Result<CString> {
+ CString::new(value).map_err(|_| Error::new(ErrorKind::ConfigInvalid, "NUL
in OSS C++ input"))
+}
+fn native_string(value: *const c_char) -> String {
+ // All native outputs are NUL terminated by bridge.cc.
+ unsafe { CStr::from_ptr(value) }
+ .to_string_lossy()
+ .into_owned()
+}
+
+#[cfg(test)]
+#[path = "storage_oss_cpp_test.rs"]
+mod tests;
diff --git a/crates/paimon/src/io/storage_oss_cpp_test.rs
b/crates/paimon/src/io/storage_oss_cpp_test.rs
new file mode 100644
index 00000000..21085258
--- /dev/null
+++ b/crates/paimon/src/io/storage_oss_cpp_test.rs
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::*;
+use crate::io::FileRead;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Mutex;
+use std::time::Duration;
+
+use axum::body::Body;
+use axum::extract::{Query, State};
+use axum::http::{HeaderMap, Method, Response, StatusCode, Uri};
+use axum::Router;
+
+fn props() -> HashMap<String, String> {
+ [
+ ("fs.oss.impl", "cpp"),
+ ("fs.oss.cpp.library.path", "/not/a/real/library.so"),
+ ("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"),
+ ("fs.oss.region", "cn-hangzhou"),
+ ("fs.oss.accessKeyId", "test-ak"),
+ ("fs.oss.accessKeySecret", "test-secret"),
+ ("fs.oss.securityToken", "test-sts"),
+ ]
+ .into_iter()
+ .map(|(k, v)| (k.to_string(), v.to_string()))
+ .collect()
+}
+
+#[test]
+fn config_defaults_validation_and_redaction() {
+ let config = oss_cpp_config_parse(props()).unwrap();
+ assert_eq!(config.concurrency, 8);
+ assert_eq!(config.retry_attempts, 3);
+ assert_eq!(
+ config.strings[0].to_str().unwrap(),
+ "https://oss-cn-hangzhou.aliyuncs.com";
+ );
+ let debug = format!("{config:?}");
+ assert!(!debug.contains("test-secret"));
+ assert!(!debug.contains("test-sts"));
+ for (key, value) in [
+ ("fs.oss.cpp.max.concurrent.requests", "0"),
+ ("fs.oss.cpp.max.concurrent.requests", "-1"),
+ ("fs.oss.cpp.request.timeout-ms", "bad"),
+ ("fs.oss.cpp.retry.max-attempts", "0"),
+ ("fs.oss.cpp.path-style", "yes"),
+ ("fs.oss.endpoint", "ftp://host";),
+ ("fs.oss.endpoint", "https://secret@host";),
+ ("fs.oss.endpoint", "https://host/path";),
+ ("fs.oss.accessKeySecret", "nul\0secret"),
+ ] {
+ let mut options = props();
+ options.insert(key.to_string(), value.to_string());
+ assert!(oss_cpp_config_parse(options).is_err(), "{key}");
+ }
+ for key in [
+ "fs.oss.region",
+ "fs.oss.endpoint",
+ "fs.oss.cpp.library.path",
+ ] {
+ let mut options = props();
+ options.remove(key);
+ assert!(oss_cpp_config_parse(options).is_err(), "{key}");
+ }
+}
+
+#[tokio::test]
+async fn selection_is_explicit_and_missing_library_fails() {
+ let file_io = crate::io::FileIOBuilder::new("oss")
+ .with_props(props())
+ .build()
+ .unwrap();
+ assert!(file_io.get_status("oss://bucket/object").await.is_err());
+ let config = oss_cpp_config_parse(props()).unwrap();
+ let op = oss_cpp_config_build(&config, "bucket").unwrap();
+ assert_eq!(op.info().scheme(), "oss-cpp");
+ assert_eq!(
+ op.write("object", "data").await.unwrap_err().kind(),
+ ErrorKind::Unsupported
+ );
+ assert_eq!(
+ op.delete("object").await.unwrap_err().kind(),
+ ErrorKind::Unsupported
+ );
+}
+
+#[test]
+fn malformed_metadata_and_error_mapping() {
+ assert!(metadata(-1, "", false).is_err());
+ assert!(metadata(1, "broken", false).is_err());
+ assert!(metadata(1, "Wed, 01 Jan 2025 00:00:00 GMT", false)
+ .unwrap()
+ .last_modified()
+ .is_some());
+ assert!(metadata(1, "2025-01-01T00:00:00Z", false)
+ .unwrap()
+ .last_modified()
+ .is_some());
+ for (status, kind) in [
+ (404, ErrorKind::NotFound),
+ (403, ErrorKind::PermissionDenied),
+ (503, ErrorKind::RateLimited),
+ (429, ErrorKind::RateLimited),
+ (416, ErrorKind::RangeNotSatisfied),
+ ] {
+ let error = NativeError {
+ status,
+ ..Default::default()
+ };
+ assert_eq!(error.check(-1).unwrap_err().kind(), kind);
+ }
+}
+
+#[derive(Clone, Default)]
+struct Mock {
+ retries: Arc<AtomicUsize>,
+ partial_attempts: Arc<AtomicUsize>,
+ active: Arc<AtomicUsize>,
+ peak: Arc<AtomicUsize>,
+ heads: Arc<AtomicUsize>,
+ tokens: Arc<Mutex<Vec<String>>>,
+ ranges: Arc<Mutex<Vec<String>>>,
+ list_requests: Arc<Mutex<Vec<(String, String)>>>,
+ user_agents: Arc<Mutex<Vec<String>>>,
+}
+
+async fn mock(
+ State(state): State<Mock>,
+ method: Method,
+ uri: Uri,
+ Query(query): Query<HashMap<String, String>>,
+ headers: HeaderMap,
+) -> Response<Body> {
+ if let Some(token) = headers.get("x-oss-security-token") {
+ state
+ .tokens
+ .lock()
+ .unwrap()
+ .push(token.to_str().unwrap().to_string());
+ }
+ if let Some(ua) = headers.get("user-agent") {
+ state
+ .user_agents
+ .lock()
+ .unwrap()
+ .push(ua.to_str().unwrap().to_string());
+ }
+ let path = uri.path();
+ let status = if path.ends_with("missing") {
+ 404
+ } else if path.ends_with("denied") {
+ 403
+ } else if path.ends_with("throttle")
+ || (path.ends_with("retry") && state.retries.fetch_add(1,
Ordering::SeqCst) == 0)
+ {
+ 503
+ } else {
+ 200
+ };
+ if status != 200 {
+ let code = match status {
+ 404 => "NoSuchKey",
+ 403 => "AccessDenied",
+ _ => "SlowDown",
+ };
+ return Response::builder().status(status).header("x-oss-request-id",
"test-request")
+
.body(Body::from(format!("<Error><Code>{code}</Code><Message>test</Message><RequestId>test-request</RequestId></Error>"))).unwrap();
+ }
+ if query.contains_key("list-type") {
+ let prefix = query.get("prefix").map(String::as_str).unwrap_or("");
+ state.list_requests.lock().unwrap().push((
+ prefix.to_string(),
+ query.get("max-keys").cloned().unwrap_or_default(),
+ ));
+ if prefix == "denied/" {
+ return Response::builder()
+ .status(StatusCode::FORBIDDEN)
+ .body(Body::from(
+
"<Error><Code>AccessDenied</Code><Message>test</Message></Error>",
+ ))
+ .unwrap();
+ }
+ if prefix == "missing/" {
+ return Response::new(Body::from(
+
"<ListBucketResult><Name>bucket</Name><IsTruncated>false</IsTruncated></ListBucketResult>",
+ ));
+ }
+ let page = if prefix == "bad/" {
+ "<IsTruncated>true</IsTruncated>".to_string()
+ } else if query.contains_key("continuation-token") {
+ assert_eq!(query["continuation-token"], "page2");
+
"<IsTruncated>false</IsTruncated><Contents><Key>objects/b</Key><Size>10</Size><LastModified>2025-01-01T00:00:00Z</LastModified></Contents>".to_string()
+ } else {
+ let common = if query.contains_key("delimiter") {
+
"<CommonPrefixes><Prefix>objects/sub/</Prefix></CommonPrefixes>"
+ } else {
+ ""
+ };
+
format!("<IsTruncated>true</IsTruncated><NextContinuationToken>page2</NextContinuationToken><Contents><Key>objects/a</Key><Size>10</Size><LastModified>2025-01-01T00:00:00Z</LastModified></Contents>{common}")
+ };
+ return Response::new(Body::from(format!(
+ "<ListBucketResult><Name>bucket</Name>{page}</ListBucketResult>"
+ )));
+ }
+ if method == Method::HEAD {
+ state.heads.fetch_add(1, Ordering::SeqCst);
+ return Response::builder()
+ .header("content-length", "10")
+ .header("last-modified", "Wed, 01 Jan 2025 00:00:00 GMT")
+ .body(Body::empty())
+ .unwrap();
+ }
+ if path.ends_with("slow") {
+ let active = state.active.fetch_add(1, Ordering::SeqCst) + 1;
+ state.peak.fetch_max(active, Ordering::SeqCst);
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ state.active.fetch_sub(1, Ordering::SeqCst);
+ }
+ let range = headers.get("range").unwrap().to_str().unwrap().to_string();
+ if path.ends_with("partial") && state.partial_attempts.fetch_add(1,
Ordering::SeqCst) == 0 {
+ let chunks = futures::stream::iter([
+ Ok(bytes::Bytes::from_static(b"xx")),
+ Err(std::io::Error::new(
+ std::io::ErrorKind::UnexpectedEof,
+ "test body disconnect",
+ )),
+ ]);
+ return Response::builder()
+ .status(206)
+ .header("content-range", "bytes 2-5/10")
+ .header("content-length", "4")
+ .body(Body::from_stream(chunks))
+ .unwrap();
+ }
+ state.ranges.lock().unwrap().push(range.clone());
+ let (start, end) =
range.trim_start_matches("bytes=").split_once('-').unwrap();
+ let start: usize = start.parse().unwrap();
+ let end: usize = end.parse().unwrap();
+ if end >= 10 || start > end {
+ return Response::builder().status(416).body(Body::empty()).unwrap();
+ }
+ let content_range = if path.ends_with("wrong") {
+ "bytes 0-3/10".to_string()
+ } else {
+ format!("bytes {start}-{end}/10")
+ };
+ let body = if path.ends_with("short") {
+ "x".as_bytes()
+ } else {
+ &b"0123456789"[start..=end]
+ };
+ Response::builder()
+ .status(if path.ends_with("ignored") {
+ StatusCode::OK
+ } else {
+ StatusCode::PARTIAL_CONTENT
+ })
+ .header("content-range", content_range)
+ .body(Body::from(body))
+ .unwrap()
+}
+
+async fn fixture(
+ concurrency: usize,
+) -> (
+ Operator,
+ Mock,
+ tokio::task::JoinHandle<()>,
+ HashMap<String, String>,
+) {
+ let library = std::env::var("OSS_CPP_BRIDGE_LIBRARY")
+ .expect("Set OSS_CPP_BRIDGE_LIBRARY to the real compiled bridge");
+ let state = Mock::default();
+ let app = Router::new().fallback(mock).with_state(state.clone());
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let address = listener.local_addr().unwrap();
+ let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
+ let mut options = props();
+ options.insert("fs.oss.endpoint".into(), format!("http://{address}";));
+ options.insert("fs.oss.cpp.library.path".into(), library);
+ options.insert("fs.oss.cpp.path-style".into(), "true".into());
+ options.insert(
+ "fs.oss.cpp.max.concurrent.requests".into(),
+ concurrency.to_string(),
+ );
+ options.insert("fs.oss.cpp.retry.max-attempts".into(), "2".into());
+ let config = oss_cpp_config_parse(options.clone()).unwrap();
+ (
+ oss_cpp_config_build(&config, "bucket").unwrap(),
+ state,
+ server,
+ options,
+ )
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "requires the compiled OSS C++ SDK bridge; no cloud credentials"]
+async fn real_sdk_reads_errors_pagination_and_credentials() {
+ let (op, state, server, mut options) = fixture(2).await;
+ assert!(op.stat("").await.unwrap().is_dir());
+ assert!(op.stat("objects/").await.unwrap().is_dir());
+ assert!(!op.exists("missing/").await.unwrap());
+ assert_eq!(
+ op.stat("denied/").await.unwrap_err().kind(),
+ ErrorKind::PermissionDenied
+ );
+ let file_io = crate::io::FileIOBuilder::new("oss")
+ .with_props(options.clone())
+ .build()
+ .unwrap();
+ assert!(file_io.exists_dir("oss://bucket/objects").await.unwrap());
+ assert!(!file_io.exists_dir("oss://bucket/missing").await.unwrap());
+ assert!(file_io.exists_dir("oss://bucket/denied").await.is_err());
+ {
+ let list_requests = state.list_requests.lock().unwrap();
+ assert!(
+ list_requests.contains(&("objects/".to_string(), "1".to_string())),
+ "list requests: {list_requests:?}"
+ );
+ }
+ assert_eq!(op.stat("data").await.unwrap().content_length(), 10);
+ let before = state.heads.load(Ordering::SeqCst);
+ let reader = op.reader("data").await.unwrap();
+ assert_eq!(reader.read(2..6).await.unwrap().to_bytes(), &b"2345"[..]);
+ assert_eq!(
+ state.heads.load(Ordering::SeqCst),
+ before,
+ "bounded range must not HEAD"
+ );
+ assert_eq!(
+ op.read("data").await.unwrap().to_bytes(),
+ &b"0123456789"[..]
+ );
+ assert_eq!(reader.read(0..0).await.unwrap().len(), 0);
+ for key in ["short", "wrong", "ignored"] {
+ assert!(
+ op.reader(key).await.unwrap().read(2..6).await.is_err(),
+ "{key}"
+ );
+ }
+ assert!(!op.exists("missing").await.unwrap());
+ assert_eq!(
+ op.stat("denied").await.unwrap_err().kind(),
+ ErrorKind::PermissionDenied
+ );
+ assert_eq!(
+ op.reader("retry")
+ .await
+ .unwrap()
+ .read(2..6)
+ .await
+ .unwrap()
+ .to_bytes(),
+ &b"2345"[..]
+ );
+ assert_eq!(state.retries.load(Ordering::SeqCst), 2);
+ assert_eq!(
+ op.reader("partial")
+ .await
+ .unwrap()
+ .read(2..6)
+ .await
+ .unwrap()
+ .to_bytes(),
+ &b"2345"[..]
+ );
+ assert_eq!(state.partial_attempts.load(Ordering::SeqCst), 2);
+ assert_eq!(
+ reader
+ .read(BytesRange::Suffix { size: 3 })
+ .await
+ .unwrap()
+ .to_bytes(),
+ &b"789"[..]
+ );
+ assert!(reader.read(11..).await.is_err());
+ assert_eq!(
+ op.reader("throttle")
+ .await
+ .unwrap()
+ .read(2..6)
+ .await
+ .unwrap_err()
+ .kind(),
+ ErrorKind::RateLimited
+ );
+ let entries = op.list("objects/").await.unwrap();
+ assert_eq!(
+ entries.iter().map(|e| e.path()).collect::<Vec<_>>(),
+ ["objects/a", "objects/sub/", "objects/b"]
+ );
+ assert_eq!(
+ op.list_with("objects/")
+ .recursive(true)
+ .await
+ .unwrap()
+ .len(),
+ 2
+ );
+ assert!(op.list("bad/").await.is_err());
+ assert!(state
+ .user_agents
+ .lock()
+ .unwrap()
+ .iter()
+ .all(|ua| ua.contains("oss-cpp")));
+ assert!(state
+ .tokens
+ .lock()
+ .unwrap()
+ .iter()
+ .all(|token| token == "test-sts"));
+ // Token refresh creates a new FileIO; do not share a client across
credentials.
+ options.insert("fs.oss.securityToken".into(), "refreshed-sts".into());
+ let refreshed = crate::io::FileIOBuilder::new("oss")
+ .with_props(options)
+ .build()
+ .unwrap();
+ assert_eq!(
+ refreshed
+ .get_status("oss://bucket/data")
+ .await
+ .unwrap()
+ .size,
+ 10
+ );
+ assert_eq!(
+ state.tokens.lock().unwrap().last().unwrap(),
+ "refreshed-sts"
+ );
+ // Existing readers retain the original client's lifetime and credentials.
+ assert_eq!(reader.read(2..6).await.unwrap().to_bytes(), &b"2345"[..]);
+ assert_eq!(state.tokens.lock().unwrap().last().unwrap(), "test-sts");
+ server.abort();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[ignore = "requires the compiled OSS C++ SDK bridge; no cloud credentials"]
+async fn real_sdk_request_gate_and_cancelled_future() {
+ let (op, state, server, _) = fixture(2).await;
+ let reads = (0..8).map(|_| {
+ let op = op.clone();
+ async move {
+ op.reader("slow").await.unwrap().read(0..4).await.unwrap();
+ }
+ });
+ futures::future::join_all(reads).await;
+ assert_eq!(state.peak.load(Ordering::SeqCst), 2);
+ server.abort();
+
+ let (_, state, server, options) = fixture(1).await;
+ let file_io = crate::io::FileIOBuilder::new("oss")
+ .with_props(options)
+ .build()
+ .unwrap();
+ let reads = (0..8).map(|i| {
+ let file_io = file_io.clone();
+ async move {
+ let bucket = if i % 2 == 0 { "bucket-a" } else { "bucket-b" };
+ let path = format!("oss://{bucket}/slow");
+ let reader =
file_io.new_input(&path).unwrap().reader().await.unwrap();
+ assert_eq!(reader.read(0..4).await.unwrap(), &b"0123"[..]);
+ }
+ });
+ futures::future::join_all(reads).await;
+ assert_eq!(
+ state.peak.load(Ordering::SeqCst),
+ 1,
+ "two buckets in one FileIO must share the request gate"
+ );
+ server.abort();
+
+ let (op, state, server, _) = fixture(1).await;
+ let first = op.clone();
+ let task = tokio::spawn(async move {
first.reader("slow").await.unwrap().read(0..4).await });
+ tokio::time::timeout(Duration::from_secs(5), async {
+ while state.active.load(Ordering::SeqCst) == 0 {
+ tokio::time::sleep(Duration::from_millis(1)).await;
+ }
+ })
+ .await
+ .unwrap();
+ task.abort();
+ op.reader("slow").await.unwrap().read(0..4).await.unwrap();
+ assert_eq!(
+ state.peak.load(Ordering::SeqCst),
+ 1,
+ "cancelled caller must retain permit until IO completes"
+ );
+ server.abort();
+}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 2d05ec11..85e6487a 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -45,6 +45,7 @@ Available storage features:
| `storage-s3` | Amazon S3 |
| `storage-oss` | Alibaba Cloud OSS|
| `storage-jindo` | Alibaba Cloud OSS through JindoSDK |
+| `storage-oss-cpp` | Alibaba Cloud OSS through C++ SDK v2 (read-only) |
| `storage-cos` | Tencent Cloud COS |
| `storage-azdls` | Azure Data Lake Storage Gen2 |
| `storage-obs` | Huawei Cloud OBS |
@@ -314,6 +315,12 @@ limit for the available network and JindoSDK connection
capacity. JindoSDK state
initialized before `fork` cannot be reused in the child; use `spawn` or
initialize
Jindo only after worker processes start.
+With `storage-oss-cpp`, set `fs.oss.impl=cpp` and
+`fs.oss.cpp.library.path` to the installed bridge library. This experimental
+backend supports stat, reads and listings, with 8 concurrent requests per
+operator by default. OpenDAL remains the default backend.
+See the [build instructions and
options](https://github.com/apache/paimon-rust/tree/main/integrations/oss-cpp).
+
Supported metastore types:
| Metastore Type | Description |
diff --git a/integrations/oss-cpp/CMakeLists.txt
b/integrations/oss-cpp/CMakeLists.txt
new file mode 100644
index 00000000..72f642e0
--- /dev/null
+++ b/integrations/oss-cpp/CMakeLists.txt
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+cmake_minimum_required(VERSION 3.15)
+project(oss_cpp_bridge LANGUAGES CXX)
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+find_package(alibabacloud_oss_v2 REQUIRED CONFIG)
+add_library(oss_cpp_bridge SHARED bridge.cc)
+target_link_libraries(oss_cpp_bridge PRIVATE alibabacloud_oss_v2::oss)
+install(TARGETS oss_cpp_bridge LIBRARY DESTINATION lib)
diff --git a/integrations/oss-cpp/README.md b/integrations/oss-cpp/README.md
new file mode 100644
index 00000000..77adf236
--- /dev/null
+++ b/integrations/oss-cpp/README.md
@@ -0,0 +1,80 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# OSS C++ SDK backend (experimental, read-only)
+
+An opt-in, read-only FileIO backend using [Alibaba Cloud OSS C++ SDK
v2](https://github.com/aliyun/alibabacloud-oss-cpp-sdk-v2)
+for HEAD, Range GET and paginated LIST. OpenDAL remains the default.
+
+## Build
+
+Install the SDK, then build the bridge and enable the Rust feature:
+
+```bash
+cmake -S integrations/oss-cpp -B build/oss-cpp \
+ -DCMAKE_PREFIX_PATH=/path/to/oss-sdk-install
+cmake --build build/oss-cpp
+cargo build -p paimon --features storage-oss-cpp
+```
+
+The Python binding already includes this feature, but neither the bridge nor
+the SDK is bundled in its wheel. Install both on each worker. Tested on Linux
+with SDK 0.2.0 (`c13a82038b0696f99c32f4e86b623ffcc9e62140`).
+
+## Configure
+
+Set the native table/catalog FileIO options:
+
+```text
+fs.oss.impl = cpp
+fs.oss.cpp.library.path = /absolute/path/liboss_cpp_bridge.so
+fs.oss.endpoint = https://oss-cn-shanghai-internal.aliyuncs.com
+fs.oss.region = cn-shanghai
+```
+
+Use the existing `fs.oss.accessKeyId`, `fs.oss.accessKeySecret` and optional
+`fs.oss.securityToken` credentials. REST catalog credentials still refresh;
+static FileIO credentials do not.
+
+| Option | Default |
+|---|---:|
+| `fs.oss.cpp.max.concurrent.requests` | 8 |
+| `fs.oss.cpp.connect.timeout-ms` | 10000 |
+| `fs.oss.cpp.request.timeout-ms` | 30000 |
+| `fs.oss.cpp.retry.max-attempts` | 3 |
+| `fs.oss.cpp.path-style` | false |
+
+The SDK handles retries. Timeouts are per connection/read-write, not
+end-to-end. Concurrency is bounded per FileIO, not per process; cancelled
+calls retain a slot until the SDK returns. Use spawned workers: SDK access
+after fork is rejected. Mutations and presigning are unsupported.
+Python-side image downloads are unaffected.
+
+## Test
+
+```bash
+cargo test -p paimon --features storage-oss-cpp --lib io::storage_oss_cpp
+
+OSS_CPP_BRIDGE_LIBRARY="$PWD/build/oss-cpp/liboss_cpp_bridge.so" \
+ cargo test -p paimon --features storage-oss-cpp --lib io::storage_oss_cpp \
+ -- --ignored
+```
+
+The ignored tests use the real SDK, fake credentials and a local HTTP server;
+no cloud account is needed.
diff --git a/integrations/oss-cpp/bridge.cc b/integrations/oss-cpp/bridge.cc
new file mode 100644
index 00000000..4f33a080
--- /dev/null
+++ b/integrations/oss-cpp/bridge.cc
@@ -0,0 +1,190 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "bridge.h"
+#include <alibabacloud/oss2/ClientConfiguration.h>
+#include <alibabacloud/oss2/OSSClient.h>
+#include <alibabacloud/oss2/credentials/CredentialsProvider.h>
+#include <alibabacloud/oss2/io/ByteWriter.h>
+#include <alibabacloud/oss2/models/BucketBasic.h>
+#include <alibabacloud/oss2/models/ObjectBasic.h>
+#include <cstdio>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <string>
+
+namespace oss = alibabacloud::oss2;
+
+static int32_t failure(OssCppBridgeError *error, const char *code,
+ int status = 0, const char *request_id = "") noexcept {
+ error->status = status;
+ std::snprintf(error->code, sizeof(error->code), "%s", code);
+ std::snprintf(error->request_id, sizeof(error->request_id), "%s",
request_id);
+ return -1;
+}
+
+template <class F>
+static int32_t guarded(OssCppBridgeError *error, F fn) noexcept {
+ *error = {};
+ try {
+ return fn();
+ }
+ // Do not include SDK exception messages: they can contain signed requests.
+ catch (...) {
+ return failure(error, "CppException");
+ }
+}
+
+template <class E>
+static int32_t sdk_failure(OssCppBridgeError *error, const E &e) noexcept {
+ return failure(error, e.getCode().c_str(), e.getStatusCode(),
+ e.getRequestId().c_str());
+}
+
+extern "C" {
+uint32_t oss_cpp_bridge_abi_version() { return 2; }
+
+void *oss_cpp_bridge_create(const OssCppBridgeConfig *in,
+ OssCppBridgeError *error) {
+ void *client = nullptr;
+ guarded(error, [&]() {
+ oss::ClientConfiguration config;
+ config.endpoint = in->endpoint;
+ config.region = in->region;
+ config.credentialsProvider =
+ std::make_shared<oss::StaticCredentialsProvider>(
+ in->access_key, in->secret_key, in->token);
+ config.userAgent = in->user_agent;
+ config.connectTimeout = in->connect_timeout_ms;
+ config.readWriteTimeout = in->request_timeout_ms;
+ config.retryMaxAttempts = in->retry_attempts;
+ config.usePathStyle = in->path_style != 0;
+ client = new oss::OSSClient(config);
+ return 0;
+ });
+ return client;
+}
+
+void oss_cpp_bridge_destroy(void *client) {
+ try {
+ delete static_cast<oss::OSSClient *>(client);
+ } catch (...) {
+ }
+}
+
+int32_t oss_cpp_bridge_stat(void *client, const char *bucket, const char *key,
+ OssCppBridgeMetadata *out,
+ OssCppBridgeError *error) {
+ return guarded(error, [&]() {
+ auto result = static_cast<oss::OSSClient *>(client)->headObject(
+ oss::models::HeadObjectRequest().setBucket(bucket).setKey(key));
+ if (!result.has_value())
+ return sdk_failure(error, result.error());
+ const auto &value = result.value();
+ if (value.getContentLength() < 0)
+ return failure(error, "InvalidContentLength");
+ out->size = value.getContentLength();
+ std::snprintf(out->modified, sizeof(out->modified), "%s",
+ value.getLastModified().c_str());
+ return 0;
+ });
+}
+
+int32_t oss_cpp_bridge_read(void *client, const char *bucket, const char *key,
+ uint64_t offset, size_t length, uint8_t *buffer,
+ OssCppBridgeError *error) {
+ return guarded(error, [&]() {
+ if (!length)
+ return 0;
+ if (offset > UINT64_MAX - (length - 1))
+ return failure(error, "InvalidRange");
+ std::shared_ptr<oss::MemoryWriter> sink;
+ oss::SinkFactory factory;
+ factory.isOneShot = false;
+ // A retry must start at the beginning of the caller's buffer.
+ factory.supplier =
+ [&](int64_t,
+ const oss::HeaderCollection &) -> std::shared_ptr<oss::ByteWriter>
{
+ sink = std::make_shared<oss::MemoryWriter>(buffer, length);
+ return sink;
+ };
+ const auto end = offset + length - 1;
+ auto result = static_cast<oss::OSSClient *>(client)->getObject(
+ oss::models::GetObjectRequest()
+ .setBucket(bucket)
+ .setKey(key)
+ .setRange("bytes=" + std::to_string(offset) + "-" +
+ std::to_string(end))
+ .setSinkFactory(factory));
+ if (!result.has_value())
+ return sdk_failure(error, result.error());
+ const auto &value = result.value();
+ const auto prefix =
+ "bytes " + std::to_string(offset) + "-" + std::to_string(end) + "/";
+ if (value.getStatusCode() != 206 ||
+ value.getContentRange().rfind(prefix, 0) != 0 ||
+ value.getContentLength() != static_cast<int64_t>(length) || !sink ||
+ sink->written() != length) {
+ return failure(error, "InvalidRangeResponse", value.getStatusCode());
+ }
+ return 0;
+ });
+}
+
+int32_t oss_cpp_bridge_list(void *client, const char *bucket,
+ const char *prefix, const char *token,
+ uint8_t recursive, uint32_t max_keys,
+ OssCppBridgeEntry entry, void *ctx,
+ char **next_token, OssCppBridgeError *error) {
+ *next_token = nullptr;
+ return guarded(error, [&]() {
+ auto request = oss::models::ListObjectsV2Request()
+ .setBucket(bucket)
+ .setPrefix(prefix)
+ .setMaxKeys(max_keys);
+ if (!recursive)
+ request.setDelimiter("/");
+ if (*token)
+ request.setContinuationToken(token);
+ auto result = static_cast<oss::OSSClient
*>(client)->listObjectsV2(request);
+ if (!result.has_value())
+ return sdk_failure(error, result.error());
+ const auto &value = result.value();
+ std::unique_ptr<char[]> next;
+ if (value.getIsTruncated()) {
+ const auto &s = value.getNextContinuationToken();
+ if (s.empty() || s == token)
+ return failure(error, "InvalidContinuationToken");
+ next = std::make_unique<char[]>(s.size() + 1);
+ std::memcpy(next.get(), s.c_str(), s.size() + 1);
+ }
+ for (const auto &item : value.getContents()) {
+ entry(ctx, item.key.data(), item.key.size(), item.size,
+ item.lastModified.c_str(),
+ !item.key.empty() && item.key.back() == '/');
+ }
+ for (const auto &item : value.getCommonPrefixes()) {
+ entry(ctx, item.prefix.data(), item.prefix.size(), 0, "", 1);
+ }
+ *next_token = next.release();
+ return 0;
+ });
+}
+
+void oss_cpp_bridge_free_string(char *s) { delete[] s; }
+}
diff --git a/integrations/oss-cpp/bridge.h b/integrations/oss-cpp/bridge.h
new file mode 100644
index 00000000..3cc18040
--- /dev/null
+++ b/integrations/oss-cpp/bridge.h
@@ -0,0 +1,65 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+#include <stddef.h>
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// ABI v2: all strings are UTF-8; inputs are borrowed for the call.
+// Clients allow concurrent calls; each call owns its error output.
+// No C++ exception may cross this boundary.
+typedef struct {
+ const char *endpoint, *region, *access_key, *secret_key, *token, *user_agent;
+ int64_t connect_timeout_ms, request_timeout_ms, retry_attempts;
+ uint8_t path_style;
+} OssCppBridgeConfig;
+
+typedef struct {
+ int32_t status;
+ char code[128];
+ char request_id[256];
+} OssCppBridgeError;
+
+typedef struct {
+ int64_t size;
+ char modified[64];
+} OssCppBridgeMetadata;
+
+typedef void (*OssCppBridgeEntry)(void *, const char *, size_t, int64_t,
+ const char *, uint8_t);
+
+uint32_t oss_cpp_bridge_abi_version(void);
+void *oss_cpp_bridge_create(const OssCppBridgeConfig *, OssCppBridgeError *);
+void oss_cpp_bridge_destroy(void *);
+int32_t oss_cpp_bridge_stat(void *, const char *, const char *,
+ OssCppBridgeMetadata *, OssCppBridgeError *);
+int32_t oss_cpp_bridge_read(void *, const char *, const char *, uint64_t,
+ size_t, uint8_t *, OssCppBridgeError *);
+// The callback borrows strings until it returns. next_token is owned by
+// the bridge and must be released with oss_cpp_bridge_free_string.
+int32_t oss_cpp_bridge_list(void *, const char *, const char *, const char *,
+ uint8_t, uint32_t, OssCppBridgeEntry, void *,
+ char **, OssCppBridgeError *);
+void oss_cpp_bridge_free_string(char *);
+
+#ifdef __cplusplus
+}
+#endif
