[core][common] Fix format tables reading and deleting files staged by a committer

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,21 @@ 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 is exactly that question, asked without a listing and without a window
+                // between the check and the delete: it succeeds when the directory is empty and
+                // fails while a concurrent writer still has files in it.
+                fileIO.delete(stagingDir, false);
+            } catch (IOException e) {
+                LOG.debug("Staging directory {} is still in use or already gone", stagingDir, e);
+            }
         }
     }
 }

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 =

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java

@@ -128,7 +128,10 @@ public void commit(List<CommitMessage> commitMessages) {
                     partitionSpecs.add(staticPartitions);
                 }
                 if (overwrite) {
-                    deletePreviousDataFile(partitionPath);
+                    // A static partition may name only the leading keys, in which case the path is
+                    // a prefix and the partition directories of the remaining keys sit below it.
+                    deletePreviousDataFile(
+                            partitionPath, partitionKeys.size() - staticPartitions.size());
                 }
                 if (!fileIO.exists(partitionPath)) {
                     fileIO.mkdirs(partitionPath);
@@ -139,7 +142,8 @@ public void commit(List<CommitMessage> commitMessages) {
                     partitionPaths.add(c.targetPath().getParent());
                 }
                 for (Path p : partitionPaths) {
-                    deletePreviousDataFile(p);
+                    // The parent of a written file is always a complete partition directory.
+                    deletePreviousDataFile(p, 0);
                 }
             }
 
