This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9bcbbc4a67 [common] Keep the shared _temporary directory while other
writers stage there (#8900)
9bcbbc4a67 is described below
commit 9bcbbc4a67f44cf52803b5578d6fe330ad3ee3ab
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Jul 30 12:01:48 2026 +0800
[common] Keep the shared _temporary directory while other writers stage
there (#8900)
---
.../paimon/fs/RenamingTwoPhaseOutputStream.java | 7 ++-
.../org/apache/paimon/fs/TwoPhaseOutputStream.java | 14 +++++-
.../fs/RenamingTwoPhaseOutputStreamTest.java | 57 ++++++++++++++++++++++
.../paimon/spark/sql/FormatTableTestBase.scala | 6 ++-
4 files changed, 80 insertions(+), 4 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 f5dcdf7a1b..a8547bd167 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 @@ import java.util.UUID;
*/
@Public
public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream {
+
private static final String TEMP_DIR_NAME = "_temporary";
private final Path targetPath;
@@ -136,7 +137,11 @@ public class RenamingTwoPhaseOutputStream extends
TwoPhaseOutputStream {
@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 5f85b087d4..931969ec68 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 abstract class TwoPhaseOutputStream extends
PositionOutputStream {
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 abstract class TwoPhaseOutputStream extends
PositionOutputStream {
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;
}
}
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 2b929f18ca..2fc4fef14f 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 @@ public class RenamingTwoPhaseOutputStreamTest {
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 720d283204..652fb9be4d 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
