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 2d0dda18 feat(storage): support Jindo for OSS scan planning (#696)
2d0dda18 is described below
commit 2d0dda18bd7b5db0acbf1aee588aaed1ed864da5
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Oct 1 22:41:30 2026 +0800
feat(storage): support Jindo for OSS scan planning (#696)
---
.github/workflows/ci.yml | 22 +
Cargo.lock | 1 +
bindings/python/src/context.rs | 128 ++-
bindings/python/src/table.rs | 12 +-
bindings/python/tests/test_datafusion.py | 30 +
crates/paimon/Cargo.toml | 5 +
crates/paimon/build.rs | 26 +
crates/paimon/src/io/jindo_ffi.cc | 93 ++
crates/paimon/src/io/mod.rs | 5 +
crates/paimon/src/io/storage.rs | 43 +
crates/paimon/src/io/storage_jindo.rs | 1675 ++++++++++++++++++++++++++++++
docs/src/getting-started.md | 10 +
12 files changed, 2045 insertions(+), 5 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 735aa628..692a4680 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -236,6 +236,10 @@ jobs:
needs_docker: true
needs_lumina: false
needs_uv: false
+ - suite: jindo
+ needs_docker: false
+ needs_lumina: false
+ needs_uv: false
env:
RUST_LOG: DEBUG
RUST_BACKTRACE: full
@@ -313,6 +317,24 @@ jobs:
working-directory: bindings/go
run: make test
+ - uses: actions/setup-python@v6
+ if: matrix.suite == 'jindo'
+ with:
+ python-version: "3.11"
+
+ - name: Install Jindo SDK
+ if: matrix.suite == 'jindo'
+ run: |
+ python -m pip install --only-binary=:all: --no-deps
pyjindosdk==6.10.4
+ echo "PAIMON_JINDO_SDK_LIBRARY=$(python -c 'import pathlib, pyjindo;
print(pathlib.Path(pyjindo.__path__[0]) / "libjindosdk_python.so")')" >>
"$GITHUB_ENV"
+
+ - name: Jindo SDK Integration Test
+ if: matrix.suite == 'jindo'
+ run: >
+ cargo test --locked -p paimon --features storage-jindo --lib
+ io::storage_jindo::tests::test_real_sdk_io_and_failure_mapping
+ -- --ignored --exact
+
- name: Stop Docker containers
if: ${{ always() && matrix.needs_docker }}
run: make docker-down
diff --git a/Cargo.lock b/Cargo.lock
index e08d17a4..1ed0be40 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4665,6 +4665,7 @@ dependencies = [
"bitflags 2.13.1",
"bytes",
"bzip2",
+ "cc",
"chrono",
"crc32fast",
"crossbeam-channel",
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 65bbe9f9..c0628460 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -83,7 +83,76 @@ fn pyarrow_compatible_batch(batch: &RecordBatch) ->
arrow::error::Result<RecordB
RecordBatch::try_new_with_options(schema, columns, &options)
}
+const OSS_IMPL: &str = "fs.oss.impl";
+const JINDO_LIBRARY_PATH: &str = "fs.jindo.library.path";
+const JINDOSDK_LIBRARY_PATH: &str = "JINDOSDK_LIBRARY_PATH";
+const JINDOSDK_HOME: &str = "JINDOSDK_HOME";
+
+fn should_discover_pyjindo_library(
+ catalog_options: &HashMap<String, String>,
+ library_path: Option<&std::ffi::OsStr>,
+ home: Option<&std::ffi::OsStr>,
+) -> bool {
+ !catalog_options.contains_key(JINDO_LIBRARY_PATH) &&
library_path.is_none() && home.is_none()
+}
+
+fn discover_pyjindo_library() -> Option<PathBuf> {
+ Python::attach(|py| {
+ let spec = py
+ .import("importlib.util")
+ .ok()?
+ .call_method1("find_spec", ("pyjindo",))
+ .ok()?;
+ if spec.is_none() {
+ return None;
+ }
+ let origin =
PathBuf::from(spec.getattr("origin").ok()?.extract::<String>().ok()?);
+ pyjindo_library_in(origin.parent()?)
+ })
+}
+
+fn pyjindo_library_in(directory: &std::path::Path) -> Option<PathBuf> {
+ let names = if cfg!(target_os = "macos") {
+ ["libjindosdk_c.dylib", "libjindosdk_python.dylib"]
+ } else {
+ ["libjindosdk_c.so", "libjindosdk_python.so"]
+ };
+ names
+ .iter()
+ .map(|name| directory.join(name))
+ .find(|path| path.is_file())
+}
+
+fn complete_jindo_options_with(
+ mut options: HashMap<String, String>,
+ library_path: Option<&std::ffi::OsStr>,
+ home: Option<&std::ffi::OsStr>,
+ discover: impl FnOnce() -> Option<PathBuf>,
+) -> HashMap<String, String> {
+ let use_jindo = options
+ .get(OSS_IMPL)
+ .is_some_and(|value| value.eq_ignore_ascii_case("jindo"));
+ if use_jindo && should_discover_pyjindo_library(&options, library_path,
home) {
+ if let Some(path) = discover() {
+ options.insert(JINDO_LIBRARY_PATH.to_string(),
path.display().to_string());
+ }
+ }
+ options
+}
+
+pub(crate) fn complete_jindo_options(options: HashMap<String, String>) ->
HashMap<String, String> {
+ let library_path = std::env::var_os(JINDOSDK_LIBRARY_PATH);
+ let home = std::env::var_os(JINDOSDK_HOME);
+ complete_jindo_options_with(
+ options,
+ library_path.as_deref(),
+ home.as_deref(),
+ discover_pyjindo_library,
+ )
+}
+
fn build_paimon_catalog(catalog_options: HashMap<String, String>) ->
PyResult<Arc<dyn Catalog>> {
+ let catalog_options = complete_jindo_options(catalog_options);
let rt = runtime();
rt.block_on(async {
let options = Options::from_map(catalog_options);
@@ -92,6 +161,62 @@ fn build_paimon_catalog(catalog_options: HashMap<String,
String>) -> PyResult<Ar
})
}
+#[cfg(test)]
+mod tests {
+ use std::fs;
+
+ use super::*;
+
+ #[test]
+ fn test_find_pyjindo_library() {
+ let directory = std::env::temp_dir().join(format!(
+ "paimon-rust-pyjindo-{}-{}",
+ std::process::id(),
+ std::thread::current().name().unwrap_or("test")
+ ));
+ fs::create_dir_all(&directory).unwrap();
+ let library = directory.join(if cfg!(target_os = "macos") {
+ "libjindosdk_python.dylib"
+ } else {
+ "libjindosdk_python.so"
+ });
+ fs::write(&library, []).unwrap();
+
+ assert_eq!(pyjindo_library_in(&directory), Some(library));
+
+ fs::remove_dir_all(directory).unwrap();
+ }
+
+ #[test]
+ fn test_pyjindo_discovery_precedence() {
+ let mut options = HashMap::new();
+ assert!(should_discover_pyjindo_library(&options, None, None));
+
+ let path = std::ffi::OsStr::new("/opt/jindo/libjindosdk_c.so");
+ assert!(!should_discover_pyjindo_library(&options, Some(path), None));
+ assert!(!should_discover_pyjindo_library(&options, None, Some(path)));
+
+ options.insert(
+ JINDO_LIBRARY_PATH.to_string(),
+ path.to_string_lossy().into(),
+ );
+ assert!(!should_discover_pyjindo_library(&options, None, None));
+ }
+
+ #[test]
+ fn test_complete_jindo_options() {
+ let options = HashMap::from([(OSS_IMPL.to_string(),
"jindo".to_string())]);
+ let discovered = PathBuf::from("/pyjindo/libjindosdk_python.so");
+ let completed =
+ complete_jindo_options_with(options, None, None, ||
Some(discovered.clone()));
+
+ assert_eq!(
+ completed.get(JINDO_LIBRARY_PATH).map(String::as_str),
+ Some(discovered.to_string_lossy().as_ref())
+ );
+ }
+}
+
fn ffi_logical_codec_from_pycapsule(obj: Bound<'_, PyAny>) ->
PyResult<FFI_LogicalExtensionCodec> {
let attr_name = "__datafusion_logical_extension_codec__";
let capsule = if obj.hasattr(attr_name)? {
@@ -459,8 +584,9 @@ impl PySQLContext {
catalog_options: HashMap<String, String>,
default_database: Option<String>,
) -> PyResult<()> {
+ let catalog_options = complete_jindo_options(catalog_options);
let rt = runtime();
- py.detach(|| {
+ py.detach(move || {
rt.block_on(async {
let options = Options::from_map(catalog_options);
let catalog =
CatalogFactory::create(options).await.map_err(to_py_err)?;
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 799af4a5..aa474a4d 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -27,6 +27,7 @@ use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyDict;
+use crate::context::complete_jindo_options;
use crate::error::to_py_err;
use crate::partition::PyPartitionStat;
use crate::read::PyReadBuilder;
@@ -66,10 +67,12 @@ impl PyTable {
) -> PyResult<Self> {
let schema: TableSchema = serde_json::from_str(schema_json)
.map_err(|err| PyValueError::new_err(format!("Invalid table schema
JSON: {err}")))?;
- let properties = options
- .map(crate::read::extract_options)
- .transpose()?
- .unwrap_or_default();
+ let properties = complete_jindo_options(
+ options
+ .map(crate::read::extract_options)
+ .transpose()?
+ .unwrap_or_default(),
+ );
let file_io = FileIO::from_path(&location)
.and_then(|builder| builder.with_props(properties).build())
.map_err(to_py_err)?;
@@ -100,6 +103,7 @@ impl PyTable {
PyValueError::new_err(format!("Invalid REST table response
JSON: {err}"))
})?;
let identifier = Identifier::new(database, table);
+ let rest_options = complete_jindo_options(rest_options);
let table = py
.detach(|| {
runtime().block_on(paimon::table::Table::from_rest_response(
diff --git a/bindings/python/tests/test_datafusion.py
b/bindings/python/tests/test_datafusion.py
index cfaa1a0e..94e63e0b 100644
--- a/bindings/python/tests/test_datafusion.py
+++ b/bindings/python/tests/test_datafusion.py
@@ -16,6 +16,7 @@
# under the License.
import io
+import importlib
import json
import os
import re
@@ -913,6 +914,35 @@ def test_table_functions_registered_with_catalog():
assert "requires 4 arguments" in str(e), str(e)
+def test_register_catalog_discovers_pyjindo(monkeypatch, tmp_path):
+ package = tmp_path / "pyjindo"
+ package.mkdir()
+ (package / "__init__.py").write_text("")
+ library = package / (
+ "libjindosdk_python.dylib"
+ if sys.platform == "darwin"
+ else "libjindosdk_python.so"
+ )
+ library.write_bytes(b"not a shared library")
+ monkeypatch.syspath_prepend(str(tmp_path))
+ monkeypatch.delenv("JINDOSDK_LIBRARY_PATH", raising=False)
+ monkeypatch.delenv("JINDOSDK_HOME", raising=False)
+ sys.modules.pop("pyjindo", None)
+ importlib.invalidate_caches()
+
+ ctx = SQLContext()
+ with pytest.raises(Exception) as error:
+ ctx.register_catalog(
+ "paimon",
+ {
+ "warehouse": "oss://jindo-test-bucket/warehouse",
+ "fs.oss.impl": "jindo",
+ },
+ )
+
+ assert str(library) in str(error.value)
+
+
def test_list_databases_and_tables():
catalog = PaimonCatalog({"warehouse": WAREHOUSE})
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 7fb7d914..1d6696c4 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -41,6 +41,7 @@ storage-all = [
"storage-obs",
"storage-gcs",
"storage-hdfs",
+ "storage-jindo",
]
fulltext = ["dep:paimon-ftindex-core"]
vortex = ["dep:vortex"]
@@ -61,6 +62,7 @@ storage-azdls = [
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"]
[dependencies]
rand = "0.8.5"
@@ -150,5 +152,8 @@ log = "0.4"
# The 1.13.3 resolver update correlates with Linux Vortex tests hanging.
unicode-segmentation = "=1.13.2"
+[build-dependencies]
+cc = "1"
+
[dev-dependencies]
axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
diff --git a/crates/paimon/build.rs b/crates/paimon/build.rs
new file mode 100644
index 00000000..911a5f77
--- /dev/null
+++ b/crates/paimon/build.rs
@@ -0,0 +1,26 @@
+// 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.
+
+fn main() {
+ if std::env::var_os("CARGO_FEATURE_STORAGE_JINDO").is_some() {
+ cc::Build::new()
+ .cpp(true)
+ .file("src/io/jindo_ffi.cc")
+ .flag_if_supported("-std=c++11")
+ .compile("paimon_jindo_ffi");
+ }
+}
diff --git a/crates/paimon/src/io/jindo_ffi.cc
b/crates/paimon/src/io/jindo_ffi.cc
new file mode 100644
index 00000000..fc7ec134
--- /dev/null
+++ b/crates/paimon/src/io/jindo_ffi.cc
@@ -0,0 +1,93 @@
+// 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 <cstddef>
+#include <exception>
+
+using ListDir = void* (*)(void*, const char*, bool, void*);
+using ListObjects = void* (*)(
+ void*, const char*, const char*, const char*, int, void*);
+
+namespace {
+
+void copy_error(char* output, std::size_t capacity, const char* message)
noexcept {
+ if (output == nullptr || capacity == 0) {
+ return;
+ }
+ std::size_t index = 0;
+ if (message != nullptr) {
+ while (index + 1 < capacity && message[index] != '\0') {
+ output[index] = message[index];
+ ++index;
+ }
+ }
+ output[index] = '\0';
+}
+
+} // namespace
+
+extern "C" int paimon_jindo_list_dir(
+ ListDir list_dir,
+ void* handle,
+ const char* path,
+ bool recursive,
+ void* options,
+ void** result,
+ char* error,
+ std::size_t error_capacity) noexcept {
+ if (list_dir == nullptr || result == nullptr) {
+ copy_error(error, error_capacity, "invalid Jindo list call");
+ return 1;
+ }
+ *result = nullptr;
+ try {
+ *result = list_dir(handle, path, recursive, options);
+ return 0;
+ } catch (const std::exception& exception) {
+ copy_error(error, error_capacity, exception.what());
+ } catch (...) {
+ copy_error(error, error_capacity, "unknown C++ exception");
+ }
+ return 1;
+}
+
+extern "C" int paimon_jindo_list_objects(
+ ListObjects list_objects,
+ void* handle,
+ const char* path,
+ const char* delimiter,
+ const char* marker,
+ int max_keys,
+ void* options,
+ void** result,
+ char* error,
+ std::size_t error_capacity) noexcept {
+ if (list_objects == nullptr || result == nullptr) {
+ copy_error(error, error_capacity, "invalid Jindo list call");
+ return 1;
+ }
+ *result = nullptr;
+ try {
+ *result = list_objects(handle, path, delimiter, marker, max_keys,
options);
+ return 0;
+ } catch (const std::exception& exception) {
+ copy_error(error, error_capacity, exception.what());
+ } catch (...) {
+ copy_error(error, error_capacity, "unknown C++ exception");
+ }
+ return 1;
+}
diff --git a/crates/paimon/src/io/mod.rs b/crates/paimon/src/io/mod.rs
index c7eea7c1..e53ed4fa 100644
--- a/crates/paimon/src/io/mod.rs
+++ b/crates/paimon/src/io/mod.rs
@@ -63,6 +63,11 @@ pub(crate) mod storage_oss;
#[cfg(feature = "storage-oss")]
use storage_oss::*;
+#[cfg(feature = "storage-jindo")]
+mod storage_jindo;
+#[cfg(feature = "storage-jindo")]
+use storage_jindo::*;
+
#[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 24885d29..984f72e9 100644
--- a/crates/paimon/src/io/storage.rs
+++ b/crates/paimon/src/io/storage.rs
@@ -39,6 +39,8 @@ use std::sync::MutexGuard;
#[cfg(feature = "storage-azdls")]
use super::AzdlsStorageConfig;
+#[cfg(feature = "storage-jindo")]
+use super::JindoStorageConfig;
#[cfg(feature = "storage-oss")]
use super::OssStorageConfig;
use opendal::Operator;
@@ -80,6 +82,11 @@ pub enum Storage {
config: Box<OssStorageConfig>,
operators: Mutex<HashMap<String, Operator>>,
},
+ #[cfg(feature = "storage-jindo")]
+ Jindo {
+ config: Box<JindoStorageConfig>,
+ operators: Mutex<HashMap<String, Operator>>,
+ },
#[cfg(feature = "storage-s3")]
S3 {
config: Box<S3Config>,
@@ -130,6 +137,23 @@ impl Storage {
}),
#[cfg(feature = "storage-oss")]
"oss" => {
+ #[cfg(feature = "storage-jindo")]
+ if super::use_jindo(&props)? {
+ let config = super::jindo_config_parse(props)?;
+ return Ok(Self::Jindo {
+ config: Box::new(config),
+ operators: Mutex::new(HashMap::new()),
+ });
+ }
+ #[cfg(not(feature = "storage-jindo"))]
+ if props
+ .get("fs.oss.impl")
+ .is_some_and(|value| value.eq_ignore_ascii_case("jindo"))
+ {
+ return Err(error::Error::IoUnsupported {
+ message: "Jindo requires the storage-jindo
feature".to_string(),
+ });
+ }
let config = super::oss_config_parse(props)?;
Ok(Self::Oss {
config: Box::new(config),
@@ -221,6 +245,15 @@ impl Storage {
let op = Self::cached_oss_operator(config, operators, path,
&bucket)?;
Ok((op, Cow::Borrowed(relative_path)))
}
+ #[cfg(feature = "storage-jindo")]
+ Storage::Jindo { config, operators } => {
+ let (bucket, relative_path) =
+ Self::bucket_and_relative_path(path, "Jindo OSS",
&["oss"])?;
+ let op = Self::cached_operator(operators, "Jindo OSS",
&bucket, || {
+ super::jindo_config_build(config, &bucket)
+ })?;
+ Ok((op, Cow::Borrowed(relative_path)))
+ }
#[cfg(feature = "storage-s3")]
Storage::S3 { config, operators } => {
let (bucket, relative_path) =
@@ -333,6 +366,7 @@ impl Storage {
#[cfg(any(
feature = "storage-cos",
feature = "storage-gcs",
+ feature = "storage-jindo",
feature = "storage-obs",
feature = "storage-oss",
feature = "storage-s3"
@@ -372,6 +406,7 @@ impl Storage {
feature = "storage-azdls",
feature = "storage-cos",
feature = "storage-gcs",
+ feature = "storage-jindo",
feature = "storage-oss",
feature = "storage-obs",
feature = "storage-s3"
@@ -474,6 +509,14 @@ mod scheme_tests {
}
}
+ #[cfg(feature = "storage-jindo")]
+ #[test]
+ fn jindo_oss_implementation_is_selected() {
+ let storage =
+ Storage::build(FileIOBuilder::new("oss").with_prop("fs.oss.impl",
"jindo")).unwrap();
+ assert!(matches!(storage, Storage::Jindo { .. }));
+ }
+
#[cfg(feature = "storage-s3")]
#[test]
fn s3_scheme_aliases_are_compatible() {
diff --git a/crates/paimon/src/io/storage_jindo.rs
b/crates/paimon/src/io/storage_jindo.rs
new file mode 100644
index 00000000..699b41b7
--- /dev/null
+++ b/crates/paimon/src/io/storage_jindo.rs
@@ -0,0 +1,1675 @@
+// 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::{CStr, CString, OsStr};
+use std::fmt::{Debug, Formatter};
+use std::os::raw::{c_char, c_void};
+use std::path::{Path, PathBuf};
+use std::sync::mpsc::{self, Sender};
+use std::sync::{Arc, Mutex, OnceLock};
+
+use libloading::Library;
+use opendal::raw::*;
+use opendal::{Buffer, Builder, BytesRange, Capability, EntryMode, ErrorKind,
Metadata};
+use opendal::{Error as OpendalError, OperationContext, Operator, Result as
OpendalResult};
+use tokio::sync::{OnceCell, Semaphore};
+
+use crate::error::Error;
+use crate::Result;
+
+const OSS_IMPL: &str = "fs.oss.impl";
+const JINDO_IMPL: &str = "jindo";
+const JINDO_LIBRARY_PATH: &str = "fs.jindo.library.path";
+const JINDO_USER: &str = "fs.jindo.user";
+const JINDO_MAX_CONCURRENT_READS: &str = "fs.jindo.max.concurrent.reads";
+const DEFAULT_JINDO_MAX_CONCURRENT_READS: usize = 8;
+const DEFAULT_JINDO_USER: &str = "root";
+const OSS_ENDPOINT: &str = "fs.oss.endpoint";
+const OSS_HTTPS_ENABLE: &str = "fs.oss.https.enable";
+const JDO_LIST_ITERATIVE: &str = "JDO_LIST_OPTS_IS_ITERATIVE";
+const JDO_LIST_MARKER: &str = "JDO_LIST_OPTS_MARKER";
+
+const JDO_FILE_NOT_FOUND_ERROR: i32 = 3001;
+const JDO_REST_HTTP_403_ERROR: i32 = 6403;
+const JDO_REST_HTTP_404_ERROR: i32 = 6404;
+const JDO_REST_HTTP_503_ERROR: i32 = 6503;
+const JINDO_EXCEPTION_BUFFER_SIZE: usize = 1024;
+
+type JdoPtr = *mut c_void;
+type JindoTeardown = Box<dyn FnOnce() + Send + 'static>;
+
+type CreateOptions = unsafe extern "C" fn() -> JdoPtr;
+type FreeOptions = unsafe extern "C" fn(JdoPtr);
+type SetOption = unsafe extern "C" fn(JdoPtr, *const c_char, *const c_char);
+type CreateStore = unsafe extern "C" fn(JdoPtr, *const c_char) -> JdoPtr;
+type DestroyStore = unsafe extern "C" fn(JdoPtr);
+type FreeStore = unsafe extern "C" fn(JdoPtr);
+type CreateHandle = unsafe extern "C" fn(JdoPtr) -> JdoPtr;
+type FreeHandle = unsafe extern "C" fn(JdoPtr);
+type HandleErrorCode = unsafe extern "C" fn(JdoPtr) -> i32;
+type HandleErrorMessage = unsafe extern "C" fn(JdoPtr) -> *const c_char;
+type Init = unsafe extern "C" fn(JdoPtr, *const c_char);
+type GetFileStatus = unsafe extern "C" fn(JdoPtr, *const c_char, JdoPtr) ->
JdoPtr;
+type FreeFileStatus = unsafe extern "C" fn(JdoPtr);
+type FileStatusPath = unsafe extern "C" fn(JdoPtr) -> *const c_char;
+type FileStatusType = unsafe extern "C" fn(JdoPtr) -> i8;
+type FileStatusSize = unsafe extern "C" fn(JdoPtr) -> i64;
+type FileStatusMtime = unsafe extern "C" fn(JdoPtr) -> i64;
+type ListDir = unsafe extern "C" fn(JdoPtr, *const c_char, bool, JdoPtr) ->
JdoPtr;
+type FreeListDirResult = unsafe extern "C" fn(JdoPtr);
+type ListDirResultSize = unsafe extern "C" fn(JdoPtr) -> i64;
+type IsListDirResultTruncated = unsafe extern "C" fn(JdoPtr) -> bool;
+type ListDirResultNextMarker = unsafe extern "C" fn(JdoPtr) -> *const c_char;
+type ListDirFileStatus = unsafe extern "C" fn(JdoPtr, usize) -> JdoPtr;
+type ListObjects = unsafe extern "C" fn(
+ JdoPtr,
+ *const c_char,
+ *const c_char,
+ *const c_char,
+ i32,
+ JdoPtr,
+) -> JdoPtr;
+type FreeListObjectsResult = unsafe extern "C" fn(JdoPtr);
+type GetObject = unsafe extern "C" fn(JdoPtr, *const c_char, *mut c_char, i64,
i64, JdoPtr) -> i64;
+
+unsafe extern "C" {
+ fn paimon_jindo_list_dir(
+ list_dir: ListDir,
+ handle: JdoPtr,
+ path: *const c_char,
+ recursive: bool,
+ options: JdoPtr,
+ result: *mut JdoPtr,
+ error: *mut c_char,
+ error_capacity: usize,
+ ) -> i32;
+ fn paimon_jindo_list_objects(
+ list_objects: ListObjects,
+ handle: JdoPtr,
+ path: *const c_char,
+ delimiter: *const c_char,
+ marker: *const c_char,
+ max_keys: i32,
+ options: JdoPtr,
+ result: *mut JdoPtr,
+ error: *mut c_char,
+ error_capacity: usize,
+ ) -> i32;
+}
+
+#[derive(Clone)]
+pub struct JindoStorageConfig {
+ library_path: Option<PathBuf>,
+ user: String,
+ max_concurrent_reads: usize,
+ properties: HashMap<String, String>,
+}
+
+impl Debug for JindoStorageConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("JindoStorageConfig")
+ .field("library_path", &self.library_path)
+ .field("user", &self.user)
+ .field("max_concurrent_reads", &self.max_concurrent_reads)
+ .finish_non_exhaustive()
+ }
+}
+
+pub(crate) fn use_jindo(props: &HashMap<String, String>) -> Result<bool> {
+ match props.get(OSS_IMPL).map(|value| value.to_ascii_lowercase()) {
+ Some(value) if value == JINDO_IMPL => Ok(true),
+ Some(value) if value == "legacy" => Ok(false),
+ Some(value) => Err(Error::ConfigInvalid {
+ message: format!("Unsupported {OSS_IMPL}: {value}"),
+ }),
+ None => Ok(false),
+ }
+}
+
+pub(crate) fn jindo_config_parse(mut props: HashMap<String, String>) ->
Result<JindoStorageConfig> {
+ let library_path = props.remove(JINDO_LIBRARY_PATH).map(PathBuf::from);
+ let user = props
+ .remove(JINDO_USER)
+ .unwrap_or_else(|| DEFAULT_JINDO_USER.to_string());
+ let max_concurrent_reads = match props.remove(JINDO_MAX_CONCURRENT_READS) {
+ Some(value) => value
+ .parse::<usize>()
+ .ok()
+ .filter(|&limit| (1..=Semaphore::MAX_PERMITS).contains(&limit))
+ .ok_or_else(|| Error::ConfigInvalid {
+ message: format!(
+ "{JINDO_MAX_CONCURRENT_READS} must be between 1 and {}",
+ Semaphore::MAX_PERMITS
+ ),
+ })?,
+ None => DEFAULT_JINDO_MAX_CONCURRENT_READS,
+ };
+ props.remove(OSS_IMPL);
+ props.insert(
+ "fs.oss.user.agent.features".to_string(),
+ "paimon-rust".to_string(),
+ );
+ props
+ .entry("logger.appender".to_string())
+ .or_insert_with(|| "file".to_string());
+ if let Some(endpoint) = props.get(OSS_ENDPOINT).cloned() {
+ if let Some((scheme, endpoint)) = endpoint.split_once("://") {
+ if scheme.eq_ignore_ascii_case("http") ||
scheme.eq_ignore_ascii_case("https") {
+ props.insert(OSS_ENDPOINT.to_string(), endpoint.to_string());
+ props
+ .entry(OSS_HTTPS_ENABLE.to_string())
+ .or_insert_with(||
scheme.eq_ignore_ascii_case("https").to_string());
+ }
+ }
+ }
+ Ok(JindoStorageConfig {
+ library_path,
+ user,
+ max_concurrent_reads,
+ properties: props,
+ })
+}
+
+pub(crate) fn jindo_config_build(config: &JindoStorageConfig, bucket: &str) ->
Result<Operator> {
+ let builder = JindoBuilder {
+ config: Some(config.clone()),
+ bucket: bucket.to_string(),
+ };
+ Operator::new(builder).map_err(|source| Error::IoUnexpected {
+ message: format!("Failed to initialize Jindo for OSS bucket
'{bucket}'"),
+ source: Box::new(source),
+ })
+}
+
+#[derive(Default)]
+struct JindoBuilder {
+ config: Option<JindoStorageConfig>,
+ bucket: String,
+}
+
+impl Builder for JindoBuilder {
+ type Config = ();
+
+ fn build(self) -> OpendalResult<impl Service> {
+ let config = self.config.ok_or_else(|| {
+ OpendalError::new(ErrorKind::ConfigInvalid, "missing Jindo
configuration")
+ })?;
+ Ok(JindoService {
+ info: ServiceInfo::new("jindo", "/", &self.bucket),
+ client: Arc::new(LazyJindoClient::new(config, self.bucket)),
+ })
+ }
+}
+
+#[derive(Clone)]
+struct JindoService {
+ info: ServiceInfo,
+ client: Arc<LazyJindoClient>,
+}
+
+impl Debug for JindoService {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("JindoService")
+ .field("info", &self.info)
+ .finish()
+ }
+}
+
+impl Service for JindoService {
+ type Reader = JindoReader;
+ type Writer = ();
+ type Lister = JindoLister;
+ 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 create_dir(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpCreateDir,
+ ) -> OpendalResult<RpCreateDir> {
+ unsupported("create_dir")
+ }
+
+ async fn stat(
+ &self,
+ _ctx: &OperationContext,
+ path: &str,
+ _args: OpStat,
+ ) -> OpendalResult<RpStat> {
+ let client = self.client.get().await?;
+ let path = path.to_string();
+ let metadata = run_blocking(move || client.stat(&path)).await?;
+ Ok(RpStat::new(metadata))
+ }
+
+ fn read(
+ &self,
+ _ctx: &OperationContext,
+ path: &str,
+ _args: OpRead,
+ ) -> OpendalResult<Self::Reader> {
+ Ok(JindoReader {
+ client: Arc::clone(&self.client),
+ path: path.to_string(),
+ })
+ }
+
+ fn write(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpWrite,
+ ) -> OpendalResult<Self::Writer> {
+ unsupported("write")
+ }
+
+ fn delete(&self, _ctx: &OperationContext) -> OpendalResult<Self::Deleter> {
+ unsupported("delete")
+ }
+
+ fn list(
+ &self,
+ _ctx: &OperationContext,
+ path: &str,
+ args: OpList,
+ ) -> OpendalResult<Self::Lister> {
+ Ok(JindoLister {
+ client: Arc::clone(&self.client),
+ path: path.to_string(),
+ recursive: args.recursive(),
+ entries: Vec::new().into_iter(),
+ marker: None,
+ finished: false,
+ })
+ }
+
+ fn copy(
+ &self,
+ _ctx: &OperationContext,
+ _from: &str,
+ _to: &str,
+ _args: OpCopy,
+ _opts: OpCopier,
+ ) -> OpendalResult<Self::Copier> {
+ unsupported("copy")
+ }
+
+ async fn rename(
+ &self,
+ _ctx: &OperationContext,
+ _from: &str,
+ _to: &str,
+ _args: OpRename,
+ ) -> OpendalResult<RpRename> {
+ unsupported("rename")
+ }
+
+ async fn presign(
+ &self,
+ _ctx: &OperationContext,
+ _path: &str,
+ _args: OpPresign,
+ ) -> OpendalResult<RpPresign> {
+ unsupported("presign")
+ }
+}
+
+struct JindoReader {
+ client: Arc<LazyJindoClient>,
+ path: String,
+}
+
+impl oio::Read for JindoReader {
+ async fn open(
+ &self,
+ range: BytesRange,
+ ) -> OpendalResult<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
+ let (_, buffer) = self.read(range).await?;
+ Ok((RpRead::default(), Box::new(buffer)))
+ }
+
+ async fn read(&self, range: BytesRange) -> OpendalResult<(RpRead, Buffer)>
{
+ let client = self.client.get().await?;
+ let path = self.path.clone();
+ let buffer =
run_blocking_with_gate(Arc::clone(&self.client.read_gate), move || {
+ client.read(&path, range)
+ })
+ .await?;
+ Ok((RpRead::default(), Buffer::from(buffer)))
+ }
+}
+
+struct JindoLister {
+ client: Arc<LazyJindoClient>,
+ path: String,
+ recursive: bool,
+ entries: std::vec::IntoIter<oio::Entry>,
+ marker: Option<String>,
+ finished: bool,
+}
+
+impl oio::List for JindoLister {
+ async fn next(&mut self) -> OpendalResult<Option<oio::Entry>> {
+ loop {
+ if let Some(entry) = self.entries.next() {
+ return Ok(Some(entry));
+ }
+ if self.finished {
+ return Ok(None);
+ }
+ let client = self.client.get().await?;
+ let path = self.path.clone();
+ let recursive = self.recursive;
+ let marker = self.marker.clone();
+ let page = run_blocking(move || client.list_page(&path, recursive,
marker)).await?;
+ self.finished = page.next_marker.is_none();
+ self.marker = page.next_marker;
+ self.entries = page.entries.into_iter();
+ }
+ }
+}
+
+struct JindoListPage {
+ entries: Vec<oio::Entry>,
+ next_marker: Option<String>,
+}
+
+struct LazyJindoClient {
+ config: JindoStorageConfig,
+ bucket: String,
+ client: OnceCell<Arc<JindoClient>>,
+ read_gate: Arc<Semaphore>,
+}
+
+impl LazyJindoClient {
+ fn new(config: JindoStorageConfig, bucket: String) -> Self {
+ let read_gate = Arc::new(Semaphore::new(config.max_concurrent_reads));
+ Self {
+ config,
+ bucket,
+ client: OnceCell::new(),
+ read_gate,
+ }
+ }
+
+ async fn get(&self) -> OpendalResult<Arc<JindoClient>> {
+ let client = self
+ .client
+ .get_or_try_init(|| {
+ let config = self.config.clone();
+ let bucket = self.bucket.clone();
+ async move {
+ run_blocking(move || JindoClient::new(&config, &bucket))
+ .await
+ .map(Arc::new)
+ }
+ })
+ .await?;
+ Ok(Arc::clone(client))
+ }
+}
+
+async fn run_blocking<T: Send + 'static>(
+ call: impl FnOnce() -> OpendalResult<T> + Send + 'static,
+) -> OpendalResult<T> {
+ tokio::task::spawn_blocking(call).await.map_err(|source| {
+ OpendalError::new(ErrorKind::Unexpected, "Jindo task
failed").set_source(source)
+ })?
+}
+
+async fn run_blocking_with_gate<T: Send + 'static>(
+ gate: Arc<Semaphore>,
+ call: impl FnOnce() -> OpendalResult<T> + Send + 'static,
+) -> OpendalResult<T> {
+ let permit = gate.acquire_owned().await.map_err(|source| {
+ OpendalError::new(ErrorKind::Unexpected, "Jindo read gate
closed").set_source(source)
+ })?;
+ run_blocking(move || {
+ // Keep the permit inside the blocking closure. Dropping an async
+ // caller cannot cancel an in-flight Jindo C SDK request.
+ let _permit = permit;
+ call()
+ })
+ .await
+}
+
+fn unsupported<T>(operation: &str) -> OpendalResult<T> {
+ Err(OpendalError::new(
+ ErrorKind::Unsupported,
+ format!("Jindo {operation} is not supported"),
+ ))
+}
+
+fn jindo_teardown_sender() -> OpendalResult<&'static Sender<JindoTeardown>> {
+ static SENDER: OnceLock<std::result::Result<Sender<JindoTeardown>,
String>> = OnceLock::new();
+
+ match SENDER.get_or_init(|| {
+ let (sender, receiver) = mpsc::channel::<JindoTeardown>();
+ std::thread::Builder::new()
+ .name("paimon-jindo-teardown".to_string())
+ .spawn(move || {
+ while let Ok(teardown) = receiver.recv() {
+ teardown();
+ }
+ })
+ .map(|_| sender)
+ .map_err(|error| error.to_string())
+ }) {
+ Ok(sender) => Ok(sender),
+ Err(message) => Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "failed to start Jindo teardown worker",
+ )
+ .with_context("message", message)),
+ }
+}
+
+fn enqueue_jindo_teardown(teardown: JindoTeardown) {
+ let Ok(sender) = jindo_teardown_sender() else {
+ // JindoClient::new starts the worker before allocating native
resources.
+ // If it later becomes unavailable, leaking is safer than blocking an
+ // async executor or calling the SDK from an arbitrary drop thread.
+ std::mem::forget(teardown);
+ return;
+ };
+ if let Err(error) = sender.send(teardown) {
+ std::mem::forget(error.0);
+ }
+}
+
+struct JindoApi {
+ create_options: CreateOptions,
+ free_options: FreeOptions,
+ set_option: SetOption,
+ create_store: CreateStore,
+ destroy_store: DestroyStore,
+ free_store: FreeStore,
+ create_handle: CreateHandle,
+ free_handle: FreeHandle,
+ handle_error_code: HandleErrorCode,
+ handle_error_message: HandleErrorMessage,
+ init: Init,
+ get_file_status: GetFileStatus,
+ free_file_status: FreeFileStatus,
+ file_status_path: FileStatusPath,
+ file_status_type: FileStatusType,
+ file_status_size: FileStatusSize,
+ file_status_mtime: FileStatusMtime,
+ list_dir: ListDir,
+ free_list_dir_result: FreeListDirResult,
+ list_dir_result_size: ListDirResultSize,
+ is_list_dir_result_truncated: IsListDirResultTruncated,
+ list_dir_result_next_marker: ListDirResultNextMarker,
+ list_dir_file_status: ListDirFileStatus,
+ list_objects: ListObjects,
+ free_list_objects_result: FreeListObjectsResult,
+ get_object: GetObject,
+ _library: Library,
+}
+
+impl Debug for JindoApi {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.write_str("JindoApi")
+ }
+}
+
+impl JindoApi {
+ fn load(explicit_path: Option<&Path>) -> OpendalResult<Arc<Self>> {
+ // Jindo owns process-wide services, so keep loaded libraries for the
process lifetime.
+ static APIS: OnceLock<Mutex<HashMap<PathBuf, Arc<JindoApi>>>> =
OnceLock::new();
+
+ let mut apis = APIS
+ .get_or_init(|| Mutex::new(HashMap::new()))
+ .lock()
+ .map_err(|_| OpendalError::new(ErrorKind::Unexpected, "Jindo API
lock poisoned"))?;
+ let candidates = library_candidates(explicit_path);
+ let mut errors = Vec::new();
+ for path in candidates {
+ if let Some(api) = apis.get(&path) {
+ return Ok(Arc::clone(api));
+ }
+ match unsafe { Self::load_path(&path) } {
+ Ok(api) => {
+ let api = Arc::new(api);
+ apis.insert(path, Arc::clone(&api));
+ return Ok(api);
+ }
+ Err(error) => errors.push(format!("{}: {error}",
path.display())),
+ }
+ }
+ Err(OpendalError::new(
+ ErrorKind::ConfigInvalid,
+ "unable to load Jindo C SDK; set fs.jindo.library.path",
+ )
+ .with_context("attempts", errors.join("; ")))
+ }
+
+ unsafe fn load_path(path: &Path) -> OpendalResult<Self> {
+ let library = unsafe { Library::new(path) }.map_err(|source| {
+ OpendalError::new(ErrorKind::ConfigInvalid, "failed to load Jindo
library")
+ .set_source(source)
+ })?;
+ macro_rules! load {
+ ($name:literal, $ty:ty) => {{
+ let symbol = unsafe { library.get::<$ty>(concat!($name,
"\0").as_bytes()) }
+ .map_err(|source| {
+ OpendalError::new(
+ ErrorKind::ConfigInvalid,
+ concat!("missing Jindo symbol ", $name),
+ )
+ .set_source(source)
+ })?;
+ *symbol
+ }};
+ }
+ let api = Self {
+ create_options: load!("jdo_createOptions", CreateOptions),
+ free_options: load!("jdo_freeOptions", FreeOptions),
+ set_option: load!("jdo_setOption", SetOption),
+ create_store: load!("jdo_createStore", CreateStore),
+ destroy_store: load!("jdo_destroyStore", DestroyStore),
+ free_store: load!("jdo_freeStore", FreeStore),
+ create_handle: load!("jdo_createHandleCtx1", CreateHandle),
+ free_handle: load!("jdo_freeHandleCtx", FreeHandle),
+ handle_error_code: load!("jdo_getHandleCtxErrorCode",
HandleErrorCode),
+ handle_error_message: load!("jdo_getHandleCtxErrorMsg",
HandleErrorMessage),
+ init: load!("jdo_init", Init),
+ get_file_status: load!("jdo_getFileStatus", GetFileStatus),
+ free_file_status: load!("jdo_freeFileStatus", FreeFileStatus),
+ file_status_path: load!("jdo_getFileStatusPath", FileStatusPath),
+ file_status_type: load!("jdo_getFileStatusType", FileStatusType),
+ file_status_size: load!("jdo_getFileStatusSize", FileStatusSize),
+ file_status_mtime: load!("jdo_getFileStatusMtime",
FileStatusMtime),
+ list_dir: load!("jdo_listDir", ListDir),
+ free_list_dir_result: load!("jdo_freeListDirResult",
FreeListDirResult),
+ list_dir_result_size: load!("jdo_getListDirResultSize",
ListDirResultSize),
+ is_list_dir_result_truncated: load!(
+ "jdo_isListDirResultTruncated",
+ IsListDirResultTruncated
+ ),
+ list_dir_result_next_marker: load!(
+ "jdo_getListDirResultNextMarker",
+ ListDirResultNextMarker
+ ),
+ list_dir_file_status: load!("jdo_getListDirFileStatus",
ListDirFileStatus),
+ list_objects: load!("jdo_listObjects", ListObjects),
+ free_list_objects_result: load!("jdo_freeListObjectsResult",
FreeListObjectsResult),
+ get_object: load!("jdo_getObject", GetObject),
+ _library: library,
+ };
+ Ok(api)
+ }
+}
+
+fn library_candidates(explicit_path: Option<&Path>) -> Vec<PathBuf> {
+ if let Some(path) = explicit_path {
+ return vec![path.to_path_buf()];
+ }
+ if let Some(path) = std::env::var_os("JINDOSDK_LIBRARY_PATH") {
+ return vec![PathBuf::from(path)];
+ }
+ if let Some(home) = std::env::var_os("JINDOSDK_HOME") {
+ let mut path = PathBuf::from(home);
+ path.push("lib");
+ path.push("native");
+ path.push(jindo_c_library_name());
+ return vec![path];
+ }
+ vec![PathBuf::from(jindo_c_library_name())]
+}
+
+fn jindo_c_library_name() -> &'static OsStr {
+ #[cfg(target_os = "macos")]
+ {
+ OsStr::new("libjindosdk_c.dylib")
+ }
+ #[cfg(not(target_os = "macos"))]
+ {
+ OsStr::new("libjindosdk_c.so")
+ }
+}
+
+struct JindoClient {
+ api: Arc<JindoApi>,
+ options: JdoPtr,
+ store: JdoPtr,
+ root: String,
+ initialized: bool,
+}
+
+// SAFETY: Jindo stores support concurrent operations through independent
handles.
+unsafe impl Send for JindoClient {}
+unsafe impl Sync for JindoClient {}
+
+impl Debug for JindoClient {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("JindoClient")
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl JindoClient {
+ fn new(config: &JindoStorageConfig, bucket: &str) -> OpendalResult<Self> {
+ jindo_teardown_sender()?;
+ let api = JindoApi::load(config.library_path.as_deref())?;
+ let root = format!("oss://{bucket}/");
+ let root_c = to_cstring(&root, "OSS root")?;
+ let user_c = to_cstring(&config.user, "Jindo user")?;
+ let properties = config
+ .properties
+ .iter()
+ .map(|(key, value)| {
+ Ok((
+ to_cstring(key, "Jindo option key")?,
+ to_cstring(value, "Jindo option value")?,
+ ))
+ })
+ .collect::<OpendalResult<Vec<_>>>()?;
+
+ let options = unsafe { (api.create_options)() };
+ if options.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo failed to create options",
+ ));
+ }
+ for (key, value) in properties {
+ unsafe { (api.set_option)(options, key.as_ptr(), value.as_ptr()) };
+ }
+
+ let store = unsafe { (api.create_store)(options, root_c.as_ptr()) };
+ if store.is_null() {
+ unsafe { (api.free_options)(options) };
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo failed to create an OSS store",
+ ));
+ }
+ let mut client = Self {
+ api,
+ options,
+ store,
+ root,
+ initialized: false,
+ };
+ client.with_handle(|handle| {
+ unsafe { (client.api.init)(handle, user_c.as_ptr()) };
+ Ok(())
+ })?;
+ client.initialized = true;
+ Ok(client)
+ }
+
+ fn stat(&self, path: &str) -> OpendalResult<Metadata> {
+ let path = to_cstring(&self.full_path(path), "Jindo path")?;
+ self.with_handle(|handle| {
+ let status =
+ unsafe { (self.api.get_file_status)(handle, path.as_ptr(),
std::ptr::null_mut()) };
+ if status.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo returned an empty file status",
+ ));
+ }
+ let _guard = FileStatusGuard {
+ api: Arc::clone(&self.api),
+ status,
+ };
+ self.file_status_metadata(status)
+ })
+ }
+
+ fn list_page(
+ &self,
+ path: &str,
+ recursive: bool,
+ marker: Option<String>,
+ ) -> OpendalResult<JindoListPage> {
+ let path = to_cstring(&self.full_path(path), "Jindo path")?;
+ let first_page = marker.is_none();
+ let handle = unsafe { (self.api.create_handle)(self.store) };
+ if handle.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo failed to create an operation context",
+ ));
+ }
+ let value = (|| {
+ let options = unsafe { (self.api.create_options)() };
+ if options.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo failed to create list options",
+ ));
+ }
+ let _options_guard = OptionsGuard {
+ api: Arc::clone(&self.api),
+ options,
+ };
+ self.set_list_option(options, JDO_LIST_ITERATIVE, "true")?;
+ if let Some(marker) = marker.as_deref() {
+ self.set_list_option(options, JDO_LIST_MARKER, marker)?;
+ }
+
+ let result = self.list_dir(handle, path.as_ptr(), recursive,
options)?;
+ if result.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo returned an empty list result",
+ ));
+ }
+ let _result_guard = ListDirResultGuard {
+ api: Arc::clone(&self.api),
+ result,
+ };
+ let size = unsafe { (self.api.list_dir_result_size)(result) };
+ let mut entries = Vec::with_capacity(size.max(0) as usize);
+ for index in 0..size.max(0) as usize {
+ let status = unsafe { (self.api.list_dir_file_status)(result,
index) };
+ if status.is_null() {
+ continue;
+ }
+ let entry_path = self.file_status_path(status)?;
+ let metadata = self.file_status_metadata(status)?;
+ entries.push(oio::Entry::new(&entry_path, metadata));
+ }
+
+ let next_marker = if unsafe {
(self.api.is_list_dir_result_truncated)(result) } {
+ let next_marker = unsafe {
+ c_string(
+ (self.api.list_dir_result_next_marker)(result),
+ "Jindo list marker",
+ )?
+ };
+ if next_marker.is_empty() || marker.as_deref() ==
Some(next_marker.as_str()) {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo list returned an invalid continuation marker",
+ ));
+ }
+ Some(next_marker)
+ } else {
+ None
+ };
+ Ok(JindoListPage {
+ entries,
+ next_marker,
+ })
+ })();
+ let code = unsafe { (self.api.handle_error_code)(handle) };
+ let error = self.handle_error(handle);
+ unsafe { (self.api.free_handle)(handle) };
+ if first_page && code == JDO_FILE_NOT_FOUND_ERROR {
+ self.verify_list_prefix(&path)?;
+ return Ok(JindoListPage {
+ entries: Vec::new(),
+ next_marker: None,
+ });
+ }
+ error?;
+ value
+ }
+
+ fn verify_list_prefix(&self, path: &CStr) -> OpendalResult<()> {
+ self.with_handle(|handle| {
+ let mut result = std::ptr::null_mut();
+ let mut error = [0 as c_char; JINDO_EXCEPTION_BUFFER_SIZE];
+ let status = unsafe {
+ paimon_jindo_list_objects(
+ self.api.list_objects,
+ handle,
+ path.as_ptr(),
+ std::ptr::null(),
+ std::ptr::null(),
+ 1,
+ std::ptr::null_mut(),
+ &mut result,
+ error.as_mut_ptr(),
+ error.len(),
+ )
+ };
+ if status != 0 {
+ let message = unsafe { CStr::from_ptr(error.as_ptr()) }
+ .to_string_lossy()
+ .into_owned();
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo list failed with a C++ exception",
+ )
+ .with_context("message", message));
+ }
+ if result.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo returned an empty list result",
+ ));
+ }
+ unsafe { (self.api.free_list_objects_result)(result) };
+ Ok(())
+ })
+ }
+
+ fn set_list_option(&self, options: JdoPtr, key: &str, value: &str) ->
OpendalResult<()> {
+ let key = to_cstring(key, "Jindo list option")?;
+ let value = to_cstring(value, "Jindo list option")?;
+ unsafe { (self.api.set_option)(options, key.as_ptr(), value.as_ptr())
};
+ Ok(())
+ }
+
+ fn list_dir(
+ &self,
+ handle: JdoPtr,
+ path: *const c_char,
+ recursive: bool,
+ options: JdoPtr,
+ ) -> OpendalResult<JdoPtr> {
+ let mut result = std::ptr::null_mut();
+ let mut error = [0 as c_char; JINDO_EXCEPTION_BUFFER_SIZE];
+ let status = unsafe {
+ paimon_jindo_list_dir(
+ self.api.list_dir,
+ handle,
+ path,
+ recursive,
+ options,
+ &mut result,
+ error.as_mut_ptr(),
+ error.len(),
+ )
+ };
+ if status != 0 {
+ let message = unsafe { CStr::from_ptr(error.as_ptr()) }
+ .to_string_lossy()
+ .into_owned();
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo list failed with a C++ exception",
+ )
+ .with_context("message", message));
+ }
+ Ok(result)
+ }
+
+ fn file_status_path(&self, status: JdoPtr) -> OpendalResult<String> {
+ let full_path = unsafe { c_string((self.api.file_status_path)(status),
"Jindo path")? };
+ Ok(relative_list_path(
+ &self.root,
+ &full_path,
+ unsafe { (self.api.file_status_type)(status) } == 1,
+ ))
+ }
+
+ fn file_status_metadata(&self, status: JdoPtr) -> OpendalResult<Metadata> {
+ let mode = match unsafe { (self.api.file_status_type)(status) } {
+ 1 => EntryMode::DIR,
+ 2 => EntryMode::FILE,
+ _ => EntryMode::Unknown,
+ };
+ let size = unsafe { (self.api.file_status_size)(status) };
+ let mtime = unsafe { (self.api.file_status_mtime)(status) };
+ let mut metadata = Metadata::new(mode).with_content_length(size.max(0)
as u64);
+ if mtime > 0 {
+ metadata.set_last_modified(Timestamp::from_millisecond(mtime)?);
+ }
+ Ok(metadata)
+ }
+
+ fn read(&self, path: &str, range: BytesRange) -> OpendalResult<Vec<u8>> {
+ let object_size = match range {
+ BytesRange::Range { size: Some(_), .. } => None,
+ _ => Some(self.stat(path)?.content_length()),
+ };
+ let (offset, size) = match range {
+ BytesRange::Range { offset, size } => {
+ let size = size.unwrap_or_else(||
object_size.unwrap_or(0).saturating_sub(offset));
+ (offset, size)
+ }
+ BytesRange::Suffix { size } => {
+ let object_size = object_size.unwrap_or(0);
+ (object_size.saturating_sub(size), size.min(object_size))
+ }
+ };
+ if size == 0 {
+ return Ok(Vec::new());
+ }
+ let offset = i64::try_from(offset).map_err(|source| {
+ OpendalError::new(
+ ErrorKind::RangeNotSatisfied,
+ "Jindo read offset is too large",
+ )
+ .set_source(source)
+ })?;
+ let size_i64 = i64::try_from(size).map_err(|source| {
+ OpendalError::new(ErrorKind::RangeNotSatisfied, "Jindo read size
is too large")
+ .set_source(source)
+ })?;
+ let size_usize = usize::try_from(size).map_err(|source| {
+ OpendalError::new(ErrorKind::RangeNotSatisfied, "Jindo read size
is too large")
+ .set_source(source)
+ })?;
+ let path = to_cstring(&self.full_path(path), "Jindo path")?;
+ let mut buffer = vec![0_u8; size_usize];
+ let read = self.with_handle(|handle| {
+ Ok(unsafe {
+ (self.api.get_object)(
+ handle,
+ path.as_ptr(),
+ buffer.as_mut_ptr().cast(),
+ size_i64,
+ offset,
+ std::ptr::null_mut(),
+ )
+ })
+ })?;
+ if read <= 0 {
+ return Ok(Vec::new());
+ }
+ buffer.truncate(read as usize);
+ Ok(buffer)
+ }
+
+ fn with_handle<T>(&self, call: impl FnOnce(JdoPtr) -> OpendalResult<T>) ->
OpendalResult<T> {
+ let handle = unsafe { (self.api.create_handle)(self.store) };
+ if handle.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ "Jindo failed to create an operation context",
+ ));
+ }
+ let value = call(handle);
+ let error = self.handle_error(handle);
+ unsafe { (self.api.free_handle)(handle) };
+ error?;
+ value
+ }
+
+ fn handle_error(&self, handle: JdoPtr) -> OpendalResult<()> {
+ let code = unsafe { (self.api.handle_error_code)(handle) };
+ if code == 0 {
+ return Ok(());
+ }
+ let message = unsafe {
+ let ptr = (self.api.handle_error_message)(handle);
+ if ptr.is_null() {
+ String::new()
+ } else {
+ CStr::from_ptr(ptr).to_string_lossy().into_owned()
+ }
+ };
+ let kind = match code {
+ JDO_FILE_NOT_FOUND_ERROR | JDO_REST_HTTP_404_ERROR =>
ErrorKind::NotFound,
+ JDO_REST_HTTP_403_ERROR => ErrorKind::PermissionDenied,
+ JDO_REST_HTTP_503_ERROR => ErrorKind::RateLimited,
+ _ => ErrorKind::Unexpected,
+ };
+ let error = OpendalError::new(kind, "Jindo operation failed")
+ .with_context("code", code.to_string())
+ .with_context("message", message);
+ if matches!(code, 2000..=2003 | 6500 | 6502 | 6503) {
+ Err(error.set_temporary())
+ } else {
+ Err(error)
+ }
+ }
+
+ fn full_path(&self, path: &str) -> String {
+ format!("{}{}", self.root, path.trim_start_matches('/'))
+ }
+}
+
+impl Drop for JindoClient {
+ fn drop(&mut self) {
+ let api = Arc::clone(&self.api);
+ let options = std::mem::replace(&mut self.options,
std::ptr::null_mut()) as usize;
+ let store = std::mem::replace(&mut self.store, std::ptr::null_mut())
as usize;
+ let initialized = std::mem::replace(&mut self.initialized, false);
+
+ enqueue_jindo_teardown(Box::new(move || {
+ let options = options as JdoPtr;
+ let store = store as JdoPtr;
+ if initialized {
+ let handle = unsafe { (api.create_handle)(store) };
+ if !handle.is_null() {
+ unsafe {
+ (api.destroy_store)(store);
+ (api.free_handle)(handle);
+ }
+ }
+ }
+ unsafe {
+ (api.free_store)(store);
+ (api.free_options)(options);
+ }
+ }));
+ }
+}
+
+struct FileStatusGuard {
+ api: Arc<JindoApi>,
+ status: JdoPtr,
+}
+
+impl Drop for FileStatusGuard {
+ fn drop(&mut self) {
+ unsafe { (self.api.free_file_status)(self.status) };
+ }
+}
+
+struct OptionsGuard {
+ api: Arc<JindoApi>,
+ options: JdoPtr,
+}
+
+impl Drop for OptionsGuard {
+ fn drop(&mut self) {
+ unsafe { (self.api.free_options)(self.options) };
+ }
+}
+
+struct ListDirResultGuard {
+ api: Arc<JindoApi>,
+ result: JdoPtr,
+}
+
+impl Drop for ListDirResultGuard {
+ fn drop(&mut self) {
+ unsafe { (self.api.free_list_dir_result)(self.result) };
+ }
+}
+
+unsafe fn c_string(value: *const c_char, name: &str) -> OpendalResult<String> {
+ if value.is_null() {
+ return Err(OpendalError::new(
+ ErrorKind::Unexpected,
+ format!("Jindo returned an empty {name}"),
+ ));
+ }
+ Ok(unsafe { CStr::from_ptr(value) }
+ .to_string_lossy()
+ .into_owned())
+}
+
+fn relative_list_path(root: &str, full_path: &str, directory: bool) -> String {
+ let mut path = full_path
+ .strip_prefix(root)
+ .unwrap_or(full_path)
+ .trim_start_matches('/')
+ .to_string();
+ if directory && !path.ends_with('/') {
+ path.push('/');
+ }
+ path
+}
+
+fn to_cstring(value: &str, name: &str) -> OpendalResult<CString> {
+ CString::new(value).map_err(|source| {
+ OpendalError::new(
+ ErrorKind::ConfigInvalid,
+ format!("{name} contains a NUL byte"),
+ )
+ .set_source(source)
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+ #[cfg(target_os = "linux")]
+ use axum::body::Body;
+ #[cfg(target_os = "linux")]
+ use axum::extract::State;
+ #[cfg(target_os = "linux")]
+ use axum::http::{HeaderMap, Method, Response, StatusCode, Uri};
+ #[cfg(target_os = "linux")]
+ use axum::Router;
+
+ #[test]
+ fn test_use_jindo() {
+ assert!(!use_jindo(&HashMap::new()).unwrap());
+ assert!(use_jindo(&HashMap::from([(
+ OSS_IMPL.to_string(),
+ "JINDO".to_string()
+ )]))
+ .unwrap());
+ assert!(!use_jindo(&HashMap::from([(
+ OSS_IMPL.to_string(),
+ "legacy".to_string()
+ )]))
+ .unwrap());
+ assert!(use_jindo(&HashMap::from([(
+ OSS_IMPL.to_string(),
+ "unknown".to_string()
+ )]))
+ .is_err());
+ }
+
+ #[test]
+ fn test_parse_config() {
+ let config = jindo_config_parse(HashMap::from([
+ (OSS_IMPL.to_string(), JINDO_IMPL.to_string()),
+ (
+ JINDO_LIBRARY_PATH.to_string(),
+ "/path/to/libjindosdk_c.so".to_string(),
+ ),
+ (
+ "fs.oss.endpoint".to_string(),
+ "https://oss-cn-hangzhou.aliyuncs.com".to_string(),
+ ),
+ ]))
+ .unwrap();
+ assert_eq!(
+ config.library_path.as_deref(),
+ Some(Path::new("/path/to/libjindosdk_c.so"))
+ );
+ assert_eq!(config.user, DEFAULT_JINDO_USER);
+ assert_eq!(
+ config.max_concurrent_reads,
+ DEFAULT_JINDO_MAX_CONCURRENT_READS
+ );
+ assert_eq!(
+ config.properties.get("fs.oss.endpoint").map(String::as_str),
+ Some("oss-cn-hangzhou.aliyuncs.com")
+ );
+ assert_eq!(
+ config.properties.get(OSS_HTTPS_ENABLE).map(String::as_str),
+ Some("true")
+ );
+ assert_eq!(
+ config.properties.get("logger.appender").map(String::as_str),
+ Some("file")
+ );
+ assert!(!config.properties.contains_key(OSS_IMPL));
+ assert!(!config.properties.contains_key(JINDO_MAX_CONCURRENT_READS));
+ }
+
+ #[test]
+ fn test_parse_max_concurrent_reads() {
+ let config = jindo_config_parse(HashMap::from([(
+ JINDO_MAX_CONCURRENT_READS.to_string(),
+ "6".to_string(),
+ )]))
+ .unwrap();
+ assert_eq!(config.max_concurrent_reads, 6);
+ assert!(!config.properties.contains_key(JINDO_MAX_CONCURRENT_READS));
+
+ let client = LazyJindoClient::new(config, "bucket".to_string());
+ assert_eq!(client.read_gate.available_permits(), 6);
+
+ for invalid in ["0", "-1", "invalid", "18446744073709551615"] {
+ assert!(jindo_config_parse(HashMap::from([(
+ JINDO_MAX_CONCURRENT_READS.to_string(),
+ invalid.to_string(),
+ )]))
+ .is_err());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_read_gate_limits_blocking_calls() {
+ let gate = Arc::new(Semaphore::new(2));
+ let active = Arc::new(AtomicUsize::new(0));
+ let peak = Arc::new(AtomicUsize::new(0));
+ let mut tasks = Vec::new();
+ for _ in 0..6 {
+ let gate = Arc::clone(&gate);
+ let active = Arc::clone(&active);
+ let peak = Arc::clone(&peak);
+ tasks.push(tokio::spawn(run_blocking_with_gate(gate, move || {
+ let current = active.fetch_add(1, Ordering::SeqCst) + 1;
+ peak.fetch_max(current, Ordering::SeqCst);
+ std::thread::sleep(std::time::Duration::from_millis(20));
+ active.fetch_sub(1, Ordering::SeqCst);
+ Ok(())
+ })));
+ }
+ for task in tasks {
+ task.await.unwrap().unwrap();
+ }
+ assert_eq!(peak.load(Ordering::SeqCst), 2);
+ assert_eq!(gate.available_permits(), 2);
+ }
+
+ #[tokio::test]
+ async fn test_cancelled_waiter_does_not_release_in_flight_permit() {
+ let gate = Arc::new(Semaphore::new(1));
+ let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+ let (release_tx, release_rx) = std::sync::mpsc::channel();
+ let task = tokio::spawn(run_blocking_with_gate(Arc::clone(&gate), move
|| {
+ started_tx.send(()).unwrap();
+ release_rx.recv().unwrap();
+ Ok(())
+ }));
+ started_rx.await.unwrap();
+ assert_eq!(gate.available_permits(), 0);
+
+ task.abort();
+ tokio::task::yield_now().await;
+ assert_eq!(gate.available_permits(), 0);
+
+ release_tx.send(()).unwrap();
+ tokio::time::timeout(std::time::Duration::from_secs(1), async {
+ while gate.available_permits() == 0 {
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .unwrap();
+ assert_eq!(gate.available_permits(), 1);
+ }
+
+ #[tokio::test(flavor = "current_thread")]
+ async fn test_teardown_does_not_block_async_runtime() {
+ let (returned_tx, returned_rx) = std::sync::mpsc::channel();
+ let (release_tx, release_rx) = std::sync::mpsc::channel();
+ let (done_tx, done_rx) = tokio::sync::oneshot::channel();
+ let returned_before_timeout = Arc::new(AtomicBool::new(false));
+ let observed = Arc::clone(&returned_before_timeout);
+ let watchdog = std::thread::spawn(move || {
+ observed.store(
+ returned_rx
+ .recv_timeout(std::time::Duration::from_millis(500))
+ .is_ok(),
+ Ordering::SeqCst,
+ );
+ release_tx.send(()).unwrap();
+ });
+
+ enqueue_jindo_teardown(Box::new(move || {
+ release_rx.recv().unwrap();
+ done_tx.send(()).unwrap();
+ }));
+ let _ = returned_tx.send(());
+
+ tokio::time::timeout(std::time::Duration::from_secs(2), done_rx)
+ .await
+ .unwrap()
+ .unwrap();
+ watchdog.join().unwrap();
+ assert!(returned_before_timeout.load(Ordering::SeqCst));
+ }
+
+ #[test]
+ fn test_parse_config_preserves_endpoint_protocol() {
+ for (endpoint, configured, expected) in [
+ ("http://localhost:9000";, None, Some("false")),
+ ("HTTPS://localhost:9000", None, Some("true")),
+ ("https://localhost:9000";, Some("false"), Some("false")),
+ ("localhost:9000", Some("true"), Some("true")),
+ ("localhost:9000", None, None),
+ ] {
+ let mut props = HashMap::from([(OSS_ENDPOINT.to_string(),
endpoint.to_string())]);
+ if let Some(configured) = configured {
+ props.insert(OSS_HTTPS_ENABLE.to_string(),
configured.to_string());
+ }
+ let config = jindo_config_parse(props).unwrap();
+ assert_eq!(
+ config.properties.get(OSS_ENDPOINT).map(String::as_str),
+ Some(
+ endpoint
+ .split_once("://")
+ .map_or(endpoint, |(_, host)| host)
+ )
+ );
+ assert_eq!(
+ config.properties.get(OSS_HTTPS_ENABLE).map(String::as_str),
+ expected
+ );
+ }
+ }
+
+ #[test]
+ fn test_explicit_library_path_is_exclusive() {
+ let path = Path::new("/custom/libjindosdk_c.so");
+ assert_eq!(library_candidates(Some(path)), vec![path]);
+ }
+
+ #[test]
+ fn test_build_does_not_load_sdk() {
+ let config = JindoStorageConfig {
+ library_path: Some(PathBuf::from("/missing/libjindosdk_c.so")),
+ user: DEFAULT_JINDO_USER.to_string(),
+ max_concurrent_reads: DEFAULT_JINDO_MAX_CONCURRENT_READS,
+ properties: HashMap::new(),
+ };
+ let operator = jindo_config_build(&config, "bucket").unwrap();
+ assert!(operator.info().capability().list);
+ }
+
+ #[test]
+ fn test_relative_list_path() {
+ assert_eq!(
+ relative_list_path(
+ "oss://bucket/",
+ "oss://bucket/table/snapshot/snapshot-1",
+ false
+ ),
+ "table/snapshot/snapshot-1"
+ );
+ assert_eq!(
+ relative_list_path("oss://bucket/", "table/snapshot", true),
+ "table/snapshot/"
+ );
+ }
+
+ #[cfg(target_os = "linux")]
+ #[derive(Clone, Default)]
+ struct OssMockState {
+ stat_requests: Arc<AtomicUsize>,
+ list_page_requests: Arc<AtomicUsize>,
+ range_requests: Arc<AtomicUsize>,
+ unavailable_requests: Arc<AtomicUsize>,
+ }
+
+ #[cfg(target_os = "linux")]
+ async fn oss_mock(
+ State(state): State<OssMockState>,
+ method: Method,
+ uri: Uri,
+ headers: HeaderMap,
+ ) -> Response<Body> {
+ const DATA: &[u8] = b"0123456789";
+ if method == Method::HEAD {
+ if uri.path().ends_with("/collision") {
+ return Response::builder()
+ .status(StatusCode::OK)
+ .header("content-length", DATA.len())
+ .header("etag", "\"jindo-test-collision\"")
+ .header("last-modified", "Mon, 01 Jan 2024 00:00:00 GMT")
+ .header("x-oss-object-type", "Normal")
+ .header("x-oss-storage-class", "Standard")
+ .header("x-oss-request-id", "jindo-test-collision")
+ .body(Body::empty())
+ .unwrap();
+ }
+ if !uri.path().contains("data.bin") {
+ return Response::builder()
+ .status(StatusCode::NOT_FOUND)
+ .header("x-oss-request-id", "jindo-test-not-found")
+ .body(Body::empty())
+ .unwrap();
+ }
+ state.stat_requests.fetch_add(1, Ordering::SeqCst);
+ return Response::builder()
+ .status(StatusCode::OK)
+ .header("content-length", DATA.len())
+ .header("etag", "\"jindo-test-etag\"")
+ .header("last-modified", "Mon, 01 Jan 2024 00:00:00 GMT")
+ .header("x-oss-object-type", "Normal")
+ .header("x-oss-storage-class", "Standard")
+ .header("x-oss-request-id", "jindo-test-stat")
+ .body(Body::empty())
+ .unwrap();
+ }
+
+ if method == Method::GET {
+ if uri.path().starts_with("/missing-bucket/") {
+ return Response::builder()
+ .status(StatusCode::NOT_FOUND)
+ .header("content-type", "application/xml")
+ .header("x-oss-request-id", "jindo-test-bucket-not-found")
+ .body(Body::from(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<Error><Code>NoSuchBucket</Code><Message>missing bucket</Message>
+<RequestId>jindo-test-bucket-not-found</RequestId></Error>"#,
+ ))
+ .unwrap();
+ }
+ if let Some(range) = headers.get("range") {
+ if uri.path().contains("unavailable.bin") {
+ state.unavailable_requests.fetch_add(1, Ordering::SeqCst);
+ return Response::builder()
+ .status(StatusCode::SERVICE_UNAVAILABLE)
+ .header("content-type", "application/xml")
+ .header("x-oss-request-id", "jindo-test-unavailable")
+ .body(Body::from(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<Error><Code>ServiceUnavailable</Code><Message>try later</Message>
+<RequestId>jindo-test-unavailable</RequestId></Error>"#,
+ ))
+ .unwrap();
+ }
+ state.range_requests.fetch_add(1, Ordering::SeqCst);
+ assert_eq!(range, "bytes=2-5");
+ return Response::builder()
+ .status(StatusCode::PARTIAL_CONTENT)
+ .header("content-length", 4)
+ .header("content-range", "bytes 2-5/10")
+ .header("etag", "\"jindo-test-etag\"")
+ .header("x-oss-request-id", "jindo-test-range")
+ .body(Body::from(&DATA[2..6]))
+ .unwrap();
+ }
+
+ let query = uri.query().unwrap_or_default();
+ let parameters = url::form_urlencoded::parse(query.as_bytes())
+ .into_owned()
+ .collect::<HashMap<_, _>>();
+ let prefix = parameters.get("prefix").cloned().unwrap_or_default();
+ let marker = parameters.get("marker").cloned();
+ if prefix.starts_with("forbidden/") {
+ return Response::builder()
+ .status(StatusCode::FORBIDDEN)
+ .header("content-type", "application/xml")
+ .header("x-oss-request-id", "jindo-test-forbidden")
+ .body(Body::from(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<Error><Code>AccessDenied</Code><Message>denied</Message>
+<RequestId>jindo-test-forbidden</RequestId></Error>"#,
+ ))
+ .unwrap();
+ }
+ if prefix.starts_with("broken-page/") && marker.is_some() {
+ return Response::builder()
+ .status(StatusCode::NOT_FOUND)
+ .header("content-type", "application/xml")
+ .header("x-oss-request-id", "jindo-test-page-not-found")
+ .body(Body::from(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<Error><Code>NoSuchKey</Code><Message>missing page</Message>
+<RequestId>jindo-test-page-not-found</RequestId></Error>"#,
+ ))
+ .unwrap();
+ }
+ if prefix.contains("missing.bin")
+ || prefix.starts_with("empty/")
+ || prefix.starts_with("empty-table/")
+ {
+ let body = format!(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://doc.oss-cn-hangzhou.aliyuncs.com";>
+ <Name>jindo-test-bucket</Name>
+ <Prefix>{prefix}</Prefix>
+ <MaxKeys>1</MaxKeys>
+ <IsTruncated>false</IsTruncated>
+</ListBucketResult>"#
+ );
+ return Response::builder()
+ .status(StatusCode::OK)
+ .header("content-type", "application/xml")
+ .header("content-length", body.len())
+ .header("x-oss-request-id", "jindo-test-empty-list")
+ .body(Body::from(body))
+ .unwrap();
+ }
+ let probe = parameters.get("max-keys").map(String::as_str) ==
Some("1");
+ let (key, truncated, next_marker) = if probe {
+ let key = if prefix.starts_with("broken-page/") {
+ "broken-page/a.bin"
+ } else {
+ "objects/a.bin"
+ };
+ (key, "false", "")
+ } else if prefix.starts_with("broken-page/") {
+ ("broken-page/a.bin", "true", "broken-page/a.bin")
+ } else if marker.as_deref() == Some("objects/a.bin") {
+ ("objects/b.bin", "false", "")
+ } else {
+ ("objects/a.bin", "true", "objects/a.bin")
+ };
+ if !probe {
+ state.list_page_requests.fetch_add(1, Ordering::SeqCst);
+ }
+ let body = format!(
+ r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://doc.oss-cn-hangzhou.aliyuncs.com";>
+ <Name>jindo-test-bucket</Name>
+ <Prefix>{prefix}</Prefix>
+ <Marker>{}</Marker>
+ <MaxKeys>1</MaxKeys>
+ <IsTruncated>{truncated}</IsTruncated>
+ <NextMarker>{next_marker}</NextMarker>
+ <Contents>
+ <Key>{key}</Key>
+ <LastModified>2024-01-01T00:00:00.000Z</LastModified>
+ <ETag>"jindo-test-etag"</ETag>
+ <Type>Normal</Type>
+ <Size>10</Size>
+ <StorageClass>Standard</StorageClass>
+ </Contents>
+</ListBucketResult>"#,
+ marker.unwrap_or_default()
+ );
+ return Response::builder()
+ .status(StatusCode::OK)
+ .header("content-type", "application/xml")
+ .header("content-length", body.len())
+ .header("x-oss-request-id", "jindo-test-list")
+ .body(Body::from(body))
+ .unwrap();
+ }
+
+ Response::builder()
+ .status(StatusCode::METHOD_NOT_ALLOWED)
+ .body(Body::empty())
+ .unwrap()
+ }
+
+ #[cfg(target_os = "linux")]
+ #[tokio::test]
+ #[ignore = "requires pyjindosdk"]
+ async fn test_real_sdk_io_and_failure_mapping() {
+ let library_path = std::env::var("PAIMON_JINDO_SDK_LIBRARY")
+ .expect("PAIMON_JINDO_SDK_LIBRARY must point to the real Jindo
SDK");
+ let state = OssMockState::default();
+ let app = Router::new().fallback(oss_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 properties = HashMap::from([
+ (OSS_IMPL.to_string(), JINDO_IMPL.to_string()),
+ (JINDO_LIBRARY_PATH.to_string(), library_path),
+ (OSS_ENDPOINT.to_string(), format!("http://{address}";)),
+ (
+ "fs.oss.accessKeyId".to_string(),
+ "test-access-key".to_string(),
+ ),
+ (
+ "fs.oss.accessKeySecret".to_string(),
+ "test-access-secret".to_string(),
+ ),
+ (
+ "fs.oss.second.level.domain.enable".to_string(),
+ "true".to_string(),
+ ),
+ ("fs.oss.retry.count".to_string(), "0".to_string()),
+ ]);
+ let config = jindo_config_parse(properties.clone()).unwrap();
+ let operator = jindo_config_build(&config,
"jindo-test-bucket").unwrap();
+
+ let metadata = operator.stat("objects/data.bin").await.unwrap();
+ assert!(metadata.is_file());
+
+ let entries = operator.list("objects/").await.unwrap();
+ assert_eq!(
+ entries.iter().map(|entry| entry.path()).collect::<Vec<_>>(),
+ ["objects/a.bin", "objects/b.bin"]
+ );
+ assert!(operator.list("empty/").await.unwrap().is_empty());
+
+ let missing_bucket_operator = jindo_config_build(&config,
"missing-bucket").unwrap();
+ let missing_bucket =
missing_bucket_operator.list("empty/").await.unwrap_err();
+ assert_eq!(missing_bucket.kind(), ErrorKind::NotFound);
+
+ let later_not_found = operator.list("broken-page/").await.unwrap_err();
+ assert_eq!(later_not_found.kind(), ErrorKind::NotFound);
+
+ let forbidden = operator.list("forbidden/").await.unwrap_err();
+ assert_eq!(forbidden.kind(), ErrorKind::PermissionDenied);
+
+ let file_io = crate::io::FileIOBuilder::new("oss")
+ .with_props(properties)
+ .build()
+ .unwrap();
+ let snapshots = crate::table::SnapshotManager::new(
+ file_io,
+ "oss://jindo-test-bucket/empty-table".to_string(),
+ );
+ assert_eq!(snapshots.get_latest_snapshot_id().await.unwrap(), None);
+ assert_eq!(snapshots.earliest_snapshot_id().await.unwrap(), None);
+ let branch_snapshots = snapshots.with_branch("dev");
+ assert_eq!(
+ branch_snapshots.get_latest_snapshot_id().await.unwrap(),
+ None
+ );
+ assert_eq!(branch_snapshots.earliest_snapshot_id().await.unwrap(),
None);
+
+ let reader = operator.reader("objects/data.bin").await.unwrap();
+ assert_eq!(reader.read(2..6).await.unwrap().to_bytes(), &b"2345"[..]);
+
+ let not_found =
operator.stat("objects/missing.bin").await.unwrap_err();
+ assert_eq!(not_found.kind(), ErrorKind::NotFound);
+
+ let reader = operator.reader("objects/unavailable.bin").await.unwrap();
+ let unavailable = reader.read(0..4).await.unwrap_err();
+ assert_eq!(unavailable.kind(), ErrorKind::RateLimited);
+ assert!(unavailable.is_temporary());
+
+ let collision = operator.list("collision/").await.unwrap_err();
+ assert_eq!(collision.kind(), ErrorKind::Unexpected);
+ assert!(collision
+ .to_string()
+ .contains("Jindo list failed with a C++ exception"));
+
+ assert_eq!(state.stat_requests.load(Ordering::SeqCst), 1);
+ assert_eq!(state.list_page_requests.load(Ordering::SeqCst), 3);
+ assert_eq!(state.range_requests.load(Ordering::SeqCst), 1);
+ assert_eq!(state.unavailable_requests.load(Ordering::SeqCst), 1);
+ server.abort();
+ }
+}
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 0a82fb3b..5297c27a 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -44,6 +44,7 @@ Available storage features:
| `storage-memory` | In-memory |
| `storage-s3` | Amazon S3 |
| `storage-oss` | Alibaba Cloud OSS|
+| `storage-jindo` | Alibaba Cloud OSS through JindoSDK |
| `storage-cos` | Tencent Cloud COS |
| `storage-azdls` | Azure Data Lake Storage Gen2 |
| `storage-obs` | Huawei Cloud OBS |
@@ -302,6 +303,15 @@ options.set(CatalogOptions::DLF_TOKEN_PATH,
"/path/to/token.json");
The JSON uses `AccessKeyId`, `AccessKeySecret`, optional `SecurityToken`, and
optional `Expiration`. Expiring credentials are reloaded before expiration.
+With `storage-jindo`, set `fs.oss.impl` to `jindo` to use an installed JindoSDK
+for OSS scan planning. Set `fs.jindo.library.path` unless the library is
+available through `JINDOSDK_HOME` or `JINDOSDK_LIBRARY_PATH`. The Python
binding
+also discovers it from an installed `pyjindosdk` package. This initial
+integration supports object stat, reads, and listings, but not writes, deletes,
+or copies. Jindo reads are limited to 8 concurrent requests per OSS operator by
+default. Set `fs.jindo.max.concurrent.reads` to a positive integer to tune this
+limit for the available network and JindoSDK connection capacity.
+
Supported metastore types:
| Metastore Type | Description |
