[common] Fix snapshot and tag reads failing when the file is deleted concurrently

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

@@ -35,9 +35,11 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.InterruptedIOException;
 import java.io.OutputStreamWriter;
 import java.io.Serializable;
 import java.net.URI;
+import java.nio.channels.ClosedByInterruptException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -341,6 +343,32 @@ default String readFileUtf8(Path path) throws IOException {
                 builder.append(line);
             }
             return builder.toString();
+        } catch (FileNotFoundException e) {
+            throw e;
+        } catch (InterruptedIOException | ClosedByInterruptException e) {
+            // An interrupted read says nothing about whether the file is still there.
+            throw e;
+        } catch (IOException e) {
+            // Some object stores throw a plain IOException for a file deleted during reading.
+            boolean missing;
+            try {
+                missing = !exists(path);
+            } catch (IOException | RuntimeException checkFailure) {
+                // Keep the original failure when the file cannot be confirmed to be gone.
+                if (checkFailure != e) {
+                    e.addSuppressed(checkFailure);
+                }
+                throw e;
+            }
+            if (!missing) {
+                throw e;
+            }
+            LOG.debug(
+                    "Read of {} failed and the file is gone, reporting it as not found.", path, e);
+            FileNotFoundException notFound =
+                    new FileNotFoundException("File " + path + " does not exist.");
+            notFound.initCause(e);
+            throw notFound;
         }
     }
 

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

@@ -25,10 +25,13 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
 
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.channels.ClosedByInterruptException;
 import java.nio.file.AccessDeniedException;
 import java.nio.file.DirectoryNotEmptyException;
 import java.nio.file.FileAlreadyExistsException;
@@ -51,6 +54,107 @@ public class FileIOTest {
 
     @TempDir java.nio.file.Path tempDir;
 
+    @Test
+    public void testReadFileUtf8PropagatesInterruptedRead() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        Path file = new Path(tempDir.resolve("file").toUri());
+        fileIO.writeFile(file, "content", false);
+        InterruptedIOException interrupted = new InterruptedIOException("Interrupted");
+        Mockito.doAnswer(
+                        invocation -> {
+                            fileIO.deleteQuietly(file);
+                            throw interrupted;
+                        })
+                .when(fileIO)
+                .newInputStream(file);
+        Mockito.clearInvocations(fileIO);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file)).isSameAs(interrupted);
+        Mockito.verify(fileIO, Mockito.never()).exists(file);
+
+        Path closed = new Path(tempDir.resolve("closed").toUri());
+        fileIO.writeFile(closed, "content", false);
+        ClosedByInterruptException closedByInterrupt = new ClosedByInterruptException();
+        Mockito.doAnswer(
+                        invocation -> {
+                            fileIO.deleteQuietly(closed);
+                            throw closedByInterrupt;
+                        })
+                .when(fileIO)
+                .newInputStream(closed);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(closed)).isSameAs(closedByInterrupt);
+    }
+
+    @Test
+    public void testReadFileUtf8KeepsReadFailureReusedByExistenceCheck() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        Path file = new Path(tempDir.resolve("file").toUri());
+        fileIO.writeFile(file, "content", false);
+        IOException failure = new IOException("Cached failure");
+        Mockito.doThrow(failure).when(fileIO).newInputStream(file);
+        Mockito.doThrow(failure).when(fileIO).exists(file);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file))
+                .isSameAs(failure)
+                .satisfies(e -> assertThat(e.getSuppressed()).isEmpty());
+    }
+
+    @Test
+    public void testReadFileUtf8ReportsFileDeletedDuringReadAsNotFound() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        Path file = new Path(tempDir.resolve("file").toUri());
+        fileIO.writeFile(file, "content", false);
+        IOException readFailure = new IOException("404 Not Found");
+        Mockito.doAnswer(
+                        invocation -> {
+                            fileIO.deleteQuietly(file);
+                            throw readFailure;
+                        })
+                .when(fileIO)
+                .newInputStream(file);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file))
+                .isInstanceOf(FileNotFoundException.class)
+                .hasCause(readFailure);
+    }
+
+    @Test
+    public void testReadFileUtf8KeepsReadFailureOfExistingFile() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        Path file = new Path(tempDir.resolve("file").toUri());
+        fileIO.writeFile(file, "content", false);
+        IOException readFailure = new IOException("Read failure");
+        Mockito.doThrow(readFailure).when(fileIO).newInputStream(file);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file)).isSameAs(readFailure);
+    }
+
+    @Test
+    public void testReadFileUtf8KeepsReadFailureWhenExistenceCheckFails() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        Path file = new Path(tempDir.resolve("file").toUri());
+        fileIO.writeFile(file, "content", false);
+        IOException readFailure = new IOException("Read failure");
+        Mockito.doThrow(readFailure).when(fileIO).newInputStream(file);
+        IOException checkFailure = new IOException("Exists failure");
+        Mockito.doThrow(checkFailure).when(fileIO).exists(file);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file))
+                .isSameAs(readFailure)
+                .satisfies(e -> assertThat(e.getSuppressed()).containsExactly(checkFailure));
+
+        IOException secondReadFailure = new IOException("Read failure");
+        Mockito.doThrow(secondReadFailure).when(fileIO).newInputStream(file);
+        RuntimeException uncheckedCheckFailure = new RuntimeException("Exists failure");
+        Mockito.doThrow(uncheckedCheckFailure).when(fileIO).exists(file);
+
+        assertThatThrownBy(() -> fileIO.readFileUtf8(file))
+                .isSameAs(secondReadFailure)
+                .satisfies(
+                        e -> assertThat(e.getSuppressed()).containsExactly(uncheckedCheckFailure));
+    }
+
     @Test
     public void testRequireOptions() throws IOException {
         Options options = new Options();

paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java

@@ -140,6 +140,27 @@ public void testSnapshotExistsRestoresInterruptedStatus() throws IOException {
         }
     }
 
