opendal_service_dbfs/
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::sync::Arc;
19
20use bytes::Buf;
21use http::StatusCode;
22use log::debug;
23use serde::Deserialize;
24
25use super::DBFS_SCHEME;
26use super::config::DbfsConfig;
27use super::core::DbfsCore;
28use super::core::parse_error;
29use super::deleter::DbfsDeleter;
30use super::lister::DbfsLister;
31use super::writer::DbfsWriter;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35/// [Dbfs](https://docs.databricks.com/api/azure/workspace/dbfs)'s REST API support.
36#[doc = include_str!("docs.md")]
37#[derive(Debug, Default)]
38pub struct DbfsBuilder {
39    pub(super) config: DbfsConfig,
40}
41
42impl DbfsBuilder {
43    /// Set root of this backend.
44    ///
45    /// All operations will happen under this root.
46    pub fn root(mut self, root: &str) -> Self {
47        self.config.root = if root.is_empty() {
48            None
49        } else {
50            Some(root.to_string())
51        };
52
53        self
54    }
55
56    /// Set endpoint of this backend.
57    ///
58    /// Endpoint must be full uri, e.g.
59    ///
60    /// - Azure: `https://adb-1234567890123456.78.azuredatabricks.net`
61    /// - Aws: `https://dbc-123a5678-90bc.cloud.databricks.com`
62    pub fn endpoint(mut self, endpoint: &str) -> Self {
63        self.config.endpoint = if endpoint.is_empty() {
64            None
65        } else {
66            Some(endpoint.trim_end_matches('/').to_string())
67        };
68        self
69    }
70
71    /// Set the token of this backend.
72    pub fn token(mut self, token: &str) -> Self {
73        if !token.is_empty() {
74            self.config.token = Some(token.to_string());
75        }
76        self
77    }
78}
79
80impl Builder for DbfsBuilder {
81    type Config = DbfsConfig;
82
83    /// Build a DbfsBackend.
84    fn build(self) -> Result<impl Service> {
85        debug!("backend build started: {:?}", self);
86
87        let root = normalize_root(&self.config.root.unwrap_or_default());
88        debug!("backend use root {root}");
89
90        let endpoint = match &self.config.endpoint {
91            Some(endpoint) => Ok(endpoint.clone()),
92            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
93                .with_operation("Builder::build")
94                .with_context("service", DBFS_SCHEME)),
95        }?;
96        debug!("backend use endpoint: {}", endpoint);
97
98        let token = match self.config.token {
99            Some(token) => token,
100            None => {
101                return Err(Error::new(
102                    ErrorKind::ConfigInvalid,
103                    "missing token for Dbfs",
104                ));
105            }
106        };
107
108        let capability = Capability {
109            stat: true,
110
111            write: true,
112            create_dir: true,
113            delete: true,
114            rename: true,
115
116            list: true,
117
118            shared: true,
119
120            ..Default::default()
121        };
122
123        Ok(DbfsBackend {
124            core: Arc::new(DbfsCore {
125                root,
126                endpoint: endpoint.to_string(),
127                token,
128            }),
129            capability,
130        })
131    }
132}
133
134/// Backend for DBFS service
135#[derive(Debug, Clone)]
136pub struct DbfsBackend {
137    core: Arc<DbfsCore>,
138    capability: Capability,
139}
140
141impl Service for DbfsBackend {
142    type Reader = ();
143    type Writer = oio::OneShotWriter<DbfsWriter>;
144    type Lister = oio::PageLister<DbfsLister>;
145    type Deleter = oio::OneShotDeleter<DbfsDeleter>;
146    type Copier = ();
147
148    fn info(&self) -> ServiceInfo {
149        ServiceInfo::new(DBFS_SCHEME, &self.core.root, "")
150    }
151
152    fn capability(&self) -> Capability {
153        self.capability
154    }
155
156    async fn create_dir(
157        &self,
158        ctx: &OperationContext,
159        path: &str,
160        _: OpCreateDir,
161    ) -> Result<RpCreateDir> {
162        let resp = self.core.dbfs_create_dir(ctx, path).await?;
163
164        let status = resp.status();
165
166        match status {
167            StatusCode::CREATED | StatusCode::OK => Ok(RpCreateDir::default()),
168            _ => Err(parse_error(resp)),
169        }
170    }
171
172    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
173        // Stat root always returns a DIR.
174        if path == "/" {
175            return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
176        }
177
178        let resp = self.core.dbfs_get_status(ctx, path).await?;
179
180        let status = resp.status();
181
182        match status {
183            StatusCode::OK => {
184                let mut meta = parse_into_metadata(path, resp.headers())?;
185                let bs = resp.into_body();
186                let decoded_response: DbfsStatus =
187                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
188                meta.set_last_modified(Timestamp::from_millisecond(
189                    decoded_response.modification_time,
190                )?);
191                match decoded_response.is_dir {
192                    true => meta.set_mode(EntryMode::DIR),
193                    false => {
194                        meta.set_mode(EntryMode::FILE);
195                        meta.set_content_length(decoded_response.file_size as u64)
196                    }
197                };
198                Ok(RpStat::new(meta))
199            }
200            StatusCode::NOT_FOUND if path.ends_with('/') => {
201                Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
202            }
203            _ => Err(parse_error(resp)),
204        }
205    }
206
207    fn read(&self, _ctx: &OperationContext, _path: &str, _args: OpRead) -> Result<Self::Reader> {
208        Err(Error::new(
209            ErrorKind::Unsupported,
210            "operation is not supported",
211        ))
212    }
213
214    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
215        let output: oio::OneShotWriter<DbfsWriter> = {
216            Ok(oio::OneShotWriter::new(DbfsWriter::new(
217                self.core.clone(),
218                ctx.clone(),
219                args,
220                path.to_string(),
221            )))
222        }?;
223
224        Ok(output)
225    }
226
227    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
228        let output: oio::OneShotDeleter<DbfsDeleter> = {
229            Ok(oio::OneShotDeleter::new(DbfsDeleter::new(
230                self.core.clone(),
231                ctx.clone(),
232            )))
233        }?;
234
235        Ok(output)
236    }
237
238    fn list(&self, ctx: &OperationContext, path: &str, _args: OpList) -> Result<Self::Lister> {
239        let output: oio::PageLister<DbfsLister> = {
240            let l = DbfsLister::new(self.core.clone(), ctx.clone(), path.to_string());
241
242            Ok(oio::PageLister::new(l))
243        }?;
244
245        Ok(output)
246    }
247
248    async fn rename(
249        &self,
250        ctx: &OperationContext,
251        from: &str,
252        to: &str,
253        _args: OpRename,
254    ) -> Result<RpRename> {
255        self.core.dbfs_ensure_parent_path(ctx, to).await?;
256
257        let resp = self.core.dbfs_rename(ctx, from, to).await?;
258
259        let status = resp.status();
260
261        match status {
262            StatusCode::OK => Ok(RpRename::default()),
263            _ => Err(parse_error(resp)),
264        }
265    }
266
267    fn copy(
268        &self,
269        _ctx: &OperationContext,
270        _from: &str,
271        _to: &str,
272        _args: OpCopy,
273        _opts: OpCopier,
274    ) -> Result<Self::Copier> {
275        Err(Error::new(
276            ErrorKind::Unsupported,
277            "operation is not supported",
278        ))
279    }
280
281    async fn presign(
282        &self,
283        _ctx: &OperationContext,
284        _path: &str,
285        _args: OpPresign,
286    ) -> Result<RpPresign> {
287        Err(Error::new(
288            ErrorKind::Unsupported,
289            "operation is not supported",
290        ))
291    }
292}
293
294#[derive(Deserialize)]
295struct DbfsStatus {
296    // Not used fields.
297    // path: String,
298    is_dir: bool,
299    file_size: i64,
300    modification_time: i64,
301}
opendal_service_dbfs/backend.rs

opendal_service_dbfs/
backend.rs
