feat(storage): support Jindo for OSS scan planning

bindings/python/src/context.rs

@@ -83,7 +83,65 @@ fn pyarrow_compatible_batch(batch: &RecordBatch) -> arrow::error::Result<RecordB
     RecordBatch::try_new_with_options(schema, columns, &options)
 }
 
-fn build_paimon_catalog(catalog_options: HashMap<String, String>) -> PyResult<Arc<dyn Catalog>> {
+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 build_paimon_catalog(
+    mut catalog_options: HashMap<String, String>,
+) -> PyResult<Arc<dyn Catalog>> {
+    let use_jindo = catalog_options
+        .get(OSS_IMPL)
+        .is_some_and(|value| value.eq_ignore_ascii_case("jindo"));
+    let library_path = std::env::var_os(JINDOSDK_LIBRARY_PATH);
+    let home = std::env::var_os(JINDOSDK_HOME);
+    if use_jindo
+        && should_discover_pyjindo_library(
+            &catalog_options,
+            library_path.as_deref(),
+            home.as_deref(),
+        )
+    {
+        if let Some(path) = discover_pyjindo_library() {
+            catalog_options.insert(JINDO_LIBRARY_PATH.to_string(), path.display().to_string());
+        }
+    }
     let rt = runtime();
     rt.block_on(async {
         let options = Options::from_map(catalog_options);
@@ -92,6 +150,49 @@ 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));
+    }
+}
+
 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)? {

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]
 url = "2.5.2"
@@ -142,6 +144,9 @@ 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"] }
 rand = "0.8.5"

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

crates/paimon/src/io/jindo_ffi.cc

@@ -0,0 +1,64 @@
+// 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*);
+
+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;
+}

crates/paimon/src/io/mod.rs

@@ -62,6 +62,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")]

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() {