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

paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java

@@ -29,6 +29,7 @@
  */
 @Public
 public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream {
+
     private static final String TEMP_DIR_NAME = "_temporary";
 
     private final Path targetPath;
@@ -136,7 +137,11 @@ public Path targetPath() {
 
         @Override
         public void clean(FileIO fileIO) {
-            fileIO.deleteDirectoryQuietly(tempPath.getParent());
+            // Only what this committer staged. '_temporary' is shared with every other writer of
+            // this directory, Paimon or not, and it is theirs to remove: seeing it empty does not
+            // mean it is unused, because a writer that has just created it has not staged its file
+            // yet, and deleting it from under that writer fails its open.
+            fileIO.deleteQuietly(tempPath);
         }
     }
 }

paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java

@@ -46,14 +46,26 @@ public interface Committer extends Serializable {
         void commit(FileIO fileIO) throws IOException;
 
         /**
-         * Discards the written data, cleaning up any temporary files or resources.
+         * Discards the written data, cleaning up any temporary files or resources. Called instead
+         * of {@link #commit} when the write is given up.
          *
          * @throws IOException if an I/O error occurs during discard
          */
         void discard(FileIO fileIO) throws IOException;
 
         Path targetPath();
 
+        /**
+         * Releases what this committer staged and no longer needs, after {@link #commit} has
+         * succeeded. May do nothing.
+         *
+         * <p>Only resources this committer created itself. A staging directory is shared with every
+         * other writer of the same location, Paimon or not, and removing it is theirs to decide:
+         * finding it empty does not mean it is unused, because a writer that has just created it
+         * has not staged its file in it yet.
+         *
+         * @throws IOException if an I/O error occurs during cleaning
+         */
         void clean(FileIO fileIO) throws IOException;
     }
 }

paimon-common/src/test/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStreamTest.java

@@ -71,6 +71,63 @@ void testSuccessfulCommit() throws IOException {
         assertThat(new String(content)).isEqualTo(testData);
     }
 
+    @Test
+    void testCleanKeepsAStagingDirectoryHoldingAnotherWritersFile() throws IOException {
+        RenamingTwoPhaseOutputStream stream =
+                new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+        stream.write("Some data".getBytes());
+        TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+
+        // A MapReduce-style writer with a task attempt still pending in the same directory.
+        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 testCleanKeepsAStagingDirectoryThatIsEmpty() throws IOException {
+        RenamingTwoPhaseOutputStream stream =
+                new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+        stream.write("Some data".getBytes());
+        TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+        Path stagingDir = new Path(targetPath.getParent(), "_temporary");
+
+        committer.commit(fileIO);
+        committer.clean(fileIO);
+
+        // Empty is not the same as unused: a writer that has just created '_temporary' has not
+        // staged its file in it yet, and removing the directory would fail that writer's open.
+        // exists(), not listStatus(): listStatus answers with no entries for a directory that is
+        // gone as much as for one that is empty, so it cannot tell the two apart.
+        assertThat(fileIO.exists(targetPath)).isTrue();
+        assertThat(fileIO.exists(stagingDir)).isTrue();
+        assertThat(fileIO.listStatus(stagingDir)).isEmpty();
+    }
+
+    @Test
+    void testCleanRemovesTheFileItStagedWhenThereWasNoCommit() throws IOException {
+        RenamingTwoPhaseOutputStream stream =
+                new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+        stream.write("Some data".getBytes());
+        TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+
+        // No commit renamed it away, so clean() is what keeps the staged file from being left
+        // behind for good.
+        committer.clean(fileIO);
+
+        Path stagingDir = new Path(targetPath.getParent(), "_temporary");
+        assertThat(fileIO.exists(targetPath)).isFalse();
+        assertThat(fileIO.exists(stagingDir)).isTrue();
+        assertThat(fileIO.listStatus(stagingDir)).isEmpty();
+    }
+
     @Test
     void testDiscard() throws IOException {
         RenamingTwoPhaseOutputStream stream =

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala

@@ -25,7 +25,7 @@ import org.apache.paimon.spark.{PaimonFormatTableScan, PaimonHiveTestBase, Paimo
 import org.apache.paimon.spark.PaimonHiveTestBase.hiveUri
 import org.apache.paimon.table.FormatTable
 import org.apache.paimon.table.source.Split
-import org.apache.paimon.utils.CompressUtils
+import org.apache.paimon.utils.{CompressUtils, PartitionPathUtils}
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.connector.read.InputPartition
@@ -214,7 +214,9 @@ abstract class FormatTableTestBase extends PaimonHiveTestBase with AdaptiveSpark
         val fileIO = table.fileIO()
         val file = fileIO
           .listStatus(new Path(table.location()))
-          .filter(file => !file.getPath.getName.startsWith("."))
+          // The same rule the reader applies: a writer's staging directory stays behind, and it
+          // is not a data file.
+          .filter(file => !PartitionPathUtils.isHiddenName(file.getPath.getName))
           .head
           .getPath
           .toUri