From 8100b0c7ec382158acc4e940eeae06dc8aa74c2a Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Tue, 22 Sep 2026 12:03:25 +0800 Subject: [PATCH 1/6] 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. --- core/services/hdfs/src/backend.rs | 18 ++++---- core/services/hdfs/src/core.rs | 70 +++++++++++++++++++++++++++++++ core/services/hdfs/src/docs.md | 2 +- 3 files changed, 82 insertions(+), 8 deletions(-) diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs index c3cc973fe44e..5065abc49285 100644 --- a/core/services/hdfs/src/backend.rs +++ b/core/services/hdfs/src/backend.rs @@ -174,6 +174,8 @@ impl Builder for HdfsBuilder { shared: true, + copy: true, + ..Default::default() }, root, @@ -195,7 +197,7 @@ impl Service for HdfsBackend { type Writer = HdfsLazyWriter; type Lister = Option; type Deleter = oio::OneShotDeleter; - type Copier = (); + type Copier = oio::OneShotCopier; type Composer = (); fn info(&self) -> ServiceInfo { @@ -258,14 +260,16 @@ impl Service for HdfsBackend { fn copy( &self, _ctx: &OperationContext, - _from: &str, - _to: &str, + from: &str, + to: &str, _args: OpCopy, ) -> Result { - Err(Error::new( - ErrorKind::Unsupported, - "operation is not supported", - )) + let core = self.core.clone(); + let from = from.to_string(); + let to = to.to_string(); + Ok(oio::OneShotCopier::new(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 84a99c5a5404..c53fb29a32f8 100644 --- a/core/services/hdfs/src/core.rs +++ b/core/services/hdfs/src/core.rs @@ -215,6 +215,76 @@ impl HdfsCore { Ok(()) } + + pub async fn hdfs_copy(&self, from: &str, to: &str) -> Result { + use futures::AsyncWriteExt; + + let from_path = build_rooted_abs_path(&self.root, from); + // Verify the source exists and is a file. + 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("input", from), + ); + } + + let to_path = build_rooted_abs_path(&self.root, to); + match self.client.metadata(&to_path) { + Ok(meta) => { + if meta.is_dir() { + return Err(Error::new(ErrorKind::IsADirectory, "path should be a file") + .with_context("to", &to_path)); + } + // Overwrite existing destination files. + self.client + .remove_file(&to_path) + .map_err(new_std_io_error)?; + } + Err(err) if err.kind() == io::ErrorKind::NotFound => { + // Ensure the parent directory of the destination exists. + 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("input", &to_path) + })? + .to_path_buf(); + + self.client + .create_dir(&parent.to_string_lossy()) + .map_err(new_std_io_error)?; + } + Err(err) => return Err(new_std_io_error(err)), + } + + let mut reader = { + let mut open_options = self.client.open_file(); + open_options.read(true); + open_options + .async_open(&from_path) + .await + .map_err(new_std_io_error)? + }; + let mut writer = { + let mut open_options = self.client.open_file(); + open_options.write(true).create(true).truncate(true); + open_options + .async_open(&to_path) + .await + .map_err(new_std_io_error)? + }; + + let size = futures::io::copy(&mut reader, &mut writer) + .await + .map_err(new_std_io_error)?; + writer.close().await.map_err(new_std_io_error)?; + + Ok(MetadataBuilder::file(size).build()) + } } #[cfg(test)] diff --git a/core/services/hdfs/src/docs.md b/core/services/hdfs/src/docs.md index 97765d3f5ed9..6ecae53a47e2 100644 --- a/core/services/hdfs/src/docs.md +++ b/core/services/hdfs/src/docs.md @@ -10,7 +10,7 @@ Depending on its configuration and the backing system, this service can expose: - [x] write - [x] delete - [x] list -- [ ] copy +- [x] copy - [x] rename - [ ] ~~presign~~ From 2751e9ce7ebed53c377fc0e6645d80867296fa5d Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Tue, 22 Sep 2026 20:16:34 +0800 Subject: [PATCH 2/6] 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. --- core/Cargo.lock | 8 ++++---- core/services/hdfs/Cargo.toml | 2 +- core/services/hdfs/src/backend.rs | 2 +- core/services/hdfs/src/core.rs | 33 +++++++------------------------ 4 files changed, 13 insertions(+), 32 deletions(-) diff --git a/core/Cargo.lock b/core/Cargo.lock index c53d7b5c4086..fd2a2f16fef9 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 0d26f199cdd9..bfd22c85a27e 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 5065abc49285..4edf5d6a45b4 100644 --- a/core/services/hdfs/src/backend.rs +++ b/core/services/hdfs/src/backend.rs @@ -268,7 +268,7 @@ impl Service for HdfsBackend { let from = from.to_string(); let to = to.to_string(); Ok(oio::OneShotCopier::new(async move { - core.hdfs_copy(&from, &to).await + core.hdfs_copy(&from, &to) })) } diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs index c53fb29a32f8..3b9b6a9b2f14 100644 --- a/core/services/hdfs/src/core.rs +++ b/core/services/hdfs/src/core.rs @@ -216,9 +216,7 @@ impl HdfsCore { Ok(()) } - pub async fn hdfs_copy(&self, from: &str, to: &str) -> Result { - use futures::AsyncWriteExt; - + pub fn hdfs_copy(&self, from: &str, to: &str) -> Result { let from_path = build_rooted_abs_path(&self.root, from); // Verify the source exists and is a file. let from_meta = self.client.metadata(&from_path).map_err(new_std_io_error)?; @@ -236,13 +234,14 @@ impl HdfsCore { return Err(Error::new(ErrorKind::IsADirectory, "path should be a file") .with_context("to", &to_path)); } - // Overwrite existing destination files. + // Overwrite existing destination files. hdrs copy_file does not + // document overwrite. self.client .remove_file(&to_path) .map_err(new_std_io_error)?; } Err(err) if err.kind() == io::ErrorKind::NotFound => { - // Ensure the parent directory of the destination exists. + // hdrs copy_file requires the destination parent to already exist. let parent = std::path::PathBuf::from(&to_path) .parent() .ok_or_else(|| { @@ -261,29 +260,11 @@ impl HdfsCore { Err(err) => return Err(new_std_io_error(err)), } - let mut reader = { - let mut open_options = self.client.open_file(); - open_options.read(true); - open_options - .async_open(&from_path) - .await - .map_err(new_std_io_error)? - }; - let mut writer = { - let mut open_options = self.client.open_file(); - open_options.write(true).create(true).truncate(true); - open_options - .async_open(&to_path) - .await - .map_err(new_std_io_error)? - }; - - let size = futures::io::copy(&mut reader, &mut writer) - .await + self.client + .copy_file(&from_path, &to_path) .map_err(new_std_io_error)?; - writer.close().await.map_err(new_std_io_error)?; - Ok(MetadataBuilder::file(size).build()) + Ok(MetadataBuilder::file(from_meta.len()).build()) } } From 124a383ee9571e72875b0af3e4217cbde1e95cde Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Tue, 22 Sep 2026 21:02:53 +0800 Subject: [PATCH 3/6] 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. --- core/services/hdfs/src/backend.rs | 11 +++-- core/services/hdfs/src/core.rs | 68 ++++++++++++------------------- 2 files changed, 33 insertions(+), 46 deletions(-) diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs index 4edf5d6a45b4..27207d98afb9 100644 --- a/core/services/hdfs/src/backend.rs +++ b/core/services/hdfs/src/backend.rs @@ -169,13 +169,13 @@ impl Builder for HdfsBuilder { list: true, + copy: true, + rename: true, rename_with_if_not_exists: true, shared: true, - copy: true, - ..Default::default() }, root, @@ -267,8 +267,11 @@ impl Service for HdfsBackend { let core = self.core.clone(); let from = from.to_string(); let to = to.to_string(); - Ok(oio::OneShotCopier::new(async move { - core.hdfs_copy(&from, &to) + 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 } })) } diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs index 3b9b6a9b2f14..e4e86ac8a6f7 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() { @@ -216,56 +202,54 @@ impl HdfsCore { Ok(()) } - pub fn hdfs_copy(&self, from: &str, to: &str) -> Result { + pub async fn hdfs_copy(&self, from: &str, to: &str) -> Result { let from_path = build_rooted_abs_path(&self.root, from); - // Verify the source exists and is a file. + // FileUtil.copy recurses when the source is a directory. 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("input", from), + .with_context("from", &from_path), ); } let to_path = build_rooted_abs_path(&self.root, to); match self.client.metadata(&to_path) { Ok(meta) => { + // FileUtil.checkDest rewrites a directory destination to dst/ + // and copies into it without error. if meta.is_dir() { - return Err(Error::new(ErrorKind::IsADirectory, "path should be a file") - .with_context("to", &to_path)); + return Err( + Error::new(ErrorKind::IsADirectory, "to path should be a file") + .with_context("to", &to_path), + ); } - // Overwrite existing destination files. hdrs copy_file does not - // document overwrite. - self.client - .remove_file(&to_path) - .map_err(new_std_io_error)?; + // hdfsCopy uses FileUtil.copy with overwrite=true, so an existing + // destination file is replaced in the subsequent copy_file call. } Err(err) if err.kind() == io::ErrorKind::NotFound => { // hdrs copy_file requires the destination parent to already exist. - 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("input", &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)?; } Err(err) => return Err(new_std_io_error(err)), } - self.client - .copy_file(&from_path, &to_path) + 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)] From 0327d2ff657415568d92fe49f41d3d529a0da242 Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Tue, 22 Sep 2026 21:09:47 +0800 Subject: [PATCH 4/6] 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. --- core/services/hdfs/src/core.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs index e4e86ac8a6f7..f9bb2ba6b428 100644 --- a/core/services/hdfs/src/core.rs +++ b/core/services/hdfs/src/core.rs @@ -224,8 +224,11 @@ impl HdfsCore { .with_context("to", &to_path), ); } - // hdfsCopy uses FileUtil.copy with overwrite=true, so an existing - // destination file is replaced in the subsequent copy_file call. + // hdfsCopy has been verified to overwrite natively via + // FileUtil.copy(..., overwrite=true). + 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. From 8c6648d76720c01ee9fd675b13dc0ef431736ae5 Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Thu, 24 Sep 2026 00:06:52 +0800 Subject: [PATCH 5/6] docs(services/hdfs): clarify copy semantics --- core/services/hdfs/src/backend.rs | 1 + core/services/hdfs/src/core.rs | 11 ++++++----- core/services/hdfs/src/docs.md | 8 ++++++++ 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs index 27207d98afb9..a97eef36a27d 100644 --- a/core/services/hdfs/src/backend.rs +++ b/core/services/hdfs/src/backend.rs @@ -267,6 +267,7 @@ impl Service for HdfsBackend { 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(); diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs index f9bb2ba6b428..8b2907f1d67f 100644 --- a/core/services/hdfs/src/core.rs +++ b/core/services/hdfs/src/core.rs @@ -204,7 +204,8 @@ impl HdfsCore { pub async fn hdfs_copy(&self, from: &str, to: &str) -> Result { let from_path = build_rooted_abs_path(&self.root, from); - // FileUtil.copy recurses when the source is a directory. + // 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( @@ -216,16 +217,16 @@ impl HdfsCore { let to_path = build_rooted_abs_path(&self.root, to); match self.client.metadata(&to_path) { Ok(meta) => { - // FileUtil.checkDest rewrites a directory destination to dst/ - // and copies into it without error. + // 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), ); } - // hdfsCopy has been verified to overwrite natively via - // FileUtil.copy(..., overwrite=true). + // 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)?; diff --git a/core/services/hdfs/src/docs.md b/core/services/hdfs/src/docs.md index 6ecae53a47e2..9a4114ddaac4 100644 --- a/core/services/hdfs/src/docs.md +++ b/core/services/hdfs/src/docs.md @@ -17,6 +17,14 @@ Depending on its configuration and the backing system, this service can expose: 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 From afc9ff2c2a08f9ee0edd110e57b80a8db5f2de1d Mon Sep 17 00:00:00 2001 From: "zhanghaobo@kanzhun.com" Date: Thu, 24 Sep 2026 09:55:47 +0800 Subject: [PATCH 6/6] ci: retrigger checks