This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a1fcfe1095 [common] Restore local cache after FileIO deserialization
(#8926)
a1fcfe1095 is described below
commit a1fcfe10959cfbf6c54b8493365bd36746f3edc7
Author: shyjsarah <[email protected]>
AuthorDate: Thu Jul 30 19:17:57 2026 +0800
[common] Restore local cache after FileIO deserialization (#8926)
---
.../org/apache/paimon/fs/cache/CachingFileIO.java | 249 ++++++++++++++++++--
.../fs/cache/CachingSeekableInputStream.java | 19 +-
.../apache/paimon/fs/cache/LocalCacheManager.java | 3 +
.../paimon/fs/cache/LocalDiskCacheManager.java | 30 ++-
.../paimon/fs/cache/LocalMemoryCacheManager.java | 15 ++
.../apache/paimon/fs/cache/CachingFileIOTest.java | 256 ++++++++++++++++++++-
.../paimon/fs/cache/LocalDiskCacheManagerTest.java | 23 ++
7 files changed, 559 insertions(+), 36 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 14b5c39abf..28d5276db4 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
@@ -33,31 +33,57 @@ import org.apache.paimon.utils.FileType;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
import java.time.Duration;
import java.util.EnumSet;
+import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
/**
* A {@link FileIO} wrapper that caches reads at block granularity.
*
* <p>Only file types in the whitelist are cached. Others are read directly
from the delegate.
*
- * <p>After deserialization, the cache is null and reads fall through to the
delegate directly.
+ * <p>After deserialization, copies of the same catalog-created wrapper lazily
reuse a JVM-local
+ * cache manager.
*/
public class CachingFileIO implements FileIO {
private static final long serialVersionUID = 1L;
+ // Copies of one serialized wrapper share a JVM-local limit instead of
allocating one per task.
+ private static final Map<LocalCacheConfiguration, SharedCacheManager>
+ DESERIALIZED_CACHE_MANAGERS = new ConcurrentHashMap<>();
+
private final FileIO delegate;
private final Set<FileType> whitelist;
+ @Nullable private final LocalCacheConfiguration cacheConfiguration;
+ private String cacheNamespace;
private transient volatile LocalCacheManager cache;
+ private transient volatile SharedCacheManager sharedCacheManager;
+ private transient volatile boolean closed;
public CachingFileIO(FileIO delegate, LocalCacheManager cache,
Set<FileType> whitelist) {
+ this(delegate, cache, whitelist, null, UUID.randomUUID().toString());
+ }
+
+ private CachingFileIO(
+ FileIO delegate,
+ LocalCacheManager cache,
+ Set<FileType> whitelist,
+ @Nullable LocalCacheConfiguration cacheConfiguration,
+ String cacheNamespace) {
this.delegate = delegate;
this.cache = cache;
this.whitelist = EnumSet.copyOf(whitelist);
+ this.cacheConfiguration = cacheConfiguration;
+ this.cacheNamespace = cacheNamespace;
}
/**
@@ -82,7 +108,13 @@ public class CachingFileIO implements FileIO {
if (whitelist.isEmpty()) {
return fileIO;
}
- return new CachingFileIO(fileIO, cache, whitelist);
+ String cacheNamespace = UUID.randomUUID().toString();
+ return new CachingFileIO(
+ fileIO,
+ cache,
+ whitelist,
+ LocalCacheConfiguration.from(context, cacheNamespace),
+ cacheNamespace);
}
/**
@@ -91,31 +123,26 @@ public class CachingFileIO implements FileIO {
*/
@Nullable
public static LocalCacheManager createCacheManager(CatalogContext context)
{
- Options options = context.options();
- if (!options.get(CatalogOptions.LOCAL_CACHE_ENABLED)) {
- return null;
- }
-
- MemorySize maxSizeOpt =
options.get(CatalogOptions.LOCAL_CACHE_MAX_SIZE);
- long maxSize = maxSizeOpt == null ? Long.MAX_VALUE :
maxSizeOpt.getBytes();
- int blockSize = (int)
options.get(CatalogOptions.LOCAL_CACHE_BLOCK_SIZE).getBytes();
-
- String cacheDir = options.get(CatalogOptions.LOCAL_CACHE_DIR);
- if (cacheDir != null) {
- return new LocalDiskCacheManager(cacheDir, maxSize, blockSize);
- } else {
- return new LocalMemoryCacheManager(maxSize, blockSize);
- }
+ LocalCacheConfiguration configuration =
LocalCacheConfiguration.from(context, "");
+ return configuration == null ? null :
configuration.createCacheManager();
}
@Override
public SeekableInputStream newInputStream(Path path) throws IOException {
- LocalCacheManager c = cache;
FileType fileType = FileType.classify(path);
- if (c == null || !whitelist.contains(fileType) ||
FileType.isMutable(path)) {
+ if (!whitelist.contains(fileType) || FileType.isMutable(path)) {
+ return delegate.newInputStream(path);
+ }
+ LocalCacheManager c = getOrCreateCacheManager();
+ if (c == null) {
return delegate.newInputStream(path);
}
- return new CachingSeekableInputStream(delegate, path, c);
+ if (c instanceof LocalDiskCacheManager) {
+ FileStatus status = delegate.getFileStatus(path);
+ return new CachingSeekableInputStream(
+ delegate, path, c, diskCacheKey(path, status),
status.getLen());
+ }
+ return new CachingSeekableInputStream(delegate, path, c,
cacheNamespace + ":" + path, -1);
}
@Override
@@ -176,6 +203,186 @@ public class CachingFileIO implements FileIO {
@Override
public void close() throws IOException {
- delegate.close();
+ SharedCacheManager shared;
+ synchronized (this) {
+ closed = true;
+ shared = sharedCacheManager;
+ sharedCacheManager = null;
+ if (shared != null) {
+ cache = null;
+ }
+ }
+
+ try {
+ delegate.close();
+ } finally {
+ if (shared != null) {
+ releaseCacheManager(cacheConfiguration, shared);
+ }
+ }
+ }
+
+ @Nullable
+ private LocalCacheManager getOrCreateCacheManager() {
+ if (closed) {
+ return null;
+ }
+ LocalCacheManager current = cache;
+ if (current == null && cacheConfiguration != null) {
+ synchronized (this) {
+ if (closed) {
+ return null;
+ }
+ current = cache;
+ if (current == null) {
+ SharedCacheManager shared = sharedCacheManager;
+ if (shared == null) {
+ shared = acquireCacheManager(cacheConfiguration);
+ sharedCacheManager = shared;
+ }
+ current = shared.getOrCreate(cacheConfiguration);
+ cache = current;
+ }
+ }
+ }
+ return current;
+ }
+
+ private static SharedCacheManager acquireCacheManager(
+ LocalCacheConfiguration cacheConfiguration) {
+ return DESERIALIZED_CACHE_MANAGERS.compute(
+ cacheConfiguration,
+ (ignored, existing) -> {
+ SharedCacheManager shared =
+ existing == null ? new SharedCacheManager() :
existing;
+ shared.retain(cacheConfiguration.cacheNamespace);
+ return shared;
+ });
+ }
+
+ private static void releaseCacheManager(
+ @Nullable LocalCacheConfiguration cacheConfiguration,
SharedCacheManager shared) {
+ if (cacheConfiguration == null) {
+ return;
+ }
+ DESERIALIZED_CACHE_MANAGERS.computeIfPresent(
+ cacheConfiguration,
+ (ignored, existing) -> {
+ if (existing != shared) {
+ return existing;
+ }
+ return existing.release(cacheConfiguration.cacheNamespace)
? null : existing;
+ });
+ }
+
+ private static String diskCacheKey(Path path, FileStatus status) {
+ return path + "\0" + status.getLen() + "\0" +
status.getModificationTime();
+ }
+
+ private void readObject(ObjectInputStream input) throws IOException,
ClassNotFoundException {
+ input.defaultReadObject();
+ if (cacheConfiguration != null) {
+ sharedCacheManager = acquireCacheManager(cacheConfiguration);
+ }
+ }
+
+ private static class SharedCacheManager {
+
+ @Nullable private LocalCacheManager cacheManager;
+ private final Map<String, Integer> namespaceReferences = new
HashMap<>();
+ private int references;
+
+ private synchronized void retain(String cacheNamespace) {
+ references++;
+ namespaceReferences.merge(cacheNamespace, 1, Integer::sum);
+ }
+
+ private synchronized LocalCacheManager getOrCreate(
+ LocalCacheConfiguration cacheConfiguration) {
+ if (cacheManager == null) {
+ cacheManager = cacheConfiguration.createCacheManager();
+ }
+ return cacheManager;
+ }
+
+ private synchronized boolean release(String cacheNamespace) {
+ if (references <= 0) {
+ throw new IllegalStateException("Cache manager has already
been released.");
+ }
+ Integer namespaceCount = namespaceReferences.get(cacheNamespace);
+ if (namespaceCount == null) {
+ throw new IllegalStateException("Cache namespace has already
been released.");
+ }
+ if (namespaceCount == 1) {
+ namespaceReferences.remove(cacheNamespace);
+ if (cacheManager != null) {
+ cacheManager.invalidate(cacheNamespace + ":");
+ }
+ } else {
+ namespaceReferences.put(cacheNamespace, namespaceCount - 1);
+ }
+ return --references == 0;
+ }
+ }
+
+ private static class LocalCacheConfiguration implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Nullable private final String cacheDir;
+ private final long maxSize;
+ private final int blockSize;
+ private final String cacheNamespace;
+
+ private LocalCacheConfiguration(
+ @Nullable String cacheDir, long maxSize, int blockSize, String
cacheNamespace) {
+ this.cacheDir = cacheDir;
+ this.maxSize = maxSize;
+ this.blockSize = blockSize;
+ this.cacheNamespace = cacheNamespace;
+ }
+
+ @Nullable
+ private static LocalCacheConfiguration from(CatalogContext context,
String cacheNamespace) {
+ Options options = context.options();
+ if (!options.get(CatalogOptions.LOCAL_CACHE_ENABLED)) {
+ return null;
+ }
+
+ MemorySize maxSizeOption =
options.get(CatalogOptions.LOCAL_CACHE_MAX_SIZE);
+ long maxSize = maxSizeOption == null ? Long.MAX_VALUE :
maxSizeOption.getBytes();
+ int blockSize = (int)
options.get(CatalogOptions.LOCAL_CACHE_BLOCK_SIZE).getBytes();
+ return new LocalCacheConfiguration(
+ options.get(CatalogOptions.LOCAL_CACHE_DIR),
+ maxSize,
+ blockSize,
+ cacheNamespace);
+ }
+
+ private LocalCacheManager createCacheManager() {
+ if (cacheDir == null) {
+ return new LocalMemoryCacheManager(maxSize, blockSize);
+ }
+ return new LocalDiskCacheManager(cacheDir, maxSize, blockSize);
+ }
+
+ @Override
+ public boolean equals(Object object) {
+ if (this == object) {
+ return true;
+ }
+ if (!(object instanceof LocalCacheConfiguration)) {
+ return false;
+ }
+ LocalCacheConfiguration that = (LocalCacheConfiguration) object;
+ return maxSize == that.maxSize
+ && blockSize == that.blockSize
+ && Objects.equals(cacheDir, that.cacheDir);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(cacheDir, maxSize, blockSize);
+ }
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
index 024bf1e409..e3f990beb6 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingSeekableInputStream.java
@@ -33,26 +33,33 @@ public class CachingSeekableInputStream extends
SeekableInputStream implements V
private final FileIO fileIO;
private final Path path;
private final LocalCacheManager cache;
+ private final String cacheKey;
private long pos;
- private long fileSize = -1;
+ private long fileSize;
@Nullable private SeekableInputStream remoteStream;
public CachingSeekableInputStream(FileIO fileIO, Path path,
LocalCacheManager cache) {
+ this(fileIO, path, cache, path.toString(), -1);
+ }
+
+ CachingSeekableInputStream(
+ FileIO fileIO, Path path, LocalCacheManager cache, String
cacheKey, long fileSize) {
this.fileIO = fileIO;
this.path = path;
this.cache = cache;
+ this.cacheKey = cacheKey;
+ this.fileSize = fileSize;
this.pos = 0;
}
private long fileSize() throws IOException {
if (fileSize == -1) {
- String pathStr = path.toString();
- long cached = cache.getFileSize(pathStr);
+ long cached = cache.getFileSize(cacheKey);
if (cached >= 0) {
fileSize = cached;
} else {
fileSize = fileIO.getFileStatus(path).getLen();
- cache.putFileSize(pathStr, fileSize);
+ cache.putFileSize(cacheKey, fileSize);
}
}
return fileSize;
@@ -142,7 +149,7 @@ public class CachingSeekableInputStream extends
SeekableInputStream implements V
}
private byte[] readBlock(int blockIndex) throws IOException {
- byte[] cached = cache.getBlock(path.toString(), blockIndex);
+ byte[] cached = cache.getBlock(cacheKey, blockIndex);
if (cached != null) {
return cached;
}
@@ -153,7 +160,7 @@ public class CachingSeekableInputStream extends
SeekableInputStream implements V
byte[] data = readRemote(offset, readSize);
- cache.putBlock(path.toString(), blockIndex, data);
+ cache.putBlock(cacheKey, blockIndex, data);
return data;
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalCacheManager.java
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalCacheManager.java
index 4929bf5cd8..7b56078232 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalCacheManager.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalCacheManager.java
@@ -34,4 +34,7 @@ public interface LocalCacheManager {
long getFileSize(String filePath);
void putFileSize(String filePath, long size);
+
+ /** Invalidates all entries whose file path starts with the given prefix.
*/
+ default void invalidate(String filePathPrefix) {}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalDiskCacheManager.java
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalDiskCacheManager.java
index 6dfc710e3b..d020feac42 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalDiskCacheManager.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalDiskCacheManager.java
@@ -44,6 +44,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class LocalDiskCacheManager implements LocalCacheManager {
private static final Logger LOG =
LoggerFactory.getLogger(LocalDiskCacheManager.class);
+ private static final String CACHE_FORMAT_VERSION = "v2";
private final File cacheDir;
private final long maxSizeBytes;
@@ -56,7 +57,8 @@ public class LocalDiskCacheManager implements
LocalCacheManager {
private long currentSize;
public LocalDiskCacheManager(String cacheDir, long maxSizeBytes, int
blockSize) {
- this.cacheDir = new File(cacheDir);
+ this.cacheDir =
+ new File(new File(cacheDir, CACHE_FORMAT_VERSION),
"block-size-" + blockSize);
this.maxSizeBytes = maxSizeBytes;
this.blockSize = blockSize;
this.entryIndex = new LinkedHashMap<>(64, 0.75f, true);
@@ -71,15 +73,27 @@ public class LocalDiskCacheManager implements
LocalCacheManager {
public byte[] getBlock(String filePath, int blockIndex) {
File path = cachePath(filePath, blockIndex);
String cacheKey = path.getPath();
+ boolean needEvict = false;
synchronized (lock) {
if (!entryIndex.containsKey(cacheKey)) {
- return null;
+ if (!path.isFile()) {
+ return null;
+ }
+ long size = path.length();
+ entryIndex.put(cacheKey, size);
+ currentSize += size;
+ needEvict = maxSizeBytes < Long.MAX_VALUE && currentSize >
maxSizeBytes;
+ } else {
+ // access to update LRU order
+ entryIndex.get(cacheKey);
}
- // access to update LRU order
- entryIndex.get(cacheKey);
}
try {
- return Files.readAllBytes(path.toPath());
+ byte[] data = Files.readAllBytes(path.toPath());
+ if (needEvict) {
+ evict();
+ }
+ return data;
} catch (IOException e) {
LOG.debug("Failed to read cache block: {}", path, e);
synchronized (lock) {
@@ -125,8 +139,8 @@ public class LocalDiskCacheManager implements
LocalCacheManager {
boolean needEvict = false;
synchronized (lock) {
- entryIndex.put(cacheKey, (long) data.length);
- currentSize += data.length;
+ Long previousSize = entryIndex.put(cacheKey, (long) data.length);
+ currentSize += data.length - (previousSize == null ? 0 :
previousSize);
needEvict = maxSizeBytes < Long.MAX_VALUE && currentSize >
maxSizeBytes;
}
if (needEvict) {
@@ -187,7 +201,7 @@ public class LocalDiskCacheManager implements
LocalCacheManager {
}
private File cachePath(String filePath, int blockIndex) {
- String key = filePath + ":" + blockIndex;
+ String key = blockSize + ":" + filePath + ":" + blockIndex;
String hex = sha256Hex(key);
String prefix = hex.substring(0, 2);
return new File(new File(cacheDir, prefix), hex);
diff --git
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalMemoryCacheManager.java
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalMemoryCacheManager.java
index 3e7baab21e..e92cb88412 100644
---
a/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalMemoryCacheManager.java
+++
b/paimon-common/src/main/java/org/apache/paimon/fs/cache/LocalMemoryCacheManager.java
@@ -89,6 +89,21 @@ public class LocalMemoryCacheManager implements
LocalCacheManager {
fileSizeCache.put(filePath, size);
}
+ @Override
+ public void invalidate(String filePathPrefix) {
+ synchronized (lock) {
+ Iterator<Map.Entry<BlockKey, byte[]>> iterator =
cache.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<BlockKey, byte[]> entry = iterator.next();
+ if (entry.getKey().filePath.startsWith(filePathPrefix)) {
+ currentSize -= entry.getValue().length;
+ iterator.remove();
+ }
+ }
+ }
+ fileSizeCache.keySet().removeIf(filePath ->
filePath.startsWith(filePathPrefix));
+ }
+
private static class BlockKey {
final String filePath;
final int blockIndex;
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 688ae36fc1..dec0d7b7d4 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
@@ -25,18 +25,29 @@ import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
import org.apache.paimon.utils.FileType;
+import org.apache.paimon.utils.InstantiationUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.time.Duration;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.paimon.options.CatalogOptions.LOCAL_CACHE_DIR;
+import static org.apache.paimon.options.CatalogOptions.LOCAL_CACHE_ENABLED;
+import static org.apache.paimon.options.CatalogOptions.LOCAL_CACHE_MAX_SIZE;
+import static org.apache.paimon.options.CatalogOptions.LOCAL_CACHE_WHITELIST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -52,6 +63,7 @@ class CachingFileIOTest {
@BeforeEach
void setUp() {
cacheDir = tempDir.resolve("cache").toString();
+ MockFileIO.resetGlobalInputStreamCalls();
}
@Test
@@ -103,7 +115,7 @@ class CachingFileIOTest {
assertThat(result).isEqualTo(data);
}
- assertThat(delegate.getFileStatusCallCount("snapshot-1")).isEqualTo(1);
+ assertThat(delegate.getFileStatusCallCount("snapshot-1")).isEqualTo(2);
}
@Test
@@ -139,6 +151,233 @@ class CachingFileIOTest {
}
}
+ @Test
+ void testCacheIsRecreatedAfterSerialization() throws Exception {
+ byte[] data = "index data".getBytes();
+ MockFileIO delegate = new MockFileIO();
+ delegate.addFile("global-index-uuid.index", data);
+
+ CatalogContext context = localCacheContext();
+ CachingFileIO cachingIO = newCatalogCachingFileIO(delegate, context);
+
+ CachingFileIO restored = InstantiationUtil.clone(cachingIO);
+
+ try (SeekableInputStream stream =
+ restored.newInputStream(new Path("global-index-uuid.index"))) {
+ assertThat(stream).isInstanceOf(CachingSeekableInputStream.class);
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+ restored.close();
+ }
+
+ @Test
+ void testDeserializedCopiesOfSameWrapperShareCacheInSameJvm() throws
Exception {
+ byte[] data = "shared index data".getBytes();
+ String fileName = "global-index-shared-uuid.index";
+
+ CatalogContext context = localCacheContext();
+
+ MockFileIO delegate = new MockFileIO();
+ delegate.addFile(fileName, data);
+ CachingFileIO original = newCatalogCachingFileIO(delegate, context);
+ CachingFileIO first = InstantiationUtil.clone(original);
+ CachingFileIO second = InstantiationUtil.clone(original);
+
+ try (SeekableInputStream stream = first.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(1);
+
+ try (SeekableInputStream stream = second.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(1);
+
+ first.close();
+ second.close();
+ }
+
+ @Test
+ void testDeserializedWrappersFromDifferentManagersDoNotShareCache() throws
Exception {
+ byte[] firstData = "first catalog data".getBytes();
+ byte[] secondData = "other catalog data".getBytes();
+ String fileName = "global-index-isolated-uuid.index";
+
+ CatalogContext context = localCacheContext();
+
+ MockFileIO firstDelegate = new MockFileIO();
+ firstDelegate.addFile(fileName, firstData);
+ CachingFileIO first =
+ InstantiationUtil.clone(newCatalogCachingFileIO(firstDelegate,
context));
+
+ MockFileIO secondDelegate = new MockFileIO();
+ secondDelegate.addFile(fileName, secondData);
+ CachingFileIO second =
+
InstantiationUtil.clone(newCatalogCachingFileIO(secondDelegate, context));
+
+ try (SeekableInputStream stream = first.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, firstData.length)).isEqualTo(firstData);
+ }
+
+ try (SeekableInputStream stream = second.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream,
secondData.length)).isEqualTo(secondData);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(2);
+
+ first.close();
+ second.close();
+ }
+
+ @Test
+ void testWrappersWithSameManagerUseDifferentSecurityScopes() throws
Exception {
+ byte[] firstData = "first security data".getBytes();
+ byte[] secondData = "other security data".getBytes();
+ String fileName = "global-index-security-scope-uuid.index";
+ CatalogContext context = localCacheContext();
+ LocalCacheManager cache = CachingFileIO.createCacheManager(context);
+
+ MockFileIO firstDelegate = new MockFileIO();
+ firstDelegate.addFile(fileName, firstData);
+ CachingFileIO first =
+ InstantiationUtil.clone(
+ (CachingFileIO)
+ CachingFileIO.wrapWithCachingIfNeeded(
+ firstDelegate, context, cache));
+
+ MockFileIO secondDelegate = new MockFileIO();
+ secondDelegate.addFile(fileName, secondData);
+ CachingFileIO second =
+ InstantiationUtil.clone(
+ (CachingFileIO)
+ CachingFileIO.wrapWithCachingIfNeeded(
+ secondDelegate, context, cache));
+
+ try (SeekableInputStream stream = first.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, firstData.length)).isEqualTo(firstData);
+ }
+ try (SeekableInputStream stream = second.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream,
secondData.length)).isEqualTo(secondData);
+ }
+
+ first.close();
+ second.close();
+ }
+
+ @Test
+ void testDeserializedWrappersShareConfiguredMemoryLimit() throws Exception
{
+ byte[] firstData = "first-cache-block".getBytes();
+ byte[] secondData = "second-cache-data".getBytes();
+ String firstFile = "global-index-limit-first.index";
+ String secondFile = "global-index-limit-second.index";
+ CatalogContext context = localCacheContext(firstData.length);
+
+ MockFileIO firstDelegate = new MockFileIO();
+ firstDelegate.addFile(firstFile, firstData);
+ CachingFileIO first =
+ InstantiationUtil.clone(newCatalogCachingFileIO(firstDelegate,
context));
+
+ MockFileIO secondDelegate = new MockFileIO();
+ secondDelegate.addFile(secondFile, secondData);
+ CachingFileIO second =
+
InstantiationUtil.clone(newCatalogCachingFileIO(secondDelegate, context));
+
+ try (SeekableInputStream stream = first.newInputStream(new
Path(firstFile))) {
+ assertThat(readAll(stream, firstData.length)).isEqualTo(firstData);
+ }
+ try (SeekableInputStream stream = second.newInputStream(new
Path(secondFile))) {
+ assertThat(readAll(stream,
secondData.length)).isEqualTo(secondData);
+ }
+ try (SeekableInputStream stream = first.newInputStream(new
Path(firstFile))) {
+ assertThat(readAll(stream, firstData.length)).isEqualTo(firstData);
+ }
+
+
assertThat(MockFileIO.globalInputStreamCallCount(firstFile)).isEqualTo(2);
+ first.close();
+ second.close();
+ }
+
+ @Test
+ void testDeserializedCacheIsReleasedAfterAllExistingWrappersClose() throws
Exception {
+ byte[] data = "released cache data".getBytes();
+ String fileName = "global-index-released-uuid.index";
+
+ MockFileIO delegate = new MockFileIO();
+ delegate.addFile(fileName, data);
+ CachingFileIO original = newCatalogCachingFileIO(delegate,
localCacheContext());
+
+ CachingFileIO first = InstantiationUtil.clone(original);
+ CachingFileIO second = InstantiationUtil.clone(original);
+ try (SeekableInputStream stream = first.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+ first.close();
+
+ try (SeekableInputStream stream = second.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(1);
+ second.close();
+
+ CachingFileIO third = InstantiationUtil.clone(original);
+ try (SeekableInputStream stream = third.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(2);
+ third.close();
+ }
+
+ @Test
+ void testDiskCacheIsReusedAfterManagerRecreation() throws Exception {
+ byte[] data = "persistent disk cache data".getBytes();
+ String fileName = "global-index-persistent-uuid.index";
+ CatalogContext context = localDiskCacheContext();
+
+ MockFileIO firstDelegate = new MockFileIO();
+ firstDelegate.addFile(fileName, data);
+ CachingFileIO first = newCatalogCachingFileIO(firstDelegate, context);
+ try (SeekableInputStream stream = first.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+ first.close();
+
+ MockFileIO secondDelegate = new MockFileIO();
+ secondDelegate.addFile(fileName, data);
+ CachingFileIO second = newCatalogCachingFileIO(secondDelegate,
context);
+ try (SeekableInputStream stream = second.newInputStream(new
Path(fileName))) {
+ assertThat(readAll(stream, data.length)).isEqualTo(data);
+ }
+
assertThat(MockFileIO.globalInputStreamCallCount(fileName)).isEqualTo(1);
+ second.close();
+ }
+
+ private CatalogContext localCacheContext() {
+ return localCacheContext(null);
+ }
+
+ private CatalogContext localCacheContext(@Nullable Integer maxSize) {
+ Options options = new Options();
+ options.set(LOCAL_CACHE_ENABLED, true);
+ options.set(LOCAL_CACHE_WHITELIST, "global-index");
+ if (maxSize != null) {
+ options.set(LOCAL_CACHE_MAX_SIZE, MemorySize.ofBytes(maxSize));
+ }
+ return CatalogContext.create(options);
+ }
+
+ private CatalogContext localDiskCacheContext() {
+ Options options = new Options();
+ options.set(LOCAL_CACHE_ENABLED, true);
+ options.set(LOCAL_CACHE_DIR, cacheDir);
+ options.set(LOCAL_CACHE_WHITELIST, "global-index");
+ return CatalogContext.create(options);
+ }
+
+ private CachingFileIO newCatalogCachingFileIO(FileIO delegate,
CatalogContext context) {
+ return (CachingFileIO)
+ CachingFileIO.wrapWithCachingIfNeeded(
+ delegate, context,
CachingFileIO.createCacheManager(context));
+ }
+
@Test
void testDataFileNotCached() throws IOException {
byte[] data = "data content".getBytes();
@@ -352,10 +591,22 @@ class CachingFileIOTest {
/** Simple in-memory FileIO for testing. */
private static class MockFileIO implements FileIO {
+ private static final Map<String, AtomicInteger>
GLOBAL_INPUT_STREAM_CALLS =
+ new ConcurrentHashMap<>();
+
private final Map<String, byte[]> files = new HashMap<>();
private final Map<String, Integer> fileStatusCalls = new HashMap<>();
private final Map<String, Integer> newInputStreamCalls = new
HashMap<>();
+ static void resetGlobalInputStreamCalls() {
+ GLOBAL_INPUT_STREAM_CALLS.clear();
+ }
+
+ static int globalInputStreamCallCount(String name) {
+ AtomicInteger count = GLOBAL_INPUT_STREAM_CALLS.get(name);
+ return count == null ? 0 : count.get();
+ }
+
void addFile(String name, byte[] data) {
files.put(name, data);
}
@@ -372,6 +623,9 @@ class CachingFileIOTest {
public SeekableInputStream newInputStream(Path path) throws
IOException {
String name = path.getName();
newInputStreamCalls.merge(name, 1, Integer::sum);
+ GLOBAL_INPUT_STREAM_CALLS
+ .computeIfAbsent(name, ignored -> new AtomicInteger())
+ .incrementAndGet();
byte[] data = files.get(name);
if (data == null) {
throw new IOException("File not found: " + name);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/fs/cache/LocalDiskCacheManagerTest.java
b/paimon-common/src/test/java/org/apache/paimon/fs/cache/LocalDiskCacheManagerTest.java
index d6fb136cf5..251d93d184 100644
---
a/paimon-common/src/test/java/org/apache/paimon/fs/cache/LocalDiskCacheManagerTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/fs/cache/LocalDiskCacheManagerTest.java
@@ -106,6 +106,29 @@ class LocalDiskCacheManagerTest {
assertThat(cache2.getBlock("f", 1)).isEqualTo(data1);
}
+ @Test
+ void testDifferentBlockSizesDoNotSharePhysicalBlocks() {
+ LocalDiskCacheManager smallBlocks = new
LocalDiskCacheManager(cacheDir, Long.MAX_VALUE, 4);
+ byte[] data = "four".getBytes();
+ smallBlocks.putBlock("f", 0, data);
+
+ LocalDiskCacheManager largeBlocks = new
LocalDiskCacheManager(cacheDir, Long.MAX_VALUE, 8);
+ assertThat(largeBlocks.getBlock("f", 0)).isNull();
+ assertThat(largeBlocks.currentSize()).isZero();
+ }
+
+ @Test
+ void testManagerDiscoversBlockWrittenAfterConstruction() {
+ LocalDiskCacheManager first = new LocalDiskCacheManager(cacheDir,
Long.MAX_VALUE, 64);
+ LocalDiskCacheManager second = new LocalDiskCacheManager(cacheDir,
Long.MAX_VALUE, 64);
+ byte[] data = "shared block".getBytes();
+
+ first.putBlock("f", 0, data);
+
+ assertThat(second.getBlock("f", 0)).isEqualTo(data);
+ assertThat(second.currentSize()).isEqualTo(data.length);
+ }
+
@Test
void testCacheDirCreated() {
String deepDir = tempDir.resolve("sub").resolve("deep").toString();
