diff --git a/core/src/services/hdfs/backend.rs b/core/src/services/hdfs/backend.rs index 754bef088212..c29333a5b370 100644 --- a/core/src/services/hdfs/backend.rs +++ b/core/src/services/hdfs/backend.rs @@ -313,11 +313,6 @@ impl Access for HdfsBackend { if !target_exists { let parent = get_parent(&target_path); self.client.create_dir(parent).map_err(new_std_io_error)?; - } else if tmp_path.is_some() { - // we must delete the target_path, otherwise the rename_file operation will fail - self.client - .remove_file(&target_path) - .map_err(new_std_io_error)?; } let mut open_options = self.client.open_file(); @@ -335,7 +330,13 @@ impl Access for HdfsBackend { Ok(( RpWrite::new(), - HdfsWriter::new(target_path, tmp_path, f, Arc::clone(&self.client)), + HdfsWriter::new( + target_path, + tmp_path, + f, + Arc::clone(&self.client), + target_exists, + ), )) } @@ -507,11 +508,6 @@ impl Access for HdfsBackend { if !target_exists { let parent = get_parent(&target_path); self.client.create_dir(parent).map_err(new_std_io_error)?; - } else if tmp_path.is_some() { - // we must delete the target_path, otherwise the rename_file operation will fail - self.client - .remove_file(&target_path) - .map_err(new_std_io_error)?; } let mut open_options = self.client.open_file(); @@ -528,7 +524,13 @@ impl Access for HdfsBackend { Ok(( RpWrite::new(), - HdfsWriter::new(target_path, tmp_path, f, Arc::clone(&self.client)), + HdfsWriter::new( + target_path, + tmp_path, + f, + Arc::clone(&self.client), + target_exists, + ), )) } diff --git a/core/src/services/hdfs/writer.rs b/core/src/services/hdfs/writer.rs index 0f60014f410d..e4123861a0f4 100644 --- a/core/src/services/hdfs/writer.rs +++ b/core/src/services/hdfs/writer.rs @@ -29,6 +29,7 @@ pub struct HdfsWriter { tmp_path: Option, f: Option, client: Arc, + target_path_exists: bool, } /// # Safety @@ -42,12 +43,14 @@ impl HdfsWriter { tmp_path: Option, f: F, client: Arc, + target_path_exists: bool, ) -> Self { Self { target_path, tmp_path, f: Some(f), client, + target_path_exists, } } } @@ -70,6 +73,12 @@ impl oio::Write for HdfsWriter { // TODO: we need to make rename async. if let Some(tmp_path) = &self.tmp_path { + // we must delete the target_path, otherwise the rename_file operation will fail + if self.target_path_exists { + self.client + .remove_file(&self.target_path) + .map_err(new_std_io_error)?; + } self.client .rename_file(tmp_path, &self.target_path) .map_err(new_std_io_error)? @@ -102,6 +111,12 @@ impl oio::BlockingWrite for HdfsWriter { f.flush().map_err(new_std_io_error)?; if let Some(tmp_path) = &self.tmp_path { + // we must delete the target_path, otherwise the rename_file operation will fail + if self.target_path_exists { + self.client + .remove_file(&self.target_path) + .map_err(new_std_io_error)?; + } self.client .rename_file(tmp_path, &self.target_path) .map_err(new_std_io_error)?;