feat(storage): support Jindo for OSS scan planning

.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

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)?;

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(

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})
 

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"] }

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");
+    }
+}