From f01ea7c075dcf67e72028da3e5a96cdb404a200c Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 15 Sep 2026 19:25:48 +0800 Subject: [PATCH 1/4] [core] Report a snapshot deleted during reading as not found Some FileIOs report a snapshot that is deleted after its input stream was opened as a generic IOException on the first read, for example "404 Not Found / NoSuchKey" from an object store. SnapshotManager#tryFromPath turned that into a RuntimeException, so callers that rely on FileNotFoundException to skip expired snapshots failed instead. A streaming writer walking snapshots in latestSnapshotOfUser for its conflict-aware writer clean checker could fail its checkpoint when a concurrent expiration deleted the snapshot it was reading. When reading fails with an IOException other than FileNotFoundException, check whether the snapshot file still exists and report it as not found if it is gone. The original failure is kept when the file exists or its existence cannot be checked. --- .../apache/paimon/utils/SnapshotManager.java | 16 ++++++ .../paimon/utils/SnapshotManagerTest.java | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java index 18740a9a3fed..434474237bd2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java @@ -870,6 +870,13 @@ public static Snapshot tryFromPath(FileIO fileIO, Path path) throws FileNotFound } catch (FileNotFoundException e) { throw e; } catch (IOException e) { + // Some FileIOs throw a plain IOException for a snapshot deleted during reading. + if (!fileExists(fileIO, path)) { + FileNotFoundException notFound = + new FileNotFoundException("Snapshot file " + path + " does not exist."); + notFound.initCause(e); + throw notFound; + } throw new RuntimeException("Fails to read snapshot from path " + path, e); } @@ -888,4 +895,13 @@ public static Snapshot tryFromPath(FileIO fileIO, Path path) throws FileNotFound } throw new RuntimeException("Retry fail after 10 times", exception); } + + private static boolean fileExists(FileIO fileIO, Path path) { + try { + return fileIO.exists(path); + } catch (IOException ignored) { + // Treat as present so that the original read failure is kept. + return true; + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java index c6dddc6dcab9..6c43a931e559 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java @@ -140,6 +140,61 @@ public void testSnapshotExistsRestoresInterruptedStatus() throws IOException { } } + @Test + public void testTryFromPathReportsSnapshotDeletedDuringReadAsNotFound() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString())); + Path path = snapshotManager.snapshotPath(1); + fileIO.tryToWriteAtomic(path, createSnapshotWithMillis(1, 1000).toJson()); + IOException readFailure = new IOException("404 Not Found"); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(path); + throw readFailure; + }) + .when(fileIO) + .readFileUtf8(path); + + assertThatThrownBy(() -> SnapshotManager.tryFromPath(fileIO, path)) + .isInstanceOf(FileNotFoundException.class) + .hasCause(readFailure); + } + + @Test + public void testTryFromPathKeepsReadFailureOfExistingSnapshot() throws IOException { + FileIO fileIO = Mockito.spy(LocalFileIO.create()); + SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString())); + Path path = snapshotManager.snapshotPath(1); + fileIO.tryToWriteAtomic(path, createSnapshotWithMillis(1, 1000).toJson()); + Mockito.doThrow(new IOException("Read failure")).when(fileIO).readFileUtf8(path); + + assertThatThrownBy(() -> SnapshotManager.tryFromPath(fileIO, path)) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Fails to read snapshot from path") + .hasRootCauseMessage("Read failure"); + } + + @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) + .readFileUtf8(expiring); + + assertThat(snapshotManager.latestSnapshotOfUser("currentCommitUser")).isEmpty(); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void testEarliestSnapshot(boolean isRaceCondition) throws IOException { From cae659f49f358f7dbb08d204fde8fa61ceb1359d Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 15 Sep 2026 23:56:19 +0800 Subject: [PATCH 2/4] [common] Report a file deleted during reading as not found in FileIO Move the existence check from SnapshotManager#tryFromPath into FileIO#readFileUtf8, so tag, changelog and schema reads also treat a file deleted during reading as not found. --- .../java/org/apache/paimon/fs/FileIO.java | 8 +++++ .../java/org/apache/paimon/fs/FileIOTest.java | 29 +++++++++++++++ .../apache/paimon/utils/SnapshotManager.java | 16 --------- .../paimon/utils/SnapshotManagerTest.java | 36 +------------------ 4 files changed, 38 insertions(+), 51 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index c20e2c70bb18..969702c9aa75 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java @@ -341,6 +341,14 @@ default String readFileUtf8(Path path) throws IOException { builder.append(line); } return builder.toString(); + } catch (FileNotFoundException e) { + throw e; + } catch (IOException e) { + // Some object stores throw a plain IOException for a file deleted during reading. + if (exists(path)) { + throw e; + } + throw new FileNotFoundException("File " + path + " does not exist."); } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java index 96e023d1eb46..e3e831753505 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java @@ -25,6 +25,7 @@ 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; @@ -51,6 +52,34 @@ public class FileIOTest { @TempDir java.nio.file.Path tempDir; + @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); + Mockito.doAnswer( + invocation -> { + fileIO.deleteQuietly(file); + throw new IOException("404 Not Found"); + }) + .when(fileIO) + .newInputStream(file); + + assertThatThrownBy(() -> fileIO.readFileUtf8(file)) + .isInstanceOf(FileNotFoundException.class); + } + + @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 testRequireOptions() throws IOException { Options options = new Options(); diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java index 434474237bd2..18740a9a3fed 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java @@ -870,13 +870,6 @@ public static Snapshot tryFromPath(FileIO fileIO, Path path) throws FileNotFound } catch (FileNotFoundException e) { throw e; } catch (IOException e) { - // Some FileIOs throw a plain IOException for a snapshot deleted during reading. - if (!fileExists(fileIO, path)) { - FileNotFoundException notFound = - new FileNotFoundException("Snapshot file " + path + " does not exist."); - notFound.initCause(e); - throw notFound; - } throw new RuntimeException("Fails to read snapshot from path " + path, e); } @@ -895,13 +888,4 @@ public static Snapshot tryFromPath(FileIO fileIO, Path path) throws FileNotFound } throw new RuntimeException("Retry fail after 10 times", exception); } - - private static boolean fileExists(FileIO fileIO, Path path) { - try { - return fileIO.exists(path); - } catch (IOException ignored) { - // Treat as present so that the original read failure is kept. - return true; - } - } } diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java index 6c43a931e559..fb09951952fd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java @@ -140,40 +140,6 @@ public void testSnapshotExistsRestoresInterruptedStatus() throws IOException { } } - @Test - public void testTryFromPathReportsSnapshotDeletedDuringReadAsNotFound() throws IOException { - FileIO fileIO = Mockito.spy(LocalFileIO.create()); - SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString())); - Path path = snapshotManager.snapshotPath(1); - fileIO.tryToWriteAtomic(path, createSnapshotWithMillis(1, 1000).toJson()); - IOException readFailure = new IOException("404 Not Found"); - Mockito.doAnswer( - invocation -> { - fileIO.deleteQuietly(path); - throw readFailure; - }) - .when(fileIO) - .readFileUtf8(path); - - assertThatThrownBy(() -> SnapshotManager.tryFromPath(fileIO, path)) - .isInstanceOf(FileNotFoundException.class) - .hasCause(readFailure); - } - - @Test - public void testTryFromPathKeepsReadFailureOfExistingSnapshot() throws IOException { - FileIO fileIO = Mockito.spy(LocalFileIO.create()); - SnapshotManager snapshotManager = newSnapshotManager(fileIO, new Path(tempDir.toString())); - Path path = snapshotManager.snapshotPath(1); - fileIO.tryToWriteAtomic(path, createSnapshotWithMillis(1, 1000).toJson()); - Mockito.doThrow(new IOException("Read failure")).when(fileIO).readFileUtf8(path); - - assertThatThrownBy(() -> SnapshotManager.tryFromPath(fileIO, path)) - .isInstanceOf(RuntimeException.class) - .hasMessageContaining("Fails to read snapshot from path") - .hasRootCauseMessage("Read failure"); - } - @Test public void testLatestSnapshotOfUserStopsAtSnapshotDeletedDuringRead() throws IOException { FileIO fileIO = Mockito.spy(LocalFileIO.create()); @@ -190,7 +156,7 @@ public void testLatestSnapshotOfUserStopsAtSnapshotDeletedDuringRead() throws IO throw new IOException("404 Not Found"); }) .when(fileIO) - .readFileUtf8(expiring); + .newInputStream(expiring); assertThat(snapshotManager.latestSnapshotOfUser("currentCommitUser")).isEmpty(); } From 6a55502466c4e0c096ad4555e0f1dedebcbe8912 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 16 Sep 2026 07:42:24 +0800 Subject: [PATCH 3/4] [common] Keep the original read failure when a file cannot be confirmed to be gone Report a file as not found only when the existence check says it is gone, and attach the read failure as the cause. When the check itself fails, keep the original failure instead, because callers skip a not-found file silently. --- .../java/org/apache/paimon/fs/FileIO.java | 17 ++++++++-- .../java/org/apache/paimon/fs/FileIOTest.java | 31 +++++++++++++++++-- .../apache/paimon/utils/TagManagerTest.java | 22 +++++++++++++ 3 files changed, 66 insertions(+), 4 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index 969702c9aa75..e0741dcca1f7 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java @@ -345,10 +345,23 @@ default String readFileUtf8(Path path) throws IOException { throw e; } catch (IOException e) { // Some object stores throw a plain IOException for a file deleted during reading. - if (exists(path)) { + boolean missing; + try { + missing = !exists(path); + } catch (IOException | RuntimeException checkFailure) { + // Keep the original failure when the file cannot be confirmed to be gone. + e.addSuppressed(checkFailure); + throw e; + } + if (!missing) { throw e; } - throw new FileNotFoundException("File " + path + " does not exist."); + 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; } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java index e3e831753505..f43fb3192941 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java @@ -57,16 +57,18 @@ public void testReadFileUtf8ReportsFileDeletedDuringReadAsNotFound() throws IOEx 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 new IOException("404 Not Found"); + throw readFailure; }) .when(fileIO) .newInputStream(file); assertThatThrownBy(() -> fileIO.readFileUtf8(file)) - .isInstanceOf(FileNotFoundException.class); + .isInstanceOf(FileNotFoundException.class) + .hasCause(readFailure); } @Test @@ -80,6 +82,31 @@ public void testReadFileUtf8KeepsReadFailureOfExistingFile() throws IOException 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(); diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java index ede52e6ef3ab..4a899950ea3f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/TagManagerTest.java +++ b/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); From 6c6aaf6ce2bedab9954dfaf85ce5ab8a7ce82687 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Thu, 17 Sep 2026 14:04:24 +0800 Subject: [PATCH 4/4] [common] Do not report an interrupted read as a missing file Propagate InterruptedIOException and ClosedByInterruptException before the existence check, so a cancelled read is not turned into FileNotFoundException and silently swallowed by callers. Skip addSuppressed when the existence check rethrows the read failure itself. --- .../java/org/apache/paimon/fs/FileIO.java | 9 +++- .../java/org/apache/paimon/fs/FileIOTest.java | 48 +++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index e0741dcca1f7..c6e8d02a46ce 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/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; @@ -343,6 +345,9 @@ default String readFileUtf8(Path path) throws IOException { 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; @@ -350,7 +355,9 @@ default String readFileUtf8(Path path) throws IOException { missing = !exists(path); } catch (IOException | RuntimeException checkFailure) { // Keep the original failure when the file cannot be confirmed to be gone. - e.addSuppressed(checkFailure); + if (checkFailure != e) { + e.addSuppressed(checkFailure); + } throw e; } if (!missing) { diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java index f43fb3192941..6642f7b62a92 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java @@ -30,6 +30,8 @@ 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; @@ -52,6 +54,52 @@ 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());