From 16bf27925ff298686971cb9563bf62eb6046e0ef Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Sat, 5 Sep 2026 10:30:20 +0800 Subject: [PATCH 1/2] [common] Version the memory-mode cache key in CachingFileIO The memory branch of newInputStream keyed the cache by path alone while the disk branch keys by path, length and modification time. consumer-* and service-* classify as META, which the default whitelist caches, and both are written in place by overwriteFileUtf8, so a consumer reset kept serving the pre-reset content. Build the same versioned key for the memory branch, at the cost of one getFileStatus per open there. --- .../apache/paimon/fs/cache/CachingFileIO.java | 12 ++++-- .../paimon/fs/cache/CachingFileIOTest.java | 43 ++++++++++++++++++- 2 files changed, 49 insertions(+), 6 deletions(-) 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); } }; } From 9c2cf5ec983d9483426e8c13856859cf7fd7a405 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 10 Sep 2026 14:20:47 +0800 Subject: [PATCH 2/2] fix: exclude in-place-overwritten metadata from the cache instead of versioning the memory key --- .../apache/paimon/fs/cache/CachingFileIO.java | 12 ++--- .../org/apache/paimon/utils/FileType.java | 14 +++++- .../paimon/fs/cache/CachingFileIOTest.java | 50 +++++++++++++------ .../org/apache/paimon/utils/FileTypeTest.java | 32 ++++++++++++ 4 files changed, 82 insertions(+), 26 deletions(-) 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 4023e3387460..65eeaa3ebfd7 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,16 +137,12 @@ 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, versionedCacheKey(path, status), status.getLen()); + delegate, path, c, diskCacheKey(path, status), status.getLen()); } - // 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()); + return new CachingSeekableInputStream(delegate, path, c, cacheNamespace + ":" + path, -1); } @Override @@ -285,7 +281,7 @@ private static void releaseCacheManager( }); } - private static String versionedCacheKey(Path path, FileStatus status) { + private static String diskCacheKey(Path path, FileStatus status) { return path + "\0" + status.getLen() + "\0" + status.getModificationTime(); } diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/FileType.java b/paimon-common/src/main/java/org/apache/paimon/utils/FileType.java index 46c767bb240c..9b30c864c1f5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/FileType.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/FileType.java @@ -114,8 +114,18 @@ public static Set parseWhitelist(String whitelist) { /** Returns {@code true} if the file is mutable and should not be cached. */ public static boolean isMutable(Path filePath) { - String name = filePath.getName(); - return "EARLIEST".equals(name) || "LATEST".equals(name); + String name = unwrapTempFileName(filePath.getName()); + // Files rewritten in place under a stable path: caching them by path keeps serving the + // pre-overwrite content (and a len+mtime key still collides when a rewrite lands at the + // same size within the same clock second). Hint files, consumer and service progress + // files, replaceable tags and _SUCCESS all go through overwriteFileUtf8. + return "EARLIEST".equals(name) + || "LATEST".equals(name) + || "_SUCCESS".equals(name) + || name.endsWith("_SUCCESS") + || name.startsWith(CONSUMER_PREFIX) + || name.startsWith(SERVICE_PREFIX) + || name.startsWith(TAG_PREFIX); } /** 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 802bf379b083..05ddd7cad08f 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 @@ -95,7 +95,7 @@ void setUp() { } @Test - void testMemoryModeServesFreshContentAfterInPlaceOverwrite() throws IOException { + void testMemoryModeDoesNotCacheInPlaceOverwrittenFiles() throws IOException { MockFileIO delegate = new MockFileIO(); CachingFileIO cachingIO = newCachingFileIO( @@ -105,27 +105,52 @@ void testMemoryModeServesFreshContentAfterInPlaceOverwrite() throws IOException 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); + // consumer-* is written in place by overwriteFileUtf8, so it is mutable and bypasses the + // cache: each read reaches the delegate and sees the current content. + delegate.addFile("consumer-1", "v1cc".getBytes()); try (SeekableInputStream in = cachingIO.newInputStream(consumer)) { + assertThat(in).isNotInstanceOf(CachingSeekableInputStream.class); 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); + delegate.addFile("consumer-1", "v2cc".getBytes()); 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 + // never cached: the overwrite is visible and the delegate was opened again assertThat(delegate.newInputStreamCallCount("consumer-1")).isEqualTo(2); } + @Test + void testMemoryModeImmutableCacheHitsDoNotRestatDelegate() throws IOException { + MockFileIO delegate = new MockFileIO(); + CachingFileIO cachingIO = + newCachingFileIO( + delegate, + new LocalMemoryCacheManager(Long.MAX_VALUE, 64), + EnumSet.of(FileType.META), + 64); + Path snapshot = new Path("snapshot-1"); + delegate.addFile("snapshot-1", "0123456789abcdef".getBytes()); + + for (int i = 0; i < 3; i++) { + try (SeekableInputStream in = cachingIO.newInputStream(snapshot)) { + assertThat(readAll(in, 16)).isEqualTo("0123456789abcdef".getBytes()); + } + } + + // An immutable file keeps the path-only memory key: opened once, then served from cache. + // Its size is resolved lazily and remembered, so repeated hits do not re-stat the + // delegate the way moving getFileStatus onto every open would. + assertThat(delegate.newInputStreamCallCount("snapshot-1")).isEqualTo(1); + assertThat(delegate.getFileStatusCallCount("snapshot-1")).isEqualTo(1); + } + @Test void testCreateBlobPresignedUrlDelegates() throws IOException { FileIO delegate = mock(FileIO.class); @@ -844,8 +869,7 @@ void testVectoredReadOpensSingleRemoteStream() throws Exception { CountDownLatch openGate = new CountDownLatch(1); delegate.blockOpensUntil(openGate); - // the file size is resolved lazily here, exercising the lazy path that only - // the testing constructor still uses + // the file size is resolved lazily here, as CachingFileIO does for the memory cache CachingSeekableInputStream stream = new CachingSeekableInputStream( delegate, @@ -1027,7 +1051,6 @@ 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<>(); @@ -1072,11 +1095,6 @@ 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); } @@ -1162,7 +1180,7 @@ public Path getPath() { @Override public long getModificationTime() { - return mtimes.getOrDefault(name, 0L); + return 0; } }; } diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/FileTypeTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/FileTypeTest.java index d946f71bb753..d043d6bf9753 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/FileTypeTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/FileTypeTest.java @@ -31,6 +31,38 @@ public class FileTypeTest { private static final String TABLE_ROOT = "hdfs://cluster/warehouse/db.db/table"; + // ===== mutable files (bypass the cache) ===== + + @Test + public void testIsMutable() { + // Files overwritten in place under a stable path must bypass the cache. + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/snapshot/EARLIEST"))).isTrue(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/snapshot/LATEST"))).isTrue(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/consumer/consumer-myGroup"))) + .isTrue(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/service/service-primary-key-lookup"))) + .isTrue(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/tag/tag-myTag"))).isTrue(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/dt=2024-01-01/bucket-0/_SUCCESS"))) + .isTrue(); + // A temp rewrite of a mutable file is still mutable. + assertThat( + FileType.isMutable( + new Path( + TABLE_ROOT + + "/consumer/.consumer-myGroup." + + UUID.randomUUID() + + ".tmp"))) + .isTrue(); + + // Write-once files stay cacheable: a new version lands under a new name. + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/snapshot/snapshot-1"))).isFalse(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/schema/schema-0"))).isFalse(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/manifest/manifest-a1b2c3d4-0"))) + .isFalse(); + assertThat(FileType.isMutable(new Path(TABLE_ROOT + "/bucket-0/data-abc.orc"))).isFalse(); + } + // ===== META files ===== @Test