feat(services/hdfs_native): Add read,write,list implementation for hdfs_native

.github/services/hdfs_native/hdfs/action.yml

@@ -0,0 +1,46 @@
+# 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.
+
+name: hdfs_native
+description: 'Behavior test for hdfs_native'
+
+runs:
+  using: "composite"
+  steps:
+    - name: Setup java env
+      uses: actions/setup-java@v4
+      with:
+        distribution: temurin
+        java-version: "11"
+    - name: Setup
+      shell: bash
+      run: |
+        curl -LsSf https://dlcdn.apache.org/hadoop/common/hadoop-3.3.5/hadoop-3.3.5.tar.gz | tar zxf - -C /home/runner
+
+        export HADOOP_HOME="/home/runner/hadoop-3.3.5"
+        export CLASSPATH=$(${HADOOP_HOME}/bin/hadoop classpath --glob)
+
+        cp ./fixtures/hdfs/hdfs-site.xml ${HADOOP_HOME}/etc/hadoop/hdfs-site.xml
+
+        cat << EOF >> $GITHUB_ENV
+        HADOOP_HOME=${HADOOP_HOME}
+        CLASSPATH=${CLASSPATH}
+        LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${HADOOP_HOME}/lib/native
+        OPENDAL_HDFS_NATIVE_ROOT=/tmp/opendal/
+        OPENDAL_HDFS_NATIVE_URL=hdfs://127.0.0.1:9000
+        OPENDAL_HDFS_NATIVE_ENABLE_APPEND=false
+        EOF

bindings/java/Cargo.toml

