From 7d6321d401afd70ccbdeaad60aacabe2dac214eb Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 29 Jul 2026 11:10:02 +0800 Subject: [PATCH 1/4] [common] Keep the shared _temporary directory while other writers stage there RenamingTwoPhaseOutputStream stages into /_temporary/.tmp., which is the same directory Hadoop's FileOutputCommitter and everything built on it use. clean() removed that directory recursively, so a Paimon commit deleted whatever another job had staged there and that job failed to commit its own files. Delete only the file this committer staged, then ask whether the directory can go by attempting a non-recursive delete: it removes the directory when nothing is staged there any more and refuses while a concurrent writer still has files in it. Removing it stays best-effort - a writer that stages here next recreates it - so a failure is logged at debug and the commit is unaffected. --- .../fs/RenamingTwoPhaseOutputStream.java | 27 +++++++++++++- .../fs/RenamingTwoPhaseOutputStreamTest.java | 35 +++++++++++++++++++ 2 files changed, 61 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..e6d604e53ca9 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,26 @@ public Path targetPath() { @Override public void clean(FileIO fileIO) { - fileIO.deleteDirectoryQuietly(tempPath.getParent()); + fileIO.deleteQuietly(tempPath); + Path stagingDir = tempPath.getParent(); + if (stagingDir == null) { + return; + } + try { + // '_temporary' is shared with every other writer of this directory, Paimon or not, + // so it may only go away once nothing is staged there any more. A non-recursive + // delete asks exactly that: it removes the directory when it is empty and refuses + // while a concurrent writer still has files in it. Removing it is best-effort -- + // a writer that stages here next recreates it anyway. + fileIO.delete(stagingDir, false); + } catch (IOException e) { + LOG.debug( + "Left staging directory {} in place: the non-recursive delete did not" + + " succeed. Usually another writer still has files staged there," + + " or the directory is already gone; see the attached exception.", + 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..d47bbbe4eeae 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,41 @@ 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(); + + // 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 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 = From ac0564b49a208655bfc0b6fe4a0a671f6b08bde1 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 29 Jul 2026 15:01:29 +0800 Subject: [PATCH 2/4] [common] Leave the shared _temporary directory to its owners Removing the directory when it looks empty is not safe: a writer that has just created '_temporary' has not staged its file in it yet, and deleting it from under that writer makes its open fail. Empty is not the same as unused. Delete only the file this committer staged. The directory outlives the commit, which is what every other writer of the same location already assumes. --- .../fs/RenamingTwoPhaseOutputStream.java | 28 +++---------------- .../fs/RenamingTwoPhaseOutputStreamTest.java | 23 +++++++++++++-- 2 files changed, 24 insertions(+), 27 deletions(-) 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 e6d604e53ca9..a8547bd1673b 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,9 +20,6 @@ import org.apache.paimon.annotation.Public; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.io.IOException; import java.util.UUID; @@ -33,8 +30,6 @@ @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; @@ -142,26 +137,11 @@ public Path targetPath() { @Override public void clean(FileIO fileIO) { + // 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); - Path stagingDir = tempPath.getParent(); - if (stagingDir == null) { - return; - } - try { - // '_temporary' is shared with every other writer of this directory, Paimon or not, - // so it may only go away once nothing is staged there any more. A non-recursive - // delete asks exactly that: it removes the directory when it is empty and refuses - // while a concurrent writer still has files in it. Removing it is best-effort -- - // a writer that stages here next recreates it anyway. - fileIO.delete(stagingDir, false); - } catch (IOException e) { - LOG.debug( - "Left staging directory {} in place: the non-recursive delete did not" - + " succeed. Usually another writer still has files staged there," - + " or the directory is already gone; see the attached exception.", - 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 d47bbbe4eeae..3137c29dfad7 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 @@ -92,18 +92,35 @@ void testCleanKeepsTheSharedStagingDirectory() throws IOException { } @Test - void testCleanRemovesTheStagingDirectoryOnceEmpty() throws IOException { + void testCleanLeavesTheStagingDirectoryForItsOwners() 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); - // Nothing else was staged, so the writer leaves no trace beyond its committed file. + // 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. assertThat(fileIO.exists(targetPath)).isTrue(); - assertThat(fileIO.exists(new Path(targetPath.getParent(), "_temporary"))).isFalse(); + 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); + + assertThat(fileIO.exists(targetPath)).isFalse(); + assertThat(fileIO.listStatus(new Path(targetPath.getParent(), "_temporary"))).isEmpty(); } @Test From 736e2c84ac1acd8ea31289e4dddf3439bfc410cf Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 29 Jul 2026 15:36:33 +0800 Subject: [PATCH 3/4] [common] Pin the empty staging directory, and write the rule on the interface The test that named the directory it protects could not see it disappear: FileIO.listStatus answers with no entries for a directory that is gone as much as for one that is empty, so asserting the listing is empty passed either way. Deleting the empty shared directory - the very race this change is about - left the whole suite green. Assert exists() instead, in both new tests. Committer.clean had no javadoc at all, so the rule this change establishes lived only in one implementation's comment: the next implementer of an @Public extension point could reintroduce the same deletion and read nothing that says otherwise. State it on the interface, and say when clean runs as opposed to discard. Also name the two tests after what separates them, an empty staging directory against one that holds another writer's file. --- .../org/apache/paimon/fs/TwoPhaseOutputStream.java | 14 +++++++++++++- .../fs/RenamingTwoPhaseOutputStreamTest.java | 11 ++++++++--- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java index 5f85b087d413..931969ec68cb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java @@ -46,7 +46,8 @@ 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 */ @@ -54,6 +55,17 @@ public interface Committer extends Serializable { Path targetPath(); + /** + * Releases what this committer staged and no longer needs, after {@link #commit} has + * succeeded. May do nothing. + * + *

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; } } 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 3137c29dfad7..2fc4fef14fbc 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 @@ -72,7 +72,7 @@ void testSuccessfulCommit() throws IOException { } @Test - void testCleanKeepsTheSharedStagingDirectory() throws IOException { + void testCleanKeepsAStagingDirectoryHoldingAnotherWritersFile() throws IOException { RenamingTwoPhaseOutputStream stream = new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); stream.write("Some data".getBytes()); @@ -92,7 +92,7 @@ void testCleanKeepsTheSharedStagingDirectory() throws IOException { } @Test - void testCleanLeavesTheStagingDirectoryForItsOwners() throws IOException { + void testCleanKeepsAStagingDirectoryThatIsEmpty() throws IOException { RenamingTwoPhaseOutputStream stream = new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); stream.write("Some data".getBytes()); @@ -104,7 +104,10 @@ void testCleanLeavesTheStagingDirectoryForItsOwners() throws IOException { // 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(); } @@ -119,8 +122,10 @@ void testCleanRemovesTheFileItStagedWhenThereWasNoCommit() throws IOException { // behind for good. committer.clean(fileIO); + Path stagingDir = new Path(targetPath.getParent(), "_temporary"); assertThat(fileIO.exists(targetPath)).isFalse(); - assertThat(fileIO.listStatus(new Path(targetPath.getParent(), "_temporary"))).isEmpty(); + assertThat(fileIO.exists(stagingDir)).isTrue(); + assertThat(fileIO.listStatus(stagingDir)).isEmpty(); } @Test From 4c0c1feeddc50f492873064290f5a46e84d07806 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 29 Jul 2026 17:21:15 +0800 Subject: [PATCH 4/4] [spark] Do not take a staging directory for a data file in the compressed-read test The test picks the first entry of the table directory and gzips it, excluding only '.'-prefixed names. Now that a writer leaves its '_temporary' behind, that entry can be the staging directory, and the test fails opening a directory as a file. Exclude what the reader excludes. --- .../org/apache/paimon/spark/sql/FormatTableTestBase.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala index 720d28320420..652fb9be4db9 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala +++ b/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