Xuanwo commented on code in PR #3763:
URL:
https://github.com/apache/incubator-opendal/pull/3763#discussion_r1432179732
##########
core/src/services/hdfs/backend.rs:
##########
@@ -231,43 +319,44 @@ impl Accessor for HdfsBackend {
}
async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite,
Self::Writer)> {
- let p = build_rooted_abs_path(&self.root, path);
-
- if let Err(err) = self.client.metadata(&p) {
- // Early return if other error happened.
- if err.kind() != io::ErrorKind::NotFound {
- return Err(new_std_io_error(err));
- }
-
- let parent = get_parent(&p);
-
- self.client.create_dir(parent).map_err(new_std_io_error)?;
-
- let mut f = self
- .client
- .open_file()
- .create(true)
- .write(true)
- .async_open(&p)
- .await
- .map_err(new_std_io_error)?;
- f.close().await.map_err(new_std_io_error)?;
- }
+ let (target_path, tmp_path) = if let Some(atomic_write_dir) =
&self.atomic_write_dir {
+ let target_path = Self::write_abs_path(self, &self.root, path)?;
+ let tmp_path = Self::ensure_write_abs_path(
+ self,
+ atomic_write_dir,
+ &tmp_file_of(path),
+ ).await?;
+ (target_path,Some(tmp_path))
+ } else {
+ let p = Self::ensure_write_abs_path(self, &self.root,
path).await?;
+ (p, None)
+ };
let mut open_options = self.client.open_file();
- open_options.create(true);
+ open_options.create(true).write(true);
+
if op.append() {
open_options.append(true);
- } else {
- open_options.write(true);
}
- let f = open_options
- .async_open(&p)
- .await
- .map_err(new_std_io_error)?;
+ if let Some(tmp_path) =&tmp_path {
+ let f = open_options
+ .async_open(tmp_path)
+ .await
+ .map_err(new_std_io_error)?;
+
+ let w =
HdfsWriter::new(target_path,Option::from(tmp_path.to_string()),
self.client.clone(),f);
Review Comment:
Hi, as I reviewed before, we need to handle the atomic wirte logic in
`HdfsWriter` as we do for `fs` service.
##########
core/src/services/hdfs/backend.rs:
##########
@@ -436,7 +525,9 @@ impl Accessor for HdfsBackend {
let f = open_options.open(&p).map_err(new_std_io_error)?;
- Ok((RpWrite::new(), HdfsWriter::new(f)))
+ let w=HdfsWriter::new("".to_string(),None,self.client.clone(),f);
Review Comment:
Seems not related to this PR. How about removing this change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