@@ -64,6 +64,7 @@ services-all = [
   "services-gdrive",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-memcached",
@@ -126,6 +127,7 @@ services-ftp = ["opendal/services-ftp"]
 services-gdrive = ["opendal/services-gdrive"]
 services-gridfs = ["opendal/services-gridfs"]
 services-hdfs = ["opendal/services-hdfs"]
+services-hdfs-native = ["opendal/services-hdfs-native"]
 services-huggingface = ["opendal/services-huggingface"]
 services-ipfs = ["opendal/services-ipfs"]
 services-koofr = ["opendal/services-koofr"]

bindings/nodejs/Cargo.toml

@@ -62,6 +62,7 @@ services-all = [
   "services-gdrive",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-memcached",
@@ -124,6 +125,7 @@ services-ftp = ["opendal/services-ftp"]
 services-gdrive = ["opendal/services-gdrive"]
 services-gridfs = ["opendal/services-gridfs"]
 services-hdfs = ["opendal/services-hdfs"]
+services-hdfs-native = ["opendal/services-hdfs-native"]
 services-huggingface = ["opendal/services-huggingface"]
 services-ipfs = ["opendal/services-ipfs"]
 services-koofr = ["opendal/services-koofr"]

bindings/python/Cargo.toml

@@ -64,6 +64,7 @@ services-all = [
   "services-gdrive",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-memcached",
@@ -126,6 +127,7 @@ services-ftp = ["opendal/services-ftp"]
 services-gdrive = ["opendal/services-gdrive"]
 services-gridfs = ["opendal/services-gridfs"]
 services-hdfs = ["opendal/services-hdfs"]
+services-hdfs-native = ["opendal/services-hdfs-native"]
 services-huggingface = ["opendal/services-huggingface"]
 services-ipfs = ["opendal/services-ipfs"]
 services-koofr = ["opendal/services-koofr"]

core/src/services/hdfs_native/backend.rs

@@ -152,24 +152,26 @@ impl Access for HdfsNativeBackend {
     type BlockingLister = ();
     type BlockingDeleter = ();
 
-    fn info(&self) -> Arc<AccessorInfo> {
+    fn info(&self) -> AccessorInfo {
         let mut am = AccessorInfo::default();
         am.set_scheme(Scheme::HdfsNative)
             .set_root(&self.root)
             .set_native_capability(Capability {
                 stat: true,
+                list: true,
+                read: true,
+                write: true,
                 stat_has_last_modified: true,
                 stat_has_content_length: true,
 
                 delete: true,
                 rename: true,
-
-                shared: true,
+                blocking: true,
 
                 ..Default::default()
             });
 
-        am.into()
+        am
     }
 
     async fn create_dir(&self, path: &str, _args: OpCreateDir) -> Result<RpCreateDir> {

core/src/services/hdfs_native/docs.md

@@ -6,12 +6,12 @@ Using [Native Rust HDFS client](https://github.com/Kimahriman/hdfs-native).
 This service can be used to:
 
 - [x] stat
-- [ ] read
-- [ ] write
+- [x] read
+- [x] write
 - [ ] create_dir
 - [x] delete
 - [x] rename
-- [ ] list
+- [x] list
 - [x] blocking
 - [ ] append
 

core/src/services/hdfs_native/lister.rs

@@ -15,28 +15,62 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::sync::Arc;
+use chrono::DateTime;
+use hdfs_native::client::ListStatusIterator;
 
-use crate::raw::oio;
 use crate::raw::oio::Entry;
+use crate::raw::{build_rel_path, oio};
+use crate::services::hdfs_native::error::parse_hdfs_error;
 use crate::*;
 
 pub struct HdfsNativeLister {
-    _path: String,
-    _client: Arc<hdfs_native::Client>,
+    root: String,
+    lsi: ListStatusIterator,
 }
 
 impl HdfsNativeLister {
-    pub fn new(path: String, client: Arc<hdfs_native::Client>) -> Self {
-        HdfsNativeLister {
-            _path: path,
-            _client: client,
+    pub fn new(root: &str, lsi: ListStatusIterator) -> Self {
+        Self {
+            root: root.to_string(),
+            lsi,
         }
     }
 }
 
 impl oio::List for HdfsNativeLister {
     async fn next(&mut self) -> Result<Option<Entry>> {
-        todo!()
+        if let Some(de) = self
+            .lsi
+            .next()
+            .await
+            .transpose()
+            .map_err(parse_hdfs_error)?
+        {
+            let path = build_rel_path(&self.root, &de.path);
+
+            let entry = if !de.isdir {
+                let odt = DateTime::from_timestamp(de.modification_time as i64, 0);
+
+                let Some(dt) = odt else {
+                    return Err(Error::new(
+                        ErrorKind::Unexpected,
+                        &format!("Failure in extracting modified_time for {}", path),
+                    ));
+                };
+                let meta = Metadata::new(EntryMode::FILE)
+                    .with_content_length(de.length as u64)
+                    .with_last_modified(dt);
+                oio::Entry::new(&path, meta)
+            } else if de.isdir {
+                // Make sure we are returning the correct path.
+                oio::Entry::new(&format!("{path}/"), Metadata::new(EntryMode::DIR))
+            } else {
+                oio::Entry::new(&path, Metadata::new(EntryMode::Unknown))
+            };
+
+            Ok(Some(entry))
+        } else {
+            Ok(None)
+        }
     }
 }

core/src/services/hdfs_native/reader.rs

@@ -18,19 +18,31 @@
 use hdfs_native::file::FileReader;
 
 use crate::raw::*;
+use crate::services::hdfs_native::error::parse_hdfs_error;
 use crate::*;
 
 pub struct HdfsNativeReader {
-    _f: FileReader,
+    f: FileReader,
 }
 
 impl HdfsNativeReader {
     pub fn new(f: FileReader) -> Self {
-        HdfsNativeReader { _f: f }
+        HdfsNativeReader { f }
     }
 }
 
 impl oio::Read for HdfsNativeReader {
+    async fn read_at(&self, offset: u64, limit: usize) -> Result<Buffer> {
+        // Perform the read operation using read_range
+        let bytes = self
+            .f
+            .read_range(offset as usize, limit)
+            .await
+            .map_err(parse_hdfs_error)?;
+
+        Ok(Buffer::from(bytes))
+    }
+
     async fn read(&mut self) -> Result<Buffer> {
         todo!()
     }

core/src/services/hdfs_native/writer.rs

@@ -18,25 +18,28 @@
 use hdfs_native::file::FileWriter;
 
 use crate::raw::oio;
+use crate::services::hdfs_native::error::parse_hdfs_error;
 use crate::*;
 
 pub struct HdfsNativeWriter {
-    _f: FileWriter,
+    f: FileWriter,
 }
 
 impl HdfsNativeWriter {
     pub fn new(f: FileWriter) -> Self {
-        HdfsNativeWriter { _f: f }
+        HdfsNativeWriter { f }
     }
 }
 
 impl oio::Write for HdfsNativeWriter {
-    async fn write(&mut self, _bs: Buffer) -> Result<()> {
-        todo!()
+    async fn write(&mut self, bs: Buffer) -> Result<usize> {
+        let bytes = bs.to_bytes();
+        let n = self.f.write(bytes).await.map_err(parse_hdfs_error)?;
+        Ok(n)
     }
 
     async fn close(&mut self) -> Result<()> {
-        todo!()
+        self.f.close().await.map_err(parse_hdfs_error)
     }
 
     async fn abort(&mut self) -> Result<()> {