+    @Test
+    public void testLatestSnapshotOfUserStopsAtSnapshotDeletedDuringRead() throws IOException {
+        FileIO fileIO = Mockito.spy(LocalFileIO.create());
+        SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString()));
+        for (long id = 1; id <= 3; id++) {
+            fileIO.tryToWriteAtomic(
+                    snapshotManager.snapshotPath(id),
+                    createSnapshotWithMillis(id, id * 1000).toJson());
+        }
+        Path expiring = snapshotManager.snapshotPath(1);
+        Mockito.doAnswer(
+                        invocation -> {
+                            fileIO.deleteQuietly(expiring);
+                            throw new IOException("404 Not Found");
+                        })
+                .when(fileIO)
+                .newInputStream(expiring);
+
+        assertThat(snapshotManager.latestSnapshotOfUser("currentCommitUser")).isEmpty();
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testEarliestSnapshot(boolean isRaceCondition) throws IOException {

paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java

@@ -40,7 +40,9 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
 
+import java.io.IOException;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
@@ -51,6 +53,7 @@
 
 import static org.apache.paimon.operation.FileStoreTestUtils.commitData;
 import static org.apache.paimon.operation.FileStoreTestUtils.partitionedData;
+import static org.apache.paimon.utils.SnapshotManagerTest.createSnapshotWithMillis;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for TagManager. */
@@ -71,6 +74,25 @@ public void setup() throws Exception {
         tagManager = null;
     }
 
+    @Test
+    public void testGetTagDeletedDuringRead() throws IOException {
+        FileIO spyFileIO = Mockito.spy(LocalFileIO.create());
+        tagManager = new TagManager(spyFileIO, new Path(root));
+        Path path = tagManager.tagPath("tag");
+        spyFileIO.tryToWriteAtomic(
+                path,
+                Tag.fromSnapshotAndTagTtl(createSnapshotWithMillis(1, 1000), null, null).toJson());
+        Mockito.doAnswer(
+                        invocation -> {
+                            spyFileIO.deleteQuietly(path);
+                            throw new IOException("404 Not Found");
+                        })
+                .when(spyFileIO)
+                .newInputStream(path);
+
+        assertThat(tagManager.get("tag")).isEmpty();
+    }
+
     @Test
     public void testCreateTagWithoutTimeRetained() throws Exception {
         TestFileStore store = createStore(TestKeyValueGenerator.GeneratorMode.NON_PARTITIONED, 4);