feat: use opendal to support hdfs native

Author: fMeow

Cargo.lock

@@ -150,7 +150,6 @@ moka = { version = "0.12", features = ["future", "sync"] }
 ndarray = { version = "0.16.1", features = ["matrixmultiply-threading"] }
 num-traits = "0.2"
 object_store = { version = "0.12.3" }
-hdfs-native-object-store = "0.15"
 opendal = { version = "0.54" }
 object_store_opendal = { version = "0.54" }
 pin-project = "1.0"

Cargo.toml

@@ -16,7 +16,6 @@ rust-version.workspace = true
 object_store = { workspace = true }
 opendal = { workspace = true, optional = true }
 object_store_opendal = { workspace = true, optional = true }
-hdfs-native-object-store = { workspace = true, optional = true }
 lance-arrow.workspace = true
 lance-core.workspace = true
 lance-namespace.workspace = true
@@ -64,14 +63,14 @@ name = "scheduler"
 harness = false
 
 [features]
-default = ["aws", "azure", "gcp"]
+default = ["aws", "azure", "gcp", "hdfs"]
 gcs-test = []
 gcp = ["object_store/gcp", "dep:opendal", "opendal/services-gcs", "dep:object_store_opendal"]
 aws = ["object_store/aws", "dep:aws-config", "dep:aws-credential-types", "dep:opendal", "opendal/services-s3", "dep:object_store_opendal"]
 azure = ["object_store/azure", "dep:opendal", "opendal/services-azblob", "dep:object_store_opendal"]
 oss = ["dep:opendal", "opendal/services-oss", "dep:object_store_opendal"]
 huggingface = ["dep:opendal", "opendal/services-huggingface", "dep:object_store_opendal"]
-hdfs = ["dep:hdfs-native-object-store"]
+hdfs = ["dep:opendal", "opendal/services-hdfs-native", "dep:object_store_opendal" ]
 test-util = []
 
 [lints]

rust/lance-io/Cargo.toml

@@ -3,15 +3,17 @@
 
 use std::sync::Arc;
 
-use crate::object_store::{
-    ObjectStore, ObjectStoreParams, ObjectStoreProvider, StorageOptions,
-    DEFAULT_CLOUD_IO_PARALLELISM,
-};
-use hdfs_native_object_store::HdfsObjectStoreBuilder;
 use lance_core::Error;
 use object_store::path::Path;
+use object_store_opendal::OpendalStore;
+use opendal::{Operator, services::HdfsNative};
 use snafu::location;
 
+use crate::object_store::{
+    DEFAULT_CLOUD_IO_PARALLELISM, ObjectStore, ObjectStoreParams, ObjectStoreProvider,
+    StorageOptions,
+};
+
 #[derive(Debug, Clone)]
 pub struct HdfsStoreProvider;
 
@@ -22,17 +24,27 @@ impl ObjectStoreProvider for HdfsStoreProvider {
         base_path: url::Url,
         params: &ObjectStoreParams,
     ) -> Result<ObjectStore, lance_core::Error> {
-        let storage_options = StorageOptions(params.storage_options.clone().unwrap_or_default());
+        let mut storage_options =
+            StorageOptions(params.storage_options.clone().unwrap_or_default());
 
         let download_retry_count = storage_options.download_retry_count();
-        let hdfs = Arc::new(
-            HdfsObjectStoreBuilder::new()
-                .with_url(base_path.clone())
-                .with_config(storage_options.0.iter())
-                .build()?,
+        storage_options.0.insert(
+            "name_node".to_string(),
+            format!("{}://{}", base_path.scheme(), base_path.authority()),
         );
+
+        let operator = Operator::from_iter::<HdfsNative>(storage_options.0.into_iter())
+            .map_err(|e| {
+                Error::invalid_input(
+                    format!("Failed to create HDFS native operator: {:?}", e),
+                    location!(),
+                )
+            })?
+            .finish();
+
+        let opendal_store = Arc::new(OpendalStore::new(operator));
         Ok(ObjectStore::new(
-            hdfs,
+            opendal_store,
             base_path,
             params.block_size,
             None,
@@ -59,7 +71,6 @@ impl ObjectStoreProvider for HdfsStoreProvider {
         })
     }
 }
-
 #[cfg(test)]
 mod tests {
     use url::Url;