From ae8a6aa09903e240e1c16477069b05c96aacd2f3 Mon Sep 17 00:00:00 2001
From: meteorgan <meteorite.gan@gmail.com>
Date: Tue, 3 Sep 2024 16:42:52 +0800
Subject: [PATCH] delete target file before rename_file

---
 core/src/services/hdfs/backend.rs | 26 ++++++++++++++------------
 core/src/services/hdfs/writer.rs  | 15 +++++++++++++++
 2 files changed, 29 insertions(+), 12 deletions(-)

diff --git a/core/src/services/hdfs/backend.rs b/core/src/services/hdfs/backend.rs
index 58e4b18c4a7..4d87d24a108 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 0f60014f410..e4123861a0f 100644
--- a/core/src/services/hdfs/writer.rs
+++ b/core/src/services/hdfs/writer.rs
@@ -29,6 +29,7 @@ pub struct HdfsWriter<F> {
     tmp_path: Option<String>,
     f: Option<F>,
     client: Arc<hdrs::Client>,
+    target_path_exists: bool,
 }
 
 /// # Safety
@@ -42,12 +43,14 @@ impl<F> HdfsWriter<F> {
         tmp_path: Option<String>,
         f: F,
         client: Arc<hdrs::Client>,
+        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<hdrs::AsyncFile> {
 
         // 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<hdrs::File> {
         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)?;