feat(services): add hdfs native layout

core/Cargo.toml

@@ -195,6 +195,7 @@ services-vercel-artifacts = []
 services-wasabi = []
 services-webdav = []
 services-webhdfs = []
+services-hdfs-native = ["dep:hdfs-native"]
 services-yandex-disk = []
 
 [lib]
@@ -306,6 +307,7 @@ tokio = { version = "1.27", features = ["sync"] }
 tokio-postgres = { version = "0.7.8", optional = true }
 tracing = { version = "0.1", optional = true }
 uuid = { version = "1", features = ["serde", "v4"] }
+hdfs-native = { version = "0.6.0",  optional = true }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 getrandom = { version = "0.2", features = ["js"] }

core/src/services/hdfs_native/backend.rs

@@ -0,0 +1,228 @@
+// 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.
+
+use std::collections::HashMap;
+use std::fmt::{Debug, Formatter};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::debug;
+use serde::Deserialize;
+use uuid::Uuid;
+
+use super::lister::NativeHdfsLister;
+use super::reader::NativeHdfsReader;
+use super::writer::NativeHdfsWriter;
+use crate::raw::*;
+use crate::*;
+
+/// [Hadoop Distributed File System (HDFS™)](https://hadoop.apache.org/) support.
+/// Using [Native Rust HDFS client](https://github.com/Kimahriman/hdfs-native).
+
+/// Config for NativeHdfs services support.
+#[derive(Default, Deserialize, Clone)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct HdfsNativeConfig {
+    /// work dir of this backend
+    pub root: Option<String>,
+    /// url of this backend
+    pub url: Option<String>,
+    /// enable the append capacity
+    pub enable_append: bool,
+}
+
+impl Debug for HdfsNativeConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("HdfsNativeConfig")
+            .field("root", &self.root)
+            .field("url", &self.url)
+            .field("enable_append", &self.enable_append)
+            .finish_non_exhaustive()
+    }
+}
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct HdfsNativeBuilder {
+    config: HdfsNativeConfig,
+}
+
+impl Debug for HdfsNativeBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("HdfsNativeBuilder")
+            .field("config", &self.config)
+            .finish()
+    }
+}
+
+impl HdfsNativeBuilder {
+    /// Set root of this backend.
+    ///
+    /// All operations will happen under this root.
+    pub fn root(&mut self, root: &str) -> &mut Self {
+        self.config.root = if root.is_empty() {
+            None
+        } else {
+            Some(root.to_string())
+        };
+
+        self
+    }
+
+    /// Set url of this backend.
+    ///
+    /// Valid format including:
+    ///
+    /// - `default`: using the default setting based on hadoop config.
+    /// - `hdfs://127.0.0.1:9000`: connect to hdfs cluster.
+    pub fn url(&mut self, url: &str) -> &mut Self {
+        if !name_node.is_empty() {
+            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
+            self.config.url = Some(url.trim_end_matches('/').to_string())
+        }
+
+        self
+    }
+
+    /// Enable append capacity of this backend.
+    ///
+    /// This should be disabled when HDFS runs in non-distributed mode.
+    pub fn enable_append(&mut self, enable_append: bool) -> &mut Self {
+        self.config.enable_append = enable_append;
+        self
+    }
+}
+
+impl Builder for HdfsNativeBuilder {
+    const SCHEME: Scheme = Scheme::NativeHdfs;
+    type Accessor = NativeHdfsBackend;
+
+    fn from_map(map: HashMap<String, String>) -> Self {
+        // Deserialize the configuration from the HashMap.
+        let config = HdfsNativeConfig::deserialize(ConfigDeserializer::new(map))
+            .expect("config deserialize must succeed");
+
+        // Create an NativeHdfsBuilder instance with the deserialized config.
+        HdfsNativeBuilder { config }
+    }
+
+    fn build(&mut self) -> Result<Self::Accessor> {
+        debug!("backend build started: {:?}", &self);
+
+        let url = match &self.config.url {
+            Some(v) => v,
+            None => {
+                return Err(Error::new(ErrorKind::ConfigInvalid, "url is empty")
+                    .with_context("service", Scheme::NativeHdfs))
+            }
+        };
+
+        let root = normalize_root(&self.config.root.take().unwrap_or_default());
+        debug!("backend use root {}", root);
+
+        let client = hdfs_native::Client::new(url).map_err(new_std_io_error)?;
+
+        // need to check if root dir exists, create if not
+
+        debug!("backend build finished: {:?}", &self);
+        Ok(NativeHdfsBackend {
+            root,
+            client: Arc::new(client),
+            enable_append: self.config.enable_append,
+        })
+    }
+}
+
+#[inline]
+fn tmp_file_of(path: &str) -> String {
+    let name = get_basename(path);
+    let uuid = Uuid::new_v4().to_string();
+
+    format!("{name}.{uuid}")
+}
+
+/// Backend for hdfs-native services.
+#[derive(Debug, Clone)]
+pub struct NativeHdfsBackend {
+    root: String,
+    client: Arc<hdfs_native::Client>,
+    enable_append: bool,
+}
+
+/// hdfs_native::Client is thread-safe.
+unsafe impl Send for NativeHdfsBackend {}
+unsafe impl Sync for NativeHdfsBackend {}
+
+#[async_trait]
+impl Accessor for NativeHdfsBackend {
+    type Reader = NativeHdfsReader;
+    type BlockingReader = ();
+    type Writer = NativeHdfsWriter;
+    type BlockingWriter = ();
+    type Lister = Option<NativeHdfsLister>;
+    type BlockingLister = ();
+
+    fn info(&self) -> AccessorInfo {
+        todo!()
+    }
+
+    async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
+        let p = build_rooted_abs_path(&self.root, path);
+
+        self.client
+            .mkdirs(&p, 0o777, true)
+            .await
+            .map_err(new_std_io_error)?;
+        Ok(RpCreateDir::default())
+    }
+
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let p = build_rooted_abs_path(&self.root, path);
+
+        let f = self.client.read(&p).await.map_err(new_std_io_error)?;
+
+        let r = NativeHdfsReader::new(f);
+
+        Ok((RpRead::new(), r))
+    }
+
+    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        todo!()
+    }
+
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        todo!()
+    }
+
+    async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+        todo!()
+    }
+
+    async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+        todo!()
+    }
+
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+        todo!()
+    }
+
+    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+        let p = build_rooted_abs_path(&self.root, path);
+        todo!()
+    }
+}

