feat(service/hdfs): enable rename in hdfs service (#3592)

* enable rename in hdfs * fix codestyle * add src check * fix code style * fix feedback --------- Co-authored-by: Qingwen Zhao <[email protected]>
apache · Nov 16, 2023 · 8ddf6e8 · 8ddf6e8
1 parent a7c046f
commit 8ddf6e8
Show file tree

Hide file tree

Showing 2 changed files with 97 additions and 2 deletions.
diff --git a/core/src/services/hdfs/backend.rs b/core/src/services/hdfs/backend.rs
@@ -184,6 +184,7 @@ impl Accessor for HdfsBackend {
                 list: true,
                 list_without_recursive: true,
 
+                rename: true,
                 blocking: true,
 
                 ..Default::default()
@@ -250,6 +251,53 @@ impl Accessor for HdfsBackend {
         Ok((RpWrite::new(), HdfsWriter::new(f)))
     }
 
+    async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+        let from_path = build_rooted_abs_path(&self.root, from);
+        self.client.metadata(&from_path).map_err(new_std_io_error)?;
+
+        let to_path = build_rooted_abs_path(&self.root, to);
+        let result = self.client.metadata(&to_path);
+        match result {
+            Err(err) => {
+                // Early return if other error happened.
+                if err.kind() != io::ErrorKind::NotFound {
+                    return Err(new_std_io_error(err));
+                }
+
+                let parent = PathBuf::from(&to_path)
+                    .parent()
+                    .ok_or_else(|| {
+                        Error::new(
+                            ErrorKind::Unexpected,
+                            "path should have parent but not, it must be malformed",
+                        )
+                        .with_context("input", &to_path)
+                    })?
+                    .to_path_buf();
+
+                self.client
+                    .create_dir(&parent.to_string_lossy())
+                    .map_err(new_std_io_error)?;
+            }
+            Ok(metadata) => {
+                if metadata.is_file() {
+                    self.client
+                        .remove_file(&to_path)
+                        .map_err(new_std_io_error)?;
+                } else {
+                    return Err(Error::new(ErrorKind::IsADirectory, "path should be a file")
+                        .with_context("input", &to_path));
+                }
+            }
+        }
+
+        self.client
+            .rename_file(&from_path, &to_path)
+            .map_err(new_std_io_error)?;
+
+        Ok(RpRename::new())
+    }
+
     async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
         let p = build_rooted_abs_path(&self.root, path);
 
@@ -367,6 +415,53 @@ impl Accessor for HdfsBackend {
         Ok((RpWrite::new(), HdfsWriter::new(f)))
     }
 
+    fn blocking_rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
+        let from_path = build_rooted_abs_path(&self.root, from);
+        self.client.metadata(&from_path).map_err(new_std_io_error)?;
+
+        let to_path = build_rooted_abs_path(&self.root, to);
+        let result = self.client.metadata(&to_path);
+        match result {
+            Err(err) => {
+                // Early return if other error happened.
+                if err.kind() != io::ErrorKind::NotFound {
+                    return Err(new_std_io_error(err));
+                }
+
+                let parent = PathBuf::from(&to_path)
+                    .parent()
+                    .ok_or_else(|| {
+                        Error::new(
+                            ErrorKind::Unexpected,
+                            "path should have parent but not, it must be malformed",
+                        )
+                        .with_context("input", &to_path)
+                    })?
+                    .to_path_buf();
+
+                self.client
+                    .create_dir(&parent.to_string_lossy())
+                    .map_err(new_std_io_error)?;
+            }
+            Ok(metadata) => {
+                if metadata.is_file() {
+                    self.client
+                        .remove_file(&to_path)
+                        .map_err(new_std_io_error)?;
+                } else {
+                    return Err(Error::new(ErrorKind::IsADirectory, "path should be a file")
+                        .with_context("input", &to_path));
+                }
+            }
+        }
+
+        self.client
+            .rename_file(&from_path, &to_path)
+            .map_err(new_std_io_error)?;
+
+        Ok(RpRename::new())
+    }
+
     fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
         let p = build_rooted_abs_path(&self.root, path);
 

diff --git a/core/src/services/hdfs/docs.md b/core/src/services/hdfs/docs.md
@@ -10,12 +10,12 @@ This service can be used to:
 - [x] create_dir
 - [x] delete
 - [ ] copy
-- [ ] rename
+- [x] rename
 - [x] list
 - [ ] ~~scan~~
 - [ ] ~~presign~~
 - [x] blocking
-- [x] append
+- [ ] append
 
 ## Differences with webhdfs