opendal_service_hdfs_native/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::sync::Arc;
20
21use log::debug;
22
23use super::HDFS_NATIVE_SCHEME;
24use super::config::HDFS_DEFAULT_AUTHORITY;
25use super::config::HDFS_SCHEME_PREFIX;
26use super::config::HdfsNativeConfig;
27use super::config::init_hdfs_config;
28use super::core::HdfsNativeCore;
29use super::core::parse_hdfs_error;
30use super::deleter::HdfsNativeDeleter;
31use super::reader::*;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35/// [Hadoop Distributed File System (HDFS™)](https://hadoop.apache.org/) support.
36/// Using [Native Rust HDFS client](https://github.com/Kimahriman/hdfs-native).
37#[doc = include_str!("docs.md")]
38#[derive(Debug, Default)]
39pub struct HdfsNativeBuilder {
40    pub(super) config: HdfsNativeConfig,
41}
42
43impl HdfsNativeBuilder {
44    /// Set root of this backend.
45    ///
46    /// All operations will happen under this root.
47    pub fn root(mut self, root: &str) -> Self {
48        self.config.root = if root.is_empty() {
49            None
50        } else {
51            Some(root.to_string())
52        };
53
54        self
55    }
56
57    /// Set name_node of this backend.
58    ///
59    /// Valid format including:
60    ///
61    /// - `default`: using the default setting based on hadoop config.
62    /// - `hdfs://127.0.0.1:9000`: connect to hdfs cluster.
63    /// - `hdfs://namenode1:9000,namenode2:9000`: connect to hdfs ha cluster.
64    pub fn name_node(mut self, name_node: &str) -> Self {
65        if !name_node.is_empty() {
66            // Trim trailing `/` so that we can accept `hdfs://127.0.0.1:9000/`
67            self.config.name_node = Some(name_node.trim_end_matches('/').to_string())
68        }
69
70        self
71    }
72
73    /// Deprecated: HDFS Native append capability is enabled by default.
74    #[deprecated(
75        since = "0.57.0",
76        note = "HDFS Native append capability is enabled by default and this option is no longer needed."
77    )]
78    pub fn enable_append(self, _enable_append: bool) -> Self {
79        self
80    }
81
82    /// Set other hdfs-native client options of this backend.
83    ///
84    /// See the [hdfs-native documentation](https://github.com/Kimahriman/hdfs-native)
85    /// for the supported options.
86    pub fn options(mut self, options: HashMap<String, String>) -> Self {
87        self.config.options = Some(options);
88        self
89    }
90}
91
92impl Builder for HdfsNativeBuilder {
93    type Config = HdfsNativeConfig;
94
95    fn build(self) -> Result<impl Service> {
96        debug!("backend build started: {:?}", self);
97
98        let name_node = match &self.config.name_node {
99            Some(v) => v,
100            None => {
101                return Err(Error::new(ErrorKind::ConfigInvalid, "name_node is empty")
102                    .with_context("service", HDFS_NATIVE_SCHEME));
103            }
104        };
105
106        let root = normalize_root(&self.config.root.unwrap_or_default());
107        debug!("backend use root {root}");
108
109        let mut hdfs_config = init_hdfs_config(name_node);
110        if let Some(options) = &self.config.options {
111            hdfs_config.extend(options.iter().map(|(k, v)| (k.clone(), v.clone())));
112        }
113
114        let client = hdfs_native::ClientBuilder::new()
115            .with_url(format!("{}{}", HDFS_SCHEME_PREFIX, HDFS_DEFAULT_AUTHORITY))
116            .with_config(hdfs_config)
117            .build()
118            .map_err(parse_hdfs_error)?;
119
120        // need to check if root dir exists, create if not
121        Ok(HdfsNativeBackend {
122            core: Arc::new(HdfsNativeCore {
123                info: ServiceInfo::new(HDFS_NATIVE_SCHEME, &root, ""),
124                capability: Capability {
125                    stat: true,
126
127                    read: true,
128
129                    write: true,
130                    write_can_append: true,
131
132                    create_dir: true,
133                    delete: true,
134
135                    list: true,
136
137                    rename: true,
138
139                    shared: true,
140
141                    ..Default::default()
142                },
143                root,
144                client: Arc::new(client),
145            }),
146        })
147    }
148}
149
150// #[inline]
151// fn tmp_file_of(path: &str) -> String {
152//     let name = get_basename(path);
153//     let uuid = Uuid::new_v4().to_string();
154
155//     format!("{name}.{uuid}")
156// }
157
158/// Backend for hdfs-native services.
159#[derive(Debug, Clone)]
160pub struct HdfsNativeBackend {
161    pub(crate) core: Arc<HdfsNativeCore>,
162}
163
164impl Service for HdfsNativeBackend {
165    type Reader = oio::PositionReader<HdfsNativeReader>;
166    type Writer = HdfsNativeLazyWriter;
167    type Lister = HdfsNativeLazyLister;
168    type Deleter = oio::OneShotDeleter<HdfsNativeDeleter>;
169    type Copier = ();
170
171    fn info(&self) -> ServiceInfo {
172        self.core.info.clone()
173    }
174
175    fn capability(&self) -> Capability {
176        self.core.capability
177    }
178
179    async fn create_dir(
180        &self,
181        _ctx: &OperationContext,
182        path: &str,
183        _args: OpCreateDir,
184    ) -> Result<RpCreateDir> {
185        self.core.hdfs_create_dir(path).await?;
186        Ok(RpCreateDir::default())
187    }
188
189    async fn stat(&self, _ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
190        let m = self.core.hdfs_stat(path).await?;
191        Ok(RpStat::new(m))
192    }
193    fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
194        Ok(oio::PositionReader::new(HdfsNativeReader::new(
195            self.core.clone(),
196            path,
197        )))
198    }
199
200    fn write(&self, _ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
201        Ok(HdfsNativeLazyWriter::new(self.core.clone(), path, args))
202    }
203
204    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
205        let output: oio::OneShotDeleter<HdfsNativeDeleter> = {
206            Ok(oio::OneShotDeleter::new(HdfsNativeDeleter::new(
207                Arc::clone(&self.core),
208            )))
209        }?;
210
211        Ok(output)
212    }
213
214    fn list(&self, _ctx: &OperationContext, path: &str, _args: OpList) -> Result<Self::Lister> {
215        Ok(HdfsNativeLazyLister::new(self.core.clone(), path))
216    }
217
218    fn copy(
219        &self,
220        _ctx: &OperationContext,
221        _from: &str,
222        _to: &str,
223        _args: OpCopy,
224        _opts: OpCopier,
225    ) -> Result<Self::Copier> {
226        Err(Error::new(
227            ErrorKind::Unsupported,
228            "operation is not supported",
229        ))
230    }
231
232    async fn rename(
233        &self,
234        _ctx: &OperationContext,
235        from: &str,
236        to: &str,
237        _args: OpRename,
238    ) -> Result<RpRename> {
239        self.core.hdfs_rename(from, to).await?;
240        Ok(RpRename::default())
241    }
242
243    async fn presign(
244        &self,
245        _ctx: &OperationContext,
246        _path: &str,
247        _args: OpPresign,
248    ) -> Result<RpPresign> {
249        Err(Error::new(
250            ErrorKind::Unsupported,
251            "operation is not supported",
252        ))
253    }
254}
opendal_service_hdfs_native/backend.rs

opendal_service_hdfs_native/
backend.rs
