You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by er...@apache.org on 2026/09/24 06:20:01 UTC
(opendal) branch main updated: feat(services/hdfs): add copy support (#8332)
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 53610f898e feat(services/hdfs): add copy support (#8332)
53610f898e is described below
commit 53610f898eb07e38803614ea6aa7cb20bcd0ddd3
Author: hfutatzhanghb <hf...@163.com>
AuthorDate: Thu Sep 24 14:19:53 2026 +0800
feat(services/hdfs): add copy support (#8332)
* feat(services/hdfs): add copy support
Enable HDFS copy via OneShotCopier by streaming bytes through hdrs
AsyncFile APIs, creating parent directories and overwriting existing
destination files when needed.
* feat(services/hdfs): copy via hdrs copy_file
Bump hdrs to 0.3.3 and replace the AsyncFile streaming copy with native Client::copy_file.
* fix(services/hdfs): run copy_file off the async worker
Move hdrs copy_file onto spawn_blocking, drop the pre-delete now that
hdfsCopy overwrites, and use OneShotCopier::new_with so temporary IO
errors can be retried.
* fix(services/hdfs): keep destination delete before copy
Restore remove_file for existing copy destinations and document that
hdfsCopy has been verified to overwrite natively.
* docs(services/hdfs): clarify copy semantics
* ci: retrigger checks
---------
Co-authored-by: zhanghaobo@kanzhun.com <zh...@kanzhun.com>
---
core/Cargo.lock | 8 ++---
core/services/hdfs/Cargo.toml | 2 +-
core/services/hdfs/src/backend.rs | 22 ++++++++----
core/services/hdfs/src/core.rs | 71 ++++++++++++++++++++++++++++++---------
core/services/hdfs/src/docs.md | 10 +++++-
5 files changed, 84 insertions(+), 29 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index c53d7b5c40..fd2a2f16fe 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -4091,9 +4091,9 @@ dependencies = [
[[package]]
name = "hdfs-sys"
-version = "0.3.0"
+version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33e2d5cefba2d51a26b44d2a493f963a32725a0f6593c91be4a610ad449c49cb"
+checksum = "18138c3dacdf0e95f834369f996b21d2b6fa99e71787ef45e2ece09b37bd1dd1"
dependencies = [
"cc",
"java-locator",
@@ -4101,9 +4101,9 @@ dependencies = [
[[package]]
name = "hdrs"
-version = "0.3.2"
+version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7c42a693bfe5dc8fcad1f24044c5ec355c5f157b8ce63c7d62f51cecbc7878d"
+checksum = "c29b0c501a568c941c1ed9044284784978ac28dc173b173f1191ff771de9a27a"
dependencies = [
"blocking",
"errno",
diff --git a/core/services/hdfs/Cargo.toml b/core/services/hdfs/Cargo.toml
index 0d26f199cd..bfd22c85a2 100644
--- a/core/services/hdfs/Cargo.toml
+++ b/core/services/hdfs/Cargo.toml
@@ -36,7 +36,7 @@ opendal-core = { path = "../../core", version = "0.59.3", default-features = fal
bytes = { workspace = true }
futures = { workspace = true }
-hdrs = { version = "0.3.2", features = ["async_file"] }
+hdrs = { version = "0.3.3", features = ["async_file"] }
log = { workspace = true }
serde = { workspace = true, features = ["derive"] }
tokio = { workspace = true, features = ["rt"] }
diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs
index c3cc973fe4..a97eef36a2 100644
--- a/core/services/hdfs/src/backend.rs
+++ b/core/services/hdfs/src/backend.rs
@@ -169,6 +169,8 @@ impl Builder for HdfsBuilder {
list: true,
+ copy: true,
+
rename: true,
rename_with_if_not_exists: true,
@@ -195,7 +197,7 @@ impl Service for HdfsBackend {
type Writer = HdfsLazyWriter;
type Lister = Option<HdfsLister>;
type Deleter = oio::OneShotDeleter<HdfsDeleter>;
- type Copier = ();
+ type Copier = oio::OneShotCopier;
type Composer = ();
fn info(&self) -> ServiceInfo {
@@ -258,14 +260,20 @@ impl Service for HdfsBackend {
fn copy(
&self,
_ctx: &OperationContext,
- _from: &str,
- _to: &str,
+ from: &str,
+ to: &str,
_args: OpCopy,
) -> Result<Self::Copier> {
- Err(Error::new(
- ErrorKind::Unsupported,
- "operation is not supported",
- ))
+ let core = self.core.clone();
+ let from = from.to_string();
+ let to = to.to_string();
+ // Recreate the future so retry layers can rerun the copy after a temporary error.
+ Ok(oio::OneShotCopier::new_with(move || {
+ let core = core.clone();
+ let from = from.clone();
+ let to = to.clone();
+ async move { core.hdfs_copy(&from, &to).await }
+ }))
}
async fn rename(
diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs
index 84a99c5a54..8b2907f1d6 100644
--- a/core/services/hdfs/src/core.rs
+++ b/core/services/hdfs/src/core.rs
@@ -124,8 +124,7 @@ impl HdfsCore {
});
if !target_exists {
- let parent = get_parent(&target_path);
- self.client.create_dir(parent).map_err(new_std_io_error)?;
+ self.ensure_parent_dir(&target_path)?;
}
if !should_append {
initial_size = 0;
@@ -175,20 +174,7 @@ impl HdfsCore {
return Err(new_std_io_error(err));
}
- let parent = std::path::PathBuf::from(&to_path)
- .parent()
- .ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "path should have parent but not, it must be malformed",
- )
- .with_context("to", &to_path)
- })?
- .to_path_buf();
-
- self.client
- .create_dir(&parent.to_string_lossy())
- .map_err(new_std_io_error)?;
+ self.ensure_parent_dir(&to_path)?;
}
Ok(metadata) => {
if metadata.is_file() {
@@ -215,6 +201,59 @@ impl HdfsCore {
Ok(())
}
+
+ pub async fn hdfs_copy(&self, from: &str, to: &str) -> Result<Metadata> {
+ let from_path = build_rooted_abs_path(&self.root, from);
+ // OpenDAL copy is file-to-file only. Reject directory sources before
+ // the HDFS API can recursively copy their contents.
+ let from_meta = self.client.metadata(&from_path).map_err(new_std_io_error)?;
+ if !from_meta.is_file() {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "from path should be a file")
+ .with_context("from", &from_path),
+ );
+ }
+
+ let to_path = build_rooted_abs_path(&self.root, to);
+ match self.client.metadata(&to_path) {
+ Ok(meta) => {
+ // OpenDAL treats `to` as the exact destination file path. Reject
+ // an existing directory instead of copying the source into it.
+ if meta.is_dir() {
+ return Err(
+ Error::new(ErrorKind::IsADirectory, "to path should be a file")
+ .with_context("to", &to_path),
+ );
+ }
+ // The HDFS copy API does not replace an existing destination,
+ // so remove it first to preserve OpenDAL's overwrite semantics.
+ self.client
+ .remove_file(&to_path)
+ .map_err(new_std_io_error)?;
+ }
+ Err(err) if err.kind() == io::ErrorKind::NotFound => {
+ // hdrs copy_file requires the destination parent to already exist.
+ self.ensure_parent_dir(&to_path)?;
+ }
+ Err(err) => return Err(new_std_io_error(err)),
+ }
+
+ let client = self.client.clone();
+ let copy_from = from_path.clone();
+ let copy_to = to_path.clone();
+ tokio::task::spawn_blocking(move || client.copy_file(©_from, ©_to))
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, "tokio task join failed").set_source(e))?
+ .map_err(new_std_io_error)?;
+
+ Ok(MetadataBuilder::file(from_meta.len()).build())
+ }
+
+ fn ensure_parent_dir(&self, path: &str) -> Result<()> {
+ self.client
+ .create_dir(get_parent(path))
+ .map_err(new_std_io_error)
+ }
}
#[cfg(test)]
diff --git a/core/services/hdfs/src/docs.md b/core/services/hdfs/src/docs.md
index 97765d3f5e..9a4114ddaa 100644
--- a/core/services/hdfs/src/docs.md
+++ b/core/services/hdfs/src/docs.md
@@ -10,13 +10,21 @@ Depending on its configuration and the backing system, this service can expose:
- [x] write
- [x] delete
- [x] list
-- [ ] copy
+- [x] copy
- [x] rename
- [ ] ~~presign~~
Inspect the effective capability set with [`opendal_core::Operator::info`] and
[`opendal_core::OperatorInfo::capability`] after building an operator.
+## Copy behavior
+
+The HDFS service copies files to exact destination file paths and rejects
+directory sources and destinations. Copy overwrites an existing destination by
+removing it before copying the source. This replacement is not atomic: if the
+copy fails after removing the destination, the destination can be missing or
+incomplete.
+
## Differences with webhdfs
The [WebHDFS service](https://docs.rs/opendal-service-webhdfs) uses HDFS's