core/src/services/hdfs_native/docs.md

@@ -0,0 +1,82 @@
+A distributed file system that provides high-throughput access to application data.
+
+## Capabilities
+
+This service can be used to:
+
+- [x] stat
+- [x] read
+- [x] write
+- [x] create_dir
+- [x] delete
+- [ ] copy
+- [x] rename
+- [x] list
+- [ ] ~~scan~~
+- [ ] ~~presign~~
+- [x] blocking
+- [x] append
+
+## Differences with webhdfs
+
+[Webhdfs][crate::services::Webhdfs] is powered by hdfs's RESTful HTTP API.
+
+## Features
+
+Native HDFS support needs to enable feature `services-native-hdfs`.
+
+## Configuration
+
+- `root`: Set the work dir for backend.
+- `name_node`: Set the name node for backend.
+- `enable_append`: enable the append capacity. Default is false.
+
+Refer to [`HdfsNativeBuilder`]'s public API docs for more information.
+
+## Environment
+
+If HDFS has High Availability (HA) enabled with multiple available NameNodes, some configuration is required:
+1. Obtain the entire HDFS config folder (usually located at HADOOP_HOME/etc/hadoop).
+2. Set the environment variable HADOOP_CONF_DIR to the path of this folder.
+```shell
+export HADOOP_CONF_DIR=<path of the config folder>
+```
+3. Use the `cluster_name` specified in the `core-site.xml` file (located in the HADOOP_CONF_DIR folder) to replace namenode:port.
+
+```rust
+builder.name_node("hdfs://cluster_name");
+```
+
+## Example
+
+### Via Builder
+
+```rust
+use std::sync::Arc;
+
+use anyhow::Result;
+use opendal::services::Hdfs;
+use opendal::Operator;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Create native hdfs backend builder.
+    let mut builder = HdfsNative::default();
+    // Set the url for hdfs.
+    builder.name_node("hdfs://127.0.0.1:9000");
+    // Set the root for hdfs, all operations will happen under this root.
+    //
+    // NOTE: the root must be absolute path.
+    builder.root("/tmp");
+
+    // Enable the append capacity for hdfs. 
+    // 
+    // Note: HDFS run in non-distributed mode doesn't support append.
+    builder.enable_append(true);
+
+    // `Accessor` provides the low level APIs, we will use `Operator` normally.
+    let op: Operator = Operator::new(builder)?.finish();
+
+    Ok(())
+}
+```

core/src/services/hdfs_native/lister.rs

@@ -0,0 +1,39 @@
+// 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.
+
+use crate::raw::oio;
+use crate::raw::oio::Entry;
+use crate::*;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+pub struct HdfsNativeLister {
+    root: String,
+    client: Arc<hdfs_native::Client>,
+}
+
+impl HdfsNativeLister {
+    pub fn new(path: String, client: Arc<hdfs_native::Client>) -> Self {
+        HdfsNativeLister { root: path, client }
+    }
+}
+
+impl oio::List for HdfsNativeLister {
+    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<Entry>>> {
+        todo!()
+    }
+}

core/src/services/hdfs_native/mod.rs

@@ -0,0 +1,24 @@
+// 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.
+
+mod backend;
+pub use backend::HdfsNativeBuilder as HdfsNative;
+pub use backend::HdfsNativeConfig;
+
+mod lister;
+mod reader;
+mod writer;