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..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 @@ -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); } } } 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 2b929f18ca1b..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 @@ -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 = 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