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

.github/scripts/test_behavior/plan.py

@@ -221,9 +221,10 @@ def generate_language_binding_cases(
 ) -> list[dict[str, str]]:
     cases = unique_cases(cases)
 
-    # Remove hdfs cases for java.
+    # Remove specified services cases for java.
+    excluded_services = {"hdfs", "hdfs-native"}  # Use a set for faster lookups
     if language == "java":
-        cases = [v for v in cases if v["service"] != "hdfs"]
+        cases = [v for v in cases if v["service"] not in excluded_services]
 
     if os.getenv("GITHUB_IS_PUSH") == "true":
         return cases

.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=http://127.0.0.1:9000
+        OPENDAL_HDFS_NATIVE_ENABLE_APPEND=false
+        EOF

bindings/java/Cargo.toml

@@ -63,6 +63,7 @@ services-all = [
   "services-gdrive",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-memcached",
@@ -125,6 +126,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

@@ -63,6 +63,7 @@ services-all = [
   "services-gridfs",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-koofr",
@@ -120,6 +121,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

@@ -60,6 +60,7 @@ services-all = [
   "services-gdrive",
   # FIXME how to support HDFS services in bindings?
   # "services-hdfs",
+  # "services-hdfs-native",
   "services-huggingface",
   "services-ipfs",
   "services-memcached",
@@ -122,6 +123,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

@@ -281,7 +281,8 @@ impl Accessor for HdfsNativeBackend {
 
     async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
         let p = build_rooted_abs_path(&self.root, path);
-        let l = HdfsNativeLister::new(p, self.client.clone());
+        let list_status_iterator = self.client.list_status_iter(&p, false);
+        let l = HdfsNativeLister::new(&self.root, list_status_iterator);
         Ok((RpList::default(), Some(l)))
     }
 }

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,20 +18,28 @@
 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> {
-        todo!()
+    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))
     }
 }

core/src/services/hdfs_native/writer.rs

@@ -18,25 +18,29 @@
 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<usize> {
-        todo!()
+    async fn write(&mut self, bs: Buffer) -> Result<usize> {
+        let bytes = bs.to_bytes();
+        let total_bytes = bytes.len();
+        self.f.write(bytes).await.map_err(parse_hdfs_error)?;
+        Ok(total_bytes)
     }
 
     async fn close(&mut self) -> Result<()> {
-        todo!()
+        self.f.close().await.map_err(parse_hdfs_error)
     }
 
     async fn abort(&mut self) -> Result<()> {

core/src/types/scheme.rs

@@ -384,7 +384,7 @@ impl FromStr for Scheme {
             "tikv" => Ok(Scheme::Tikv),
             "azfile" => Ok(Scheme::Azfile),
             "mongodb" => Ok(Scheme::Mongodb),
-            "hdfs_native" => Ok(Scheme::HdfsNative),
+            "hdfs-native" => Ok(Scheme::HdfsNative),
             "surrealdb" => Ok(Scheme::Surrealdb),
             _ => Ok(Scheme::Custom(Box::leak(s.into_boxed_str()))),
         }
@@ -454,7 +454,7 @@ impl From<Scheme> for &'static str {
             Scheme::Upyun => "upyun",
             Scheme::YandexDisk => "yandex_disk",
             Scheme::Pcloud => "pcloud",
-            Scheme::HdfsNative => "hdfs_native",
+            Scheme::HdfsNative => "hdfs-native",
             Scheme::Surrealdb => "surrealdb",
             Scheme::Custom(v) => v,
         }