@@ -267,17 +271,18 @@ public void abort(List<CommitMessage> commitMessages) {
     @Override
     public void close() throws Exception {}
 
-    private void deletePreviousDataFile(Path partitionPath) throws IOException {
+    private void deletePreviousDataFile(Path partitionPath, int partitionLevels)
+            throws IOException {
         if (fileIO.exists(partitionPath)) {
-            FileStatus[] files = fileIO.listFiles(partitionPath, true);
-            for (FileStatus file : files) {
-                if (FormatTableScan.isDataFileName(file.getPath().getName())) {
-                    try {
-                        fileIO.delete(file.getPath(), false);
-                    } catch (FileNotFoundException ignore) {
-                    } catch (IOException e) {
-                        throw new RuntimeException(e);
-                    }
+            // Committed data files only: what sits under a staging directory belongs to a writer
+            // that is still running.
+            for (FileStatus file :
+                    FormatTableScan.listDataFiles(fileIO, partitionPath, partitionLevels)) {
+                try {
+                    fileIO.delete(file.getPath(), false);
+                } catch (FileNotFoundException ignore) {
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
                 }
             }
         }

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java

@@ -23,6 +23,8 @@
 import org.apache.paimon.casting.CastExecutors;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.partition.PartitionPredicate;
@@ -48,6 +50,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -99,7 +102,69 @@ public InnerTableScan withFilter(Predicate predicate) {
     }
 
     public static boolean isDataFileName(String fileName) {
-        return fileName != null && !fileName.startsWith(".") && !fileName.startsWith("_");
+        return fileName != null && !PartitionPathUtils.isHiddenName(fileName);
+    }
+
+    /**
+     * Lists the data files under {@code listedRoot}, skipping committer staging trees ({@code
+     * _temporary/}, {@code __magic_job-<id>/}, {@code .hive-staging_*}) without descending into
+     * them. Files staged there carry ordinary data file names, so a name alone cannot tell them
+     * apart from committed data; the directory above them can.
+     *
+     * <p>Only entries below {@code listedRoot} are judged, never the root itself, which may
+     * legitimately sit under a warehouse path such as {@code oss://bucket/_warehouse/db/t}.
+     *
+     * @param partitionLevels how many directory levels below {@code listedRoot} hold partition
+     *     directories rather than table content. Those levels are descended whatever they are
+     *     named: a partition value may start with {@code '_'} — the default partition name does, in
+     *     the value-only layout, and {@link PartitionPathUtils} lets the scan read that partition —
+     *     so treating it as staging would leave its files behind. Pass 0 when {@code listedRoot} is
+     *     a single partition or an unpartitioned table.
+     * @throws FileNotFoundException if {@code listedRoot} does not exist; a directory that
+     *     disappears further down is skipped instead, leaving the rest of the listing complete
+     */
+    static List<FileStatus> listDataFiles(FileIO fileIO, Path listedRoot, int partitionLevels)
+            throws IOException {
+        List<FileStatus> dataFiles = new ArrayList<>();
+        List<Path> level = new ArrayList<>();
+        // A missing root is the caller's signal, e.g. a partition that the catalog knows but whose
+        // directory is gone, so let it surface.
+        collectDataFiles(fileIO.listStatus(listedRoot), partitionLevels >= 1, dataFiles, level);
+        for (int depth = 1; !level.isEmpty(); depth++) {
+            boolean childrenArePartitions = partitionLevels >= depth + 1;
+            List<Path> next = new ArrayList<>();
+            for (Path directory : level) {
+                try {
+                    collectDataFiles(
+                            fileIO.listStatus(directory), childrenArePartitions, dataFiles, next);
+                } catch (FileNotFoundException e) {
+                    // The directory vanished after its parent listed it; the rest of the listing
+                    // is still complete.
+                }
+            }
+            level = next;
+        }
+        return dataFiles;
+    }
+
+    private static void collectDataFiles(
+            @Nullable FileStatus[] children,
+            boolean childrenArePartitions,
+            List<FileStatus> dataFiles,
+            List<Path> directories) {
+        if (children == null) {
+            return;
+        }
+        for (FileStatus child : children) {
+            boolean hidden = PartitionPathUtils.isHiddenName(child.getPath().getName());
+            if (child.isDir()) {
+                if (childrenArePartitions || !hidden) {
+                    directories.add(child.getPath());
+                }
+            } else if (!hidden) {
+                dataFiles.add(child);
+            }
+        }
     }
 
     BinaryRow toPartitionRow(LinkedHashMap<String, String> partitionSpec) {

paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java

@@ -40,7 +40,6 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.LinkedHashMap;
@@ -109,12 +108,12 @@ BinaryRow toPartitionRow(LinkedHashMap<String, String> partitionSpec) {
     List<Split> createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition)
             throws IOException {
         List<FormatDataSplit.FileMeta> segments = new ArrayList<>();
-        FileStatus[] files = fileIO.listFiles(path, true);
-        Arrays.sort(files, Comparator.comparing(file -> file.getPath().toString()));
+        // The listed directory is a single partition, or the table itself when unpartitioned:
+        // no partition directories below it.
+        List<FileStatus> files = FormatTableScan.listDataFiles(fileIO, path, 0);
+        files.sort(Comparator.comparing(file -> file.getPath().toString()));
         for (FileStatus file : files) {
-            if (FormatTableScan.isDataFileName(file.getPath().getName())) {
-                segments.addAll(toSegments(file));
-            }
+            segments.addAll(toSegments(file));
         }
 
         List<Split> splits = new ArrayList<>();

paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java

@@ -686,6 +686,15 @@ private static boolean isHiddenFile(
         if (onlyValueInPath && defaultPartValue != null && defaultPartValue.equals(name)) {
             return false;
         }
-        return name.startsWith("_") || name.startsWith(".");
+        return isHiddenName(name);
+    }
+
+    /** Whether a single path component is hidden by the {@code '_'} / {@code '.'} convention. */
+    public static boolean isHiddenName(String name) {
+        return name != null && !name.isEmpty() && isHiddenFirstChar(name.charAt(0));
+    }
+
+    private static boolean isHiddenFirstChar(char c) {
+        return c == '_' || c == '.';
     }
 }

paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java

@@ -180,7 +180,6 @@ void testListPartitionEntriesUsesCatalogVisibility() throws Exception {
         assertThat(entries.get(0).lastFileCreationTime()).isEqualTo(44L);
         assertThat(entries.get(0).totalBuckets()).isEqualTo(5);
         assertThat(fileIO.listedPaths).isEmpty();
-        assertThat(fileIO.statusListedPaths).isEmpty();
     }
 
     @Test
@@ -311,7 +310,7 @@ void testMissingCatalogRegisteredPartitionReadsAsEmpty() throws Exception {
         LocalFileIO fileIO =
                 new LocalFileIO() {
                     @Override
-                    public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+                    public FileStatus[] listStatus(Path path) throws IOException {
                         assertThat(path).isEqualTo(missingPath);
                         throw new FileNotFoundException(path.toString());
                     }
@@ -497,18 +496,11 @@ private static List<Path> plannedFiles(List<Split> splits) {
     private static class TrackingLocalFileIO extends LocalFileIO {
 
         private final List<Path> listedPaths = new ArrayList<>();
-        private final List<Path> statusListedPaths = new ArrayList<>();
 
         @Override
         public FileStatus[] listStatus(Path path) throws IOException {
-            statusListedPaths.add(path);
-            return super.listStatus(path);
-        }
-
-        @Override
-        public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
             listedPaths.add(path);
-            return super.listFiles(path, recursive);
+            return super.listStatus(path);
         }
     }
 
@@ -546,7 +538,7 @@ void testParallelListingSkipsMissingCatalogRegisteredPartition() throws Exceptio
                         partition("2025", "10"), partition("2025", "11"), partition("2025", "12"));
         Path oct = writeDataFile(fileIO, tablePath, "year=2025/month=10");
         Path dec = writeDataFile(fileIO, tablePath, "year=2025/month=12");
-        fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing"));
+        fileIO.failListingContaining("month=11", new FileNotFoundException("missing"));
 
         List<Path> files =
                 plannedFiles(
@@ -569,7 +561,7 @@ void testCatalogPartitionListingIoErrorFailsWholeScan() throws Exception {
                 Arrays.asList(partition("2025", "10"), partition("2025", "11"));
         writeDataFile(fileIO, tablePath, "year=2025/month=10");
         writeDataFile(fileIO, tablePath, "year=2025/month=11");
-        fileIO.failListFilesContaining("month=11", new IOException("boom"));
+        fileIO.failListingContaining("month=11", new IOException("boom"));
 
         FormatTable table =
                 stringPartitionTable(fileIO, tablePath, recordingCatalog(partitions), 4);
@@ -585,7 +577,7 @@ void testFilesystemDiscoveredMissingDirectoryStillFailsScan() throws Exception {
         Path tablePath = new Path(tempDir.toUri());
         writeDataFile(fileIO, tablePath, "year=2025/month=10");
         writeDataFile(fileIO, tablePath, "year=2025/month=11");
-        fileIO.failListFilesContaining("month=11", new FileNotFoundException("missing"));
+        fileIO.failListingContaining("month=11", new FileNotFoundException("missing"));
 
         FormatTable table = createStringPartitionTable(fileIO, tablePath, null);
         assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
@@ -670,15 +662,15 @@ private static class ParallelTrackingLocalFileIO extends LocalFileIO {
         private final AtomicInteger maxConcurrentListings = new AtomicInteger();
 
         @Override
-        public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+        public FileStatus[] listStatus(Path path) throws IOException {
             int active = activeListings.incrementAndGet();
             maxConcurrentListings.updateAndGet(current -> Math.max(current, active));
             concurrentListings.countDown();
             try {
                 if (!concurrentListings.await(10, TimeUnit.SECONDS)) {
                     throw new IOException("Partition file listings did not run concurrently");
                 }
-                return super.listFiles(path, recursive);
+                return super.listStatus(path);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException("Interrupted while waiting for concurrent listings", e);
@@ -698,12 +690,12 @@ private static class SerialTrackingLocalFileIO extends LocalFileIO {
         private final AtomicInteger maxConcurrentListings = new AtomicInteger();
 
         @Override
-        public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+        public FileStatus[] listStatus(Path path) throws IOException {
             int active = activeListings.incrementAndGet();
             maxConcurrentListings.updateAndGet(current -> Math.max(current, active));
             try {
                 Thread.sleep(50);
-                return super.listFiles(path, recursive);
+                return super.listStatus(path);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException("Interrupted while tracking file listings", e);
@@ -721,18 +713,18 @@ private static class InjectingLocalFileIO extends LocalFileIO {
 
         private final Map<String, IOException> listFailures = new LinkedHashMap<>();
 
-        void failListFilesContaining(String marker, IOException failure) {
+        void failListingContaining(String marker, IOException failure) {
             listFailures.put(marker, failure);
         }
 
         @Override
-        public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+        public FileStatus[] listStatus(Path path) throws IOException {
             for (Map.Entry<String, IOException> entry : listFailures.entrySet()) {
                 if (path.toString().contains(entry.getKey())) {
                     throw entry.getValue();
                 }
             }
-            return super.listFiles(path, recursive);
+            return super.listStatus(path);
         }
     }