[spark] Clean up orphan files left by Spark speculative execution

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -294,6 +294,25 @@ default void deleteQuietly(Path file) {
         }
     }
 
+    /**
+     * Same as {@link #deleteQuietly(Path)}, but temporarily clears the calling thread's interrupt
+     * status so Hadoop/RPC deletes are not rejected with {@link java.io.InterruptedIOException}.
+     *
+     * <p>Use this only for abort/cleanup paths (e.g. speculative task kill) where best-effort
+     * deletion must succeed even though the task thread has been interrupted. The previous
+     * interrupt status is restored afterwards.
+     */
+    default void deleteQuietlyIgnoringInterrupt(Path file) {
+        boolean interrupted = Thread.interrupted();
+        try {
+            deleteQuietly(file);
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
     default void deleteFilesQuietly(List<Path> files) {
         for (Path file : files) {
             deleteQuietly(file);

paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java

@@ -29,6 +29,7 @@
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.nio.file.AccessDeniedException;
 import java.nio.file.DirectoryNotEmptyException;
 import java.nio.file.FileAlreadyExistsException;
@@ -138,6 +139,27 @@ public static void testOverwriteFileUtf8(Path file, FileIO fileIO) throws Interr
         assertThat(exception.get()).isNull();
     }
 
+    @Test
+    public void testDeleteQuietlyIgnoringInterrupt() throws Exception {
+        Path file = new Path(tempDir.resolve("interrupted-delete.txt").toUri());
+        LocalFileIO local = new LocalFileIO();
+        local.writeFile(file, "data", false);
+
+        FileIO interruptSensitive = new InterruptSensitiveFileIO(local);
+        Thread.currentThread().interrupt();
+        try {
+            // Plain deleteQuietly fails under interrupt with Hadoop-like FileIO.
+            interruptSensitive.deleteQuietly(file);
+            assertThat(local.exists(file)).isTrue();
+
+            interruptSensitive.deleteQuietlyIgnoringInterrupt(file);
+            assertThat(Thread.currentThread().isInterrupted()).isTrue();
+            assertThat(local.exists(file)).isFalse();
+        } finally {
+            Thread.interrupted();
+        }
+    }
+
     @Test
     public void testListFiles() throws Exception {
         Path fileA = new Path(tempDir.resolve("a").toUri());
@@ -198,6 +220,84 @@ public void testDefaultArchiveUnsupported() {
                 .hasMessageContaining("unarchive");
     }
 
+    /**
+     * Delegates to another {@link FileIO}, but mimics Hadoop RPC behavior by failing with {@link
+     * InterruptedIOException} when the calling thread is interrupted.
+     */
+    private static class InterruptSensitiveFileIO implements FileIO {
+
+        private final FileIO delegate;
+
+        private InterruptSensitiveFileIO(FileIO delegate) {
+            this.delegate = delegate;
+        }
+
+        private void failIfInterrupted() throws InterruptedIOException {
+            if (Thread.currentThread().isInterrupted()) {
+                throw new InterruptedIOException("Call interrupted");
+            }
+        }
+
+        @Override
+        public boolean isObjectStore() {
+            return delegate.isObjectStore();
+        }
+
+        @Override
+        public void configure(CatalogContext context) {
+            delegate.configure(context);
+        }
+
+        @Override
+        public SeekableInputStream newInputStream(Path path) throws IOException {
+            failIfInterrupted();
+            return delegate.newInputStream(path);
+        }
+
+        @Override
+        public PositionOutputStream newOutputStream(Path path, boolean overwrite)
+                throws IOException {
+            failIfInterrupted();
+            return delegate.newOutputStream(path, overwrite);
+        }
+
+        @Override
+        public FileStatus getFileStatus(Path path) throws IOException {
+            failIfInterrupted();
+            return delegate.getFileStatus(path);
+        }
+
+        @Override
+        public FileStatus[] listStatus(Path path) throws IOException {
+            failIfInterrupted();
+            return delegate.listStatus(path);
+        }
+
+        @Override
+        public boolean exists(Path path) throws IOException {
+            failIfInterrupted();
+            return delegate.exists(path);
+        }
+
+        @Override
+        public boolean delete(Path path, boolean recursive) throws IOException {
+            failIfInterrupted();
+            return delegate.delete(path, recursive);
+        }
+
+        @Override
+        public boolean mkdirs(Path path) throws IOException {
+            failIfInterrupted();
+            return delegate.mkdirs(path);
+        }
+
+        @Override
+        public boolean rename(Path src, Path dst) throws IOException {
+            failIfInterrupted();
+            return delegate.rename(src, dst);
+        }
+    }
+
     /** A {@link FileIO} on local filesystem to test various default implementations. */
     private static class DummyFileIO implements FileIO {
         private static final ReentrantLock RENAME_LOCK = new ReentrantLock();

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -279,16 +279,31 @@ public void close() throws Exception {
         sync();
 
         compactManager.close();
+
+        // Delete newly generated but not committed files (matches RecordWriter.close contract /
+        // MergeTreeWriter). Includes buffer-flush files that never reached prepareCommit.
+        List<DataFileMeta> toDelete = new ArrayList<>(newFiles);
+        newFiles.clear();
+        deletedFiles.clear();
         for (DataFileMeta file : compactAfter) {
             // appendOnlyCompactManager will rewrite the file and no file upgrade will occur, so we
             // can directly delete the file in compactAfter.
-            fileIO.deleteQuietly(pathFactory.toPath(file));
+            toDelete.add(file);
         }
+        compactAfter.clear();
+        compactBefore.clear();
 
-        sinkWriter.close();
-
-        if (compactDeletionFile != null) {
-            compactDeletionFile.clean();
+        // Abort in-flight rolling writer even if delete loops fail.
+        try {
+            sinkWriter.close();
+        } finally {
+            for (DataFileMeta file : toDelete) {
+                // Interrupt-tolerant: close runs after speculative task kill.
+                fileIO.deleteQuietlyIgnoringInterrupt(pathFactory.toPath(file));
+            }
+            if (compactDeletionFile != null) {
+                compactDeletionFile.clean();
+            }
         }
     }
 
@@ -309,7 +324,7 @@ public void toBufferedWriter() throws Exception {
             } finally {
                 // remove small files
                 for (DataFileMeta file : files) {
-                    fileIO.deleteQuietly(pathFactory.toPath(file));
+                    fileIO.deleteQuietlyIgnoringInterrupt(pathFactory.toPath(file));
                 }
             }
         }

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java

@@ -33,9 +33,6 @@
 import org.apache.paimon.utils.RecordWriter;
 import org.apache.paimon.utils.SetUtils;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -48,8 +45,6 @@
 /** Compacts normal structured files of a data evolution table. */
 public class DataEvolutionNormalCompactTask extends DataEvolutionCompactTask {
 
-    private static final Logger LOG = LoggerFactory.getLogger(DataEvolutionNormalCompactTask.class);
-
     public DataEvolutionNormalCompactTask(BinaryRow partition, List<DataFileMeta> files) {
         super(partition, files);
     }
@@ -98,32 +93,51 @@ public CommitMessage doCompact(FileStoreTable table, String commitUser) throws E
         storeWrite.withWriteType(readWriteType);
         RecordWriter<InternalRow> writer = storeWrite.createWriter(partition, 0);
 
-        reader.forEachRemaining(
-                row -> {
-                    try {
-                        writer.write(row);
-                    } catch (Exception e) {
-                        throw new RuntimeException(e);
-                    }
-                });
-
-        List<DataFileMeta> writeResult = writer.prepareCommit(false).newFilesIncrement().newFiles();
-        checkArgument(
-                writeResult.size() == 1, "Data evolution compaction should produce one file.");
-
         try {
-            writer.close();
-            storeWrite.close();
-        } catch (Exception e) {
-            LOG.warn("Failed to close reader and writer.", e);
+            reader.forEachRemaining(
+                    row -> {
+                        try {
+                            writer.write(row);
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    });
+
+            List<DataFileMeta> writeResult =
+                    writer.prepareCommit(false).newFilesIncrement().newFiles();
+            checkArgument(
+                    writeResult.size() == 1, "Data evolution compaction should produce one file.");
+
+            DataFileMeta dataFileMeta = writeResult.get(0).assignFirstRowId(firstRowId);
+            dataFileMeta =
+                    dataFileMeta.assignSequenceNumber(
+                            minSequenceId(compactBefore), maxSequenceId(compactBefore));
+            compactAfter.add(dataFileMeta);
+        } finally {
+            // Close on the failure path too: RecordWriter.close deletes flushed-but-unprepared
+            // files, so a mid-task failure (e.g. speculative kill) does not leave orphans.
+            // Each close runs independently so a reader failure cannot skip writer cleanup.
+            boolean interrupted = Thread.interrupted();
+            try {
+                try {
+                    reader.close();
+                } catch (Exception ignored) {
+                }
+                try {
+                    writer.close();
+                } catch (Exception ignored) {
+                }
+                try {
+                    storeWrite.close();
+                } catch (Exception ignored) {
+                }
+            } finally {
+                if (interrupted) {
+                    Thread.currentThread().interrupt();
+                }
+            }
         }
 
-        DataFileMeta dataFileMeta = writeResult.get(0).assignFirstRowId(firstRowId);
-        dataFileMeta =
-                dataFileMeta.assignSequenceNumber(
-                        minSequenceId(compactBefore), maxSequenceId(compactBefore));
-        compactAfter.add(dataFileMeta);
-
         return commitMessage(compactBefore, compactAfter);
     }
 }

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionFileWriter.java

@@ -52,6 +52,11 @@ public long getPos() {
         return out.size();
     }
 
+    /** Path of the in-flight deletion file, used to delete the partial file on failure. */
+    Path path() {
+        return path;
+    }
+
     public void write(String key, DeletionVector deletionVector) throws IOException {
         int start = out.size();
         int length = deletionVector.serializeTo(out);

paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVectorIndexFileWriter.java

@@ -53,12 +53,21 @@ public DeletionVectorIndexFileWriter(
      */
     public IndexFileMeta writeSingleFile(Map<String, DeletionVector> input) throws IOException {
         DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory, fileIO);
+        boolean success = false;
         try {
             for (Map.Entry<String, DeletionVector> entry : input.entrySet()) {
                 writer.write(entry.getKey(), entry.getValue());
             }
-        } finally {
             writer.close();
+            success = true;
+        } finally {
+            if (!success) {
+                // Delete the partial index file (e.g. speculative task kill interrupted the
+                // write); the caller never sees its meta, so nobody else can reclaim it.
+                // Interrupt-tolerant because task kill leaves the thread interrupted.
+                closeQuietly(writer);
+                fileIO.deleteQuietlyIgnoringInterrupt(writer.path());
+            }
         }
         return writer.result();
     }
@@ -70,15 +79,25 @@ public List<IndexFileMeta> writeWithRolling(Map<String, DeletionVector> input)
         }
         List<IndexFileMeta> result = new ArrayList<>();
         Iterator<Map.Entry<String, DeletionVector>> iterator = input.entrySet().iterator();
-        while (iterator.hasNext()) {
-            result.add(tryWriter(iterator));
+        try {
+            while (iterator.hasNext()) {
+                result.add(tryWriter(iterator));
+            }
+        } catch (Throwable t) {
+            // A failed roll loses the returned metas; delete the already-written files so they
+            // do not stay as orphans.
+            for (IndexFileMeta meta : result) {
+                fileIO.deleteQuietlyIgnoringInterrupt(indexPathFactory.toPath(meta));
+            }
+            throw t;
         }
         return result;
     }
 
     private IndexFileMeta tryWriter(Iterator<Map.Entry<String, DeletionVector>> iterator)
             throws IOException {
         DeletionFileWriter writer = new DeletionFileWriter(indexPathFactory, fileIO);
+        boolean success = false;
         try {
             while (iterator.hasNext()) {
                 Map.Entry<String, DeletionVector> entry = iterator.next();
@@ -87,9 +106,22 @@ private IndexFileMeta tryWriter(Iterator<Map.Entry<String, DeletionVector>> iter
                     break;
                 }
             }
-        } finally {
             writer.close();
+            success = true;
+        } finally {
+            if (!success) {
+                closeQuietly(writer);
+                fileIO.deleteQuietlyIgnoringInterrupt(writer.path());
+            }
         }
         return writer.result();
     }
+
+    private static void closeQuietly(DeletionFileWriter writer) {
+        try {
+            writer.close();
+        } catch (Throwable ignored) {
+            // best-effort stream close on the failure path
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/io/FileWriterAbortExecutor.java

@@ -32,6 +32,8 @@ public FileWriterAbortExecutor(FileIO fileIO, Path path) {
     }
 
     public void abort() {
-        fileIO.deleteQuietly(path);
+        // Task kill leaves Thread interrupted; use interrupt-tolerant delete so abort
+        // during prepareCommit (speculative loser) can still remove written files.
+        fileIO.deleteQuietlyIgnoringInterrupt(path);
     }
 }

paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java

@@ -256,7 +256,8 @@ public void deleteFile(DataFileMeta file) {
         // in WriteFormatKey
         DataFilePathFactory pathFactory =
                 formatContext.pathFactory(new WriteFormatKey(file.level(), false));
-        file.collectFiles(pathFactory).forEach(fileIO::deleteQuietly);
+        // Interrupt-tolerant: used by MergeTreeWriter.close unpublished cleanup after task kill.
+        file.collectFiles(pathFactory).forEach(fileIO::deleteQuietlyIgnoringInterrupt);
     }
 
     public FileIO getFileIO() {