feat: Support hdfs with OpenDAL

Author: wForget

native/core/Cargo.toml

@@ -74,6 +74,9 @@ aws-credential-types = { workspace = true }
 parking_lot = "0.12.3"
 datafusion-comet-objectstore-hdfs = { path = "../hdfs", optional = true, default-features = false, features = ["hdfs"] }
 
+object_store_opendal = {version = "0.54.0", optional = true}
+opendal = { version ="0.54.0", optional = true, features = ["services-hdfs"] }
+
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs = "0.17.0"
 
@@ -89,6 +92,7 @@ datafusion-functions-nested = { version = "49.0.2" }
 [features]
 default = []
 hdfs = ["datafusion-comet-objectstore-hdfs"]
+hdfs-opendal = ["opendal", "object_store_opendal"]
 jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"]
 
 # exclude optional packages from cargo machete verifications

native/core/src/parquet/parquet_support.rs

@@ -38,8 +38,13 @@ use datafusion_comet_spark_expr::EvalMode;
 use object_store::path::Path;
 use object_store::{parse_url, ObjectStore};
 use std::collections::HashMap;
+use std::path::PathBuf;
 use std::time::Duration;
-use std::{fmt::Debug, hash::Hash, sync::Arc};
+use std::{
+    fmt::{Debug, Write},
+    hash::Hash,
+    sync::Arc,
+};
 use url::Url;
 
 use super::objectstore;
@@ -352,7 +357,42 @@ fn parse_hdfs_url(url: &Url) -> Result<(Box<dyn ObjectStore>, Path), object_stor
     }
 }
 
-#[cfg(not(feature = "hdfs"))]
+#[cfg(feature = "hdfs-opendal")]
+fn parse_hdfs_url(url: &Url) -> Result<(Box<dyn ObjectStore>, Path), object_store::Error> {
+    let name_node = get_name_node_uri(url)?;
+    let builder = opendal::services::Hdfs::default().name_node(&name_node);
+
+    let op = opendal::Operator::new(builder)
+        .map_err(|error| object_store::Error::Generic {
+            store: "hdfs-opendal",
+            source: error.into(),
+        })?
+        .finish();
+    let store = object_store_opendal::OpendalStore::new(op);
+    let path = Path::parse(url.path())?;
+    Ok((Box::new(store), path))
+}
+
+fn get_name_node_uri(url: &Url) -> Result<String, object_store::Error> {
+    if let Some(host) = url.host() {
+        let schema = url.scheme();
+        let mut uri_builder = String::new();
+        write!(&mut uri_builder, "{schema}://{host}").unwrap();
+
+        if let Some(port) = url.port() {
+            write!(&mut uri_builder, ":{port}").unwrap();
+        }
+        Ok(uri_builder)
+    } else {
+        Err(object_store::Error::InvalidPath {
+            source: object_store::path::Error::InvalidPath {
+                path: PathBuf::from(url.as_str()),
+            },
+        })
+    }
+}
+
+#[cfg(all(not(feature = "hdfs"), not(feature = "hdfs-opendal")))]
 fn parse_hdfs_url(_url: &Url) -> Result<(Box<dyn ObjectStore>, Path), object_store::Error> {
     Err(object_store::Error::Generic {
         store: "HadoopFileSystem",