[common] Version the memory-mode cache key in CachingFileIO

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();
     }
 

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<String, byte[]> files = new HashMap<>();
         private final Map<String, Long> reportedLengths = new HashMap<>();
+        private final Map<String, Long> mtimes = new HashMap<>();
         // concurrent so the thread-safety tests below can count from several reader threads
         private final Map<String, Integer> fileStatusCalls = new ConcurrentHashMap<>();
         private final Map<String, Integer> 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);
                 }
             };
         }