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 =