diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java index 65eeaa3ebfd7..4023e3387460 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java @@ -137,12 +137,16 @@ public SeekableInputStream newInputStream(Path path) throws IOException { if (c == null) { return delegate.newInputStream(path); } + FileStatus status = delegate.getFileStatus(path); if (c instanceof LocalDiskCacheManager) { - FileStatus status = delegate.getFileStatus(path); return new CachingSeekableInputStream( - delegate, path, c, diskCacheKey(path, status), status.getLen()); + delegate, path, c, versionedCacheKey(path, status), status.getLen()); } - return new CachingSeekableInputStream(delegate, path, c, cacheNamespace + ":" + path, -1); + // Version the key by len+mtime as the disk branch does, otherwise an in-place + // overwrite of a whitelisted path keeps serving the cached content. The namespace + // prefix stays, since SharedCacheManager invalidation matches on it. + String cacheKey = cacheNamespace + ":" + versionedCacheKey(path, status); + return new CachingSeekableInputStream(delegate, path, c, cacheKey, status.getLen()); } @Override @@ -281,7 +285,7 @@ private static void releaseCacheManager( }); } - private static String diskCacheKey(Path path, FileStatus status) { + private static String versionedCacheKey(Path path, FileStatus status) { return path + "\0" + status.getLen() + "\0" + status.getModificationTime(); } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java index ded23667a972..802bf379b083 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/cache/CachingFileIOTest.java @@ -94,6 +94,38 @@ void setUp() { MockFileIO.resetGlobalInputStreamCalls(); } + @Test + void testMemoryModeServesFreshContentAfterInPlaceOverwrite() throws IOException { + MockFileIO delegate = new MockFileIO(); + CachingFileIO cachingIO = + newCachingFileIO( + delegate, + new LocalMemoryCacheManager(Long.MAX_VALUE, 64), + EnumSet.of(FileType.META), + 64); + Path consumer = new Path("consumer-1"); + + // Same path overwritten in place with new content and a new mtime; the + // memory cache must not keep serving the first version's blocks. + delegate.addFile("consumer-1", "v1cc".getBytes(), 1000L); + try (SeekableInputStream in = cachingIO.newInputStream(consumer)) { + byte[] buf = new byte[4]; + in.read(buf, 0, 4); + assertThat(new String(buf)).isEqualTo("v1cc"); + } + // one remote open, after which the first version's blocks are cached + assertThat(delegate.newInputStreamCallCount("consumer-1")).isEqualTo(1); + + delegate.addFile("consumer-1", "v2cc".getBytes(), 2000L); + try (SeekableInputStream in = cachingIO.newInputStream(consumer)) { + byte[] buf = new byte[4]; + in.read(buf, 0, 4); + assertThat(new String(buf)).isEqualTo("v2cc"); + } + // the new version has a different key, forcing a fresh remote read + assertThat(delegate.newInputStreamCallCount("consumer-1")).isEqualTo(2); + } + @Test void testCreateBlobPresignedUrlDelegates() throws IOException { FileIO delegate = mock(FileIO.class); @@ -812,7 +844,8 @@ void testVectoredReadOpensSingleRemoteStream() throws Exception { CountDownLatch openGate = new CountDownLatch(1); delegate.blockOpensUntil(openGate); - // the file size is resolved lazily here, as CachingFileIO does for the memory cache + // the file size is resolved lazily here, exercising the lazy path that only + // the testing constructor still uses CachingSeekableInputStream stream = new CachingSeekableInputStream( delegate, @@ -994,6 +1027,7 @@ private static class MockFileIO implements FileIO { private final Map files = new HashMap<>(); private final Map reportedLengths = new HashMap<>(); + private final Map mtimes = new HashMap<>(); // concurrent so the thread-safety tests below can count from several reader threads private final Map fileStatusCalls = new ConcurrentHashMap<>(); private final Map newInputStreamCalls = new ConcurrentHashMap<>(); @@ -1038,6 +1072,11 @@ static int globalInputStreamCallCount(String name) { return count == null ? 0 : count.get(); } + void addFile(String name, byte[] data, long mtime) { + files.put(name, data); + mtimes.put(name, mtime); + } + void addFile(String name, byte[] data) { files.put(name, data); } @@ -1123,7 +1162,7 @@ public Path getPath() { @Override public long getModificationTime() { - return 0; + return mtimes.getOrDefault(name, 0L); } }; }