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 c3cc973fe44e..a97eef36a27d 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; type Deleter = oio::OneShotDeleter; - 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 { - 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 84a99c5a5404..8b2907f1d67f 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 { + 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 97765d3f5ed9..9a4114ddaac4 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