From f69f8329765196ca68dfb9ca15c020f0f03db5e0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?=
 <dapeng.sdp@alibaba-inc.com>
Date: Mon, 27 Jul 2026 17:16:46 +0800
Subject: [PATCH] [common] Keep the shared _temporary directory while other
 writers stage there

TempFileCommitter#clean deleted the whole '_temporary' directory of the target
directory after committing one file. That directory is shared with every other
writer: two concurrent writers into the same directory wiped each other's
pending files, and a Hadoop FileOutputCommitter job staging there lost its
output. Delete this stream's own temporary file, and the directory itself only
once nothing is staged there any more.
---
 .../fs/RenamingTwoPhaseOutputStream.java      | 25 ++++++++++++-
 .../fs/RenamingTwoPhaseOutputStreamTest.java  | 36 +++++++++++++++++++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
index f5dcdf7a1b28..0d2cf05643b1 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
@@ -20,6 +20,9 @@
 
 import org.apache.paimon.annotation.Public;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.util.UUID;
 
@@ -29,6 +32,9 @@
  */
 @Public
 public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream {
+
+    private static final Logger LOG = LoggerFactory.getLogger(RenamingTwoPhaseOutputStream.class);
+
     private static final String TEMP_DIR_NAME = "_temporary";
 
     private final Path targetPath;
@@ -136,7 +142,24 @@ public Path targetPath() {
 
         @Override
         public void clean(FileIO fileIO) {
-            fileIO.deleteDirectoryQuietly(tempPath.getParent());
+            fileIO.deleteQuietly(tempPath);
+            // '_temporary' is shared with every other writer of this directory, Paimon or not, so
+            // it can only be removed once nothing is staged there any more: deleting it outright
+            // would throw away the pending files of jobs that are still running.
+            Path stagingDir = tempPath.getParent();
+            if (stagingDir == null) {
+                return;
+            }
+            try {
+                FileStatus[] staged = fileIO.listStatus(stagingDir);
+                if (staged != null && staged.length == 0) {
+                    fileIO.deleteQuietly(stagingDir);
+                }
+            } catch (IOException e) {
+                // Already gone, or not listable: an empty staging directory left behind is
+                // skipped by readers anyway.
+                LOG.debug("Could not check whether {} is still in use", stagingDir, e);
+            }
         }
     }
 }
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java
index 2b929f18ca1b..991a8f3f5796 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java
@@ -71,6 +71,42 @@ void testSuccessfulCommit() throws IOException {
         assertThat(new String(content)).isEqualTo(testData);
     }
 
+    @Test
+    void testCleanKeepsTheSharedStagingDirectory() throws IOException {
+        RenamingTwoPhaseOutputStream stream =
+                new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+        stream.write("Some data".getBytes());
+        TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+
+        // '_temporary' is shared with every other writer of this directory, Paimon or not: a
+        // concurrent job's pending files must survive this stream's cleanup.
+        Path otherWriterPending =
+                new Path(targetPath.getParent(), "_temporary/attempt_0001_m_000010_15/part-00010");
+        fileIO.writeFile(otherWriterPending, "concurrent", false);
+
+        committer.commit(fileIO);
+        committer.clean(fileIO);
+
+        assertThat(fileIO.exists(targetPath)).isTrue();
+        assertThat(fileIO.exists(otherWriterPending)).isTrue();
+        assertThat(fileIO.exists(new Path(targetPath.getParent(), "_temporary"))).isTrue();
+    }
+
+    @Test
+    void testCleanRemovesTheStagingDirectoryOnceEmpty() throws IOException {
+        RenamingTwoPhaseOutputStream stream =
+                new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+        stream.write("Some data".getBytes());
+        TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+
+        committer.commit(fileIO);
+        committer.clean(fileIO);
+
+        // Nothing else was staged, so the writer leaves no trace beyond its committed file.
+        assertThat(fileIO.exists(targetPath)).isTrue();
+        assertThat(fileIO.exists(new Path(targetPath.getParent(), "_temporary"))).isFalse();
+    }
+
     @Test
     void testDiscard() throws IOException {
         RenamingTwoPhaseOutputStream stream =