【wip】[common] Keep the shared _temporary directory while other writers stage there

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);
+            }
         }
     }
 }

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 =