[core] Unify block cache ownership and avoid redundant SST reads (#9663)
diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/Cache.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/Cache.java
index 8f96cdf..d100f6b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/io/cache/Cache.java
+++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/Cache.java
@@ -18,7 +18,7 @@
package org.apache.paimon.io.cache;
-import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySlice;
import javax.annotation.Nullable;
@@ -28,6 +28,10 @@
/** Cache interface in Paimon. */
public interface Cache {
+ /** Looks up an entry without loading it, recording an access according to the cache policy. */
+ @Nullable
+ CacheValue getIfPresent(CacheKey key);
+
@Nullable
CacheValue get(CacheKey key, Function<CacheKey, CacheValue> supplier);
@@ -44,11 +48,11 @@
/** Value for cache. */
class CacheValue {
- final MemorySegment segment;
+ final MemorySlice slice;
final CacheCallback callback;
- CacheValue(MemorySegment segment, CacheCallback callback) {
- this.segment = segment;
+ CacheValue(MemorySlice slice, CacheCallback callback) {
+ this.slice = slice;
this.callback = callback;
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheBuilder.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheBuilder.java
index 9c4782e..11f9b6b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheBuilder.java
+++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheBuilder.java
@@ -23,6 +23,8 @@
import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.RemovalCause;
+import java.util.function.BiConsumer;
+
/** Builder for a Caffeine cache. */
public class CacheBuilder {
private MemorySize memorySize;
@@ -37,22 +39,28 @@
}
public Cache build() {
+ return build(
+ (key, value) -> {
+ if (value != null) {
+ value.callback.onRemoval(key);
+ }
+ });
+ }
+
+ Cache build(BiConsumer<CacheKey, Cache.CacheValue> onRemoval) {
return new CaffeineCache(
Caffeine.newBuilder()
.weigher(CacheBuilder::weigh)
.maximumWeight(memorySize.getBytes())
- .removalListener(this::onRemoval)
+ .removalListener(
+ (CacheKey key, Cache.CacheValue value, RemovalCause cause) ->
+ onRemoval.accept(key, value))
.executor(Runnable::run)
.build());
}
- private void onRemoval(CacheKey key, Cache.CacheValue value, RemovalCause cause) {
- if (value != null) {
- value.callback.onRemoval(key);
- }
- }
-
private static int weigh(CacheKey cacheKey, Cache.CacheValue cacheValue) {
- return cacheValue.segment.size();
+ // A slice can exclude a trailer while retaining the complete heap allocation.
+ return cacheValue.slice.segment().size();
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheKey.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheKey.java
index caad913..c038196 100644
--- a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheKey.java
+++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheKey.java
@@ -58,6 +58,10 @@
this.hashCode = 31 * hashCode + Boolean.hashCode(isIndex);
}
+ public Path filePath() {
+ return filePath;
+ }
+
public long position() {
return position;
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheManager.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheManager.java
index 4fe040e..1eacce2 100644
--- a/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheManager.java
+++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/CacheManager.java
@@ -19,14 +19,26 @@
package org.apache.paimon.io.cache;
import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.cache.Cache.CacheValue;
+import org.apache.paimon.io.cache.CacheKey.PositionCacheKey;
import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.utils.ExceptionUtils;
import org.apache.paimon.utils.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -35,14 +47,16 @@
private static final Logger LOG = LoggerFactory.getLogger(CacheManager.class);
- /**
- * Refreshing the cache comes with some costs, so not every time we visit the CacheManager, but
- * every 10 visits, refresh the LRU strategy.
- */
- public static final int REFRESH_COUNT = 10;
+ private static final CacheCallback NO_CALLBACK = key -> {};
+
+ // Only loading/removal touches this index. Hits use the shared cache directly. The links are
+ // owned by the manager, never by a reader, and empty file entries are removed immediately.
+ private final Map<Path, FilePages> files = new HashMap<>();
private final Cache dataCache;
private final Cache indexCache;
+ private final long maxDataCacheBytes;
+ private final long maxIndexCacheBytes;
private final boolean offHeap;
public CacheManager(MemorySize maxMemorySize, double highPriorityPoolRatio) {
@@ -57,11 +71,16 @@
MemorySize.ofBytes((long) (maxMemorySize.getBytes() * highPriorityPoolRatio));
MemorySize dataCacheSize =
MemorySize.ofBytes((long) (maxMemorySize.getBytes() * (1 - highPriorityPoolRatio)));
- this.dataCache = CacheBuilder.newBuilder().maximumWeight(dataCacheSize).build();
+ this.maxDataCacheBytes = dataCacheSize.getBytes();
+ this.maxIndexCacheBytes =
+ highPriorityPoolRatio == 0 ? maxDataCacheBytes : indexCacheSize.getBytes();
+ this.dataCache =
+ CacheBuilder.newBuilder().maximumWeight(dataCacheSize).build(this::onRemoval);
if (highPriorityPoolRatio == 0) {
this.indexCache = dataCache;
} else {
- this.indexCache = CacheBuilder.newBuilder().maximumWeight(indexCacheSize).build();
+ this.indexCache =
+ CacheBuilder.newBuilder().maximumWeight(indexCacheSize).build(this::onRemoval);
}
this.offHeap = offHeap;
LOG.info(
@@ -86,20 +105,71 @@
return indexCache;
}
+ /** Whether a page fits its configured pool; this does not guarantee admission. */
+ public boolean canFitPage(int pageSize, boolean isIndex) {
+ return pageSize <= (isIndex ? maxIndexCacheBytes : maxDataCacheBytes);
+ }
+
+ /** Returns a cached decoded range, or null if a load is needed. */
+ @Nullable
+ public MemorySlice getPageSliceIfPresent(CacheKey key) {
+ CacheValue value = (key.isIndex() ? indexCache : dataCache).getIfPresent(key);
+ return value == null ? null : value.slice;
+ }
+
+ /** Returns a cached page without retaining a loader on the hit path. */
+ @Nullable
+ public MemorySegment getPageIfPresent(CacheKey key) {
+ MemorySlice slice = getPageSliceIfPresent(key);
+ return slice == null ? null : asSegment(slice);
+ }
+
+ public MemorySegment getPage(CacheKey key, CacheReader reader) {
+ return getPage(key, reader, NO_CALLBACK);
+ }
+
public MemorySegment getPage(CacheKey key, CacheReader reader, CacheCallback callback) {
+ return asSegment(getPageSlice(key, reader, MemorySlice::wrap, callback));
+ }
+
+ private static MemorySegment asSegment(MemorySlice slice) {
+ if (slice.offset() == 0 && slice.length() == slice.segment().size()) {
+ return slice.segment();
+ }
+ // The same key may have been loaded through the slice API. Do not expose framing bytes
+ // outside the decoded range through the original whole-segment API.
+ return MemorySegment.wrap(slice.copyBytes());
+ }
+
+ /** Caches the decoded range, allowing an uncompressed block to share its read buffer. */
+ public MemorySlice getPageSlice(
+ CacheKey key, CacheReader reader, Function<byte[], MemorySlice> decoder) {
+ return getPageSlice(key, reader, decoder, NO_CALLBACK);
+ }
+
+ private MemorySlice getPageSlice(
+ CacheKey key,
+ CacheReader reader,
+ Function<byte[], MemorySlice> decoder,
+ CacheCallback callback) {
Cache cache = key.isIndex() ? indexCache : dataCache;
Cache.CacheValue value =
cache.get(
key,
k -> {
try {
- return new Cache.CacheValue(
- toMemorySegment(reader.read(key)), callback);
+ Page page =
+ new Page(
+ key,
+ toMemorySlice(decoder.apply(reader.read(key))),
+ callback);
+ register(page);
+ return page;
} catch (IOException e) {
throw new RuntimeException(e);
}
});
- return checkNotNull(value, "Cache result for key(%s) is null", key).segment;
+ return checkNotNull(value, "Cache result for key(%s) is null", key).slice;
}
public boolean contains(CacheKey key) {
@@ -118,13 +188,115 @@
}
}
- private MemorySegment toMemorySegment(byte[] bytes) {
- if (!offHeap) {
- return MemorySegment.wrap(bytes);
+ /**
+ * Invalidates this file's registered pages, including pages loaded by other readers. Concurrent
+ * reads may repopulate the cache; callers must stop reads before deleting or replacing a file.
+ */
+ public void invalidFile(Path filePath) throws IOException {
+ List<Page> pages = new ArrayList<>();
+ synchronized (files) {
+ FilePages file = files.remove(filePath);
+ if (file == null) {
+ return;
+ }
+ Page page = file.head;
+ while (page != null) {
+ pages.add(page);
+ Page next = page.next;
+ page.owner = null;
+ page.previous = null;
+ page.next = null;
+ page = next;
+ }
+ file.head = null;
}
- MemorySegment segment = MemorySegment.allocateOffHeapMemory(bytes.length);
- segment.put(0, bytes);
- return segment;
+
+ // Never call the cache while holding the file index lock: loaders and removal listeners
+ // acquire that lock from inside Caffeine. Delete by identity so an old invalidation cannot
+ // delete a replacement loaded for the same key.
+ Throwable collected = null;
+ for (Page page : pages) {
+ try {
+ invalidPage(page.key, page);
+ } catch (Throwable t) {
+ collected = ExceptionUtils.firstOrSuppressed(t, collected);
+ }
+ }
+ if (collected != null) {
+ if (collected instanceof Error) {
+ throw (Error) collected;
+ }
+ if (collected instanceof RuntimeException) {
+ throw (RuntimeException) collected;
+ }
+ throw new IOException(collected);
+ }
+ }
+
+ /** Invalidates a particular page without removing a newer value for the same key. */
+ protected void invalidPage(CacheKey key, CacheValue expected) {
+ (key.isIndex() ? indexCache : dataCache).asMap().remove(key, expected);
+ }
+
+ private void register(Page page) {
+ if (!(page.key instanceof PositionCacheKey)) {
+ return;
+ }
+ Path path = ((PositionCacheKey) page.key).filePath();
+ synchronized (files) {
+ FilePages file = files.computeIfAbsent(path, ignored -> new FilePages(path));
+ page.owner = file;
+ page.next = file.head;
+ if (file.head != null) {
+ file.head.previous = page;
+ }
+ file.head = page;
+ }
+ }
+
+ private void onRemoval(CacheKey key, CacheValue value) {
+ if (value == null) {
+ return;
+ }
+ if (value instanceof Page) {
+ Page page = (Page) value;
+ synchronized (files) {
+ FilePages file = page.owner;
+ if (file != null) {
+ if (page.previous == null) {
+ file.head = page.next;
+ } else {
+ page.previous.next = page.next;
+ }
+ if (page.next != null) {
+ page.next.previous = page.previous;
+ }
+ page.owner = null;
+ page.previous = null;
+ page.next = null;
+ if (file.head == null) {
+ files.remove(file.path, file);
+ }
+ }
+ }
+ }
+ value.callback.onRemoval(key);
+ }
+
+ @VisibleForTesting
+ int cachedFileCount() {
+ synchronized (files) {
+ return files.size();
+ }
+ }
+
+ private MemorySlice toMemorySlice(MemorySlice decoded) {
+ if (!offHeap) {
+ return decoded;
+ }
+ MemorySegment segment = MemorySegment.allocateOffHeapMemory(decoded.length());
+ decoded.segment().copyTo(decoded.offset(), segment, 0, decoded.length());
+ return MemorySlice.wrap(segment);
}
@Override
@@ -135,25 +307,25 @@
}
}
- /** The container for the segment. */
- public static class SegmentContainer {
+ private static class FilePages {
+ private final Path path;
+ private Page head;
- private final MemorySegment segment;
-
- private int accessCount;
-
- public SegmentContainer(MemorySegment segment) {
- this.segment = segment;
- this.accessCount = 0;
+ private FilePages(Path path) {
+ this.path = path;
}
+ }
- public MemorySegment access() {
- this.accessCount++;
- return segment;
- }
+ /** Links are guarded by the file index lock; they are detached before notifying callers. */
+ private static class Page extends CacheValue {
+ private final CacheKey key;
+ private FilePages owner;
+ private Page previous;
+ private Page next;
- public int getAccessCount() {
- return accessCount;
+ private Page(CacheKey key, MemorySlice slice, CacheCallback callback) {
+ super(slice, callback);
+ this.key = key;
}
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/CaffeineCache.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/CaffeineCache.java
index 3bb47f8..9d86656 100644
--- a/paimon-common/src/main/java/org/apache/paimon/io/cache/CaffeineCache.java
+++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/CaffeineCache.java
@@ -38,7 +38,16 @@
@Nullable
@Override
+ public CacheValue getIfPresent(CacheKey key) {
+ // Every hit contributes to admission and eviction, including when the hot set changes.
+ return this.cache.getIfPresent(key);
+ }
+
+ @Nullable
+ @Override
public CacheValue get(CacheKey key, Function<CacheKey, CacheValue> supplier) {
+ // The loading path already follows a failed probe. Let Caffeine handle atomic loading
+ // directly instead of performing another quiet lookup.
return this.cache.get(key, supplier);
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/sst/BlockCache.java b/paimon-common/src/main/java/org/apache/paimon/sst/BlockCache.java
index 32e341b..0f72283 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/BlockCache.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/BlockCache.java
@@ -23,17 +23,14 @@
import org.apache.paimon.fs.VectoredReadable;
import org.apache.paimon.io.cache.CacheKey;
import org.apache.paimon.io.cache.CacheManager;
-import org.apache.paimon.io.cache.CacheManager.SegmentContainer;
import org.apache.paimon.memory.MemorySegment;
-import org.apache.paimon.utils.ExceptionUtils;
+import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.utils.IOUtils;
+import javax.annotation.Nullable;
+
import java.io.Closeable;
import java.io.IOException;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
/** Cache for block reading. */
@@ -42,13 +39,11 @@
private final Path filePath;
private final SeekableInputStream input;
private final CacheManager cacheManager;
- private final Map<CacheKey, SegmentContainer> blocks;
public BlockCache(Path filePath, SeekableInputStream input, CacheManager cacheManager) {
this.filePath = filePath;
this.input = input;
this.cacheManager = cacheManager;
- this.blocks = new ConcurrentHashMap<>();
}
private byte[] readFrom(long offset, int length) throws IOException {
@@ -68,53 +63,41 @@
long position, int length, Function<byte[], byte[]> decompressFunc, boolean isIndex) {
CacheKey cacheKey = CacheKey.forPosition(filePath, position, length, isIndex);
- SegmentContainer container = blocks.get(cacheKey);
- if (container == null || container.getAccessCount() == CacheManager.REFRESH_COUNT) {
- MemorySegment segment =
- cacheManager.getPage(
- cacheKey,
- key -> {
- byte[] bytes = readFrom(position, length);
- return decompressFunc.apply(bytes);
- },
- blocks::remove);
- container = new SegmentContainer(segment);
- if (cacheManager.contains(cacheKey)) {
- blocks.put(cacheKey, container);
- }
+ // Construct the capturing loader only on misses; hits use the shared cache directly.
+ MemorySegment cached = cacheManager.getPageIfPresent(cacheKey);
+ if (cached != null) {
+ return cached;
}
- return container.access();
+ return cacheManager.getPage(
+ cacheKey,
+ key -> {
+ byte[] bytes = readFrom(position, length);
+ return decompressFunc.apply(bytes);
+ });
+ }
+
+ /** Returns a decoded block if resident, without reading the file. */
+ @Nullable
+ public MemorySlice getBlockSliceIfPresent(long position, int length, boolean isIndex) {
+ return cacheManager.getPageSliceIfPresent(
+ CacheKey.forPosition(filePath, position, length, isIndex));
+ }
+
+ /**
+ * Reads and caches a decoded block without copying an uncompressed payload out of its buffer.
+ */
+ public MemorySlice getBlockSlice(
+ long position, int length, Function<byte[], MemorySlice> decoder, boolean isIndex) {
+ CacheKey cacheKey = CacheKey.forPosition(filePath, position, length, isIndex);
+ MemorySlice cached = cacheManager.getPageSliceIfPresent(cacheKey);
+ if (cached != null) {
+ return cached;
+ }
+ return cacheManager.getPageSlice(cacheKey, key -> readFrom(position, length), decoder);
}
@Override
public void close() throws IOException {
- // Every page has to be handed back to the shared cache manager. Stopping at the first
- // failure would leave the rest of this file's pages resident in a cache that is shared
- // across readers, with nothing left holding a reference to invalidate them later.
- Set<CacheKey> sets = new HashSet<>(blocks.keySet());
- Throwable collected = null;
- for (CacheKey key : sets) {
- try {
- cacheManager.invalidPage(key);
- } catch (Throwable t) {
- collected = ExceptionUtils.firstOrSuppressed(t, collected);
- }
- }
- if (collected != null) {
- rethrowAsIOException(collected);
- }
- }
-
- private static void rethrowAsIOException(Throwable failure) throws IOException {
- if (failure instanceof IOException) {
- throw (IOException) failure;
- }
- if (failure instanceof Error) {
- throw (Error) failure;
- }
- if (failure instanceof RuntimeException) {
- throw (RuntimeException) failure;
- }
- throw new IOException(failure);
+ cacheManager.invalidFile(filePath);
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
index a501172..8346464 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
@@ -20,7 +20,6 @@
import org.apache.paimon.compression.BlockCompressionFactory;
import org.apache.paimon.compression.BlockDecompressor;
-import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.memory.MemorySliceInput;
import org.apache.paimon.utils.ExceptionUtils;
@@ -50,6 +49,10 @@
private final BlockReader indexBlock;
@Nullable private final FileBasedBloomFilter bloomFilter;
+ private boolean hasCompressedBlocks;
+ private boolean loadBloomOnMiss;
+ private long dataBytesWithoutBloom;
+
public SstFileReader(
Comparator<MemorySlice> comparator,
BlockCache blockCache,
@@ -57,8 +60,9 @@
@Nullable FileBasedBloomFilter bloomFilter) {
this.comparator = comparator;
this.blockCache = blockCache;
- this.indexBlock = readBlock(indexBlockHandle, true);
this.bloomFilter = bloomFilter;
+ this.loadBloomOnMiss = bloomFilter != null && bloomFilter.isCacheable();
+ this.indexBlock = readBlock(indexBlockHandle, true);
}
/**
@@ -69,8 +73,19 @@
*/
@Nullable
public byte[] lookup(byte[] key) throws IOException {
- if (bloomFilter != null && !bloomFilter.testHash(MurmurHashUtils.hashBytes(key))) {
- return null;
+ int hash = 0;
+ Boolean bloomMatch = null;
+ if (bloomFilter != null) {
+ hash = MurmurHashUtils.hashBytes(key);
+ bloomMatch = bloomFilter.testHashIfPresent(hash);
+ if (bloomMatch != null) {
+ // Another reader may have admitted a previously rejected filter.
+ loadBloomOnMiss = true;
+ dataBytesWithoutBloom = 0;
+ if (!bloomMatch) {
+ return null;
+ }
+ }
}
MemorySlice keySlice = MemorySlice.wrap(key);
@@ -80,8 +95,48 @@
// if indexIterator does not have a next, it means the key does not exist in this iterator
if (indexBlockIterator.hasNext()) {
+ BlockHandle handle =
+ BlockHandle.readBlockHandle(indexBlockIterator.next().getValue().toInput());
+ MemorySlice cachedData = null;
+ boolean bloomProbed = bloomMatch != null;
+ if (bloomFilter != null && bloomMatch == null) {
+ // A missing filter must not cause I/O when the exact data is already resident.
+ cachedData =
+ blockCache.getBlockSliceIfPresent(
+ handle.offset(), handle.getFullBlockSize(), false);
+ if (cachedData == null
+ && (loadBloomOnMiss
+ || hasCompressedBlocks
+ || bloomFilter.size() < handle.getFullBlockSize())) {
+ boolean matches = bloomFilter.testHash(hash);
+ // Try to warm a filter that fits, but after rejection only reload it when
+ // its read cost or the saved decompression justifies bypassing the data.
+ loadBloomOnMiss = bloomFilter.isCached();
+ dataBytesWithoutBloom = 0;
+ bloomProbed = true;
+ if (!matches) {
+ return null;
+ }
+ }
+ }
+ BlockReader dataBlock =
+ cachedData == null
+ ? readBlock(handle, false)
+ : createBlockReader(handle, cachedData);
+ if (!bloomProbed
+ && cachedData == null
+ && bloomFilter != null
+ && bloomFilter.isCacheable()) {
+ // Retry after cold data accesses cost as much as a filter read. This allows
+ // recovery from transient rejection without loading Bloom on every miss.
+ dataBytesWithoutBloom += handle.getFullBlockSize();
+ if (dataBytesWithoutBloom >= bloomFilter.size()) {
+ loadBloomOnMiss = true;
+ dataBytesWithoutBloom = 0;
+ }
+ }
// seek the current iterator to the key
- BlockIterator current = getNextBlock(indexBlockIterator);
+ BlockIterator current = dataBlock.iterator();
if (current.seekTo(keySlice)) {
return current.next().getValue().copyBytes();
}
@@ -119,41 +174,46 @@
* @return The reader of the target block.
*/
private BlockReader readBlock(BlockHandle blockHandle, boolean index) {
- // read block trailer
- MemorySegment trailerData =
- blockCache.getBlock(
- blockHandle.offset() + blockHandle.size(),
- BlockTrailer.ENCODED_LENGTH,
- b -> b,
- true);
- BlockTrailer blockTrailer =
- BlockTrailer.readBlockTrailer(MemorySlice.wrap(trailerData).toInput());
-
- MemorySegment unCompressedBlock =
- blockCache.getBlock(
+ MemorySlice unCompressedBlock =
+ blockCache.getBlockSlice(
blockHandle.offset(),
- blockHandle.size(),
- bytes -> decompressBlock(bytes, blockTrailer),
+ blockHandle.getFullBlockSize(),
+ this::decompressBlock,
index);
- return BlockReader.create(MemorySlice.wrap(unCompressedBlock), comparator);
+ return createBlockReader(blockHandle, unCompressedBlock);
}
- private byte[] decompressBlock(byte[] compressedBytes, BlockTrailer blockTrailer) {
- MemorySegment compressed = MemorySegment.wrap(compressedBytes);
+ private BlockReader createBlockReader(BlockHandle blockHandle, MemorySlice unCompressedBlock) {
+ // The writer uses compression only when it reduces the payload size. Comparing sizes
+ // also detects compressed blocks obtained from the shared cache, including the index.
+ if (bloomFilter != null && unCompressedBlock.length() != blockHandle.size()) {
+ hasCompressedBlocks = true;
+ }
+ return BlockReader.create(unCompressedBlock, comparator);
+ }
+
+ private MemorySlice decompressBlock(byte[] blockBytes) {
+ MemorySlice fullBlock = MemorySlice.wrap(blockBytes);
+ int blockSize = blockBytes.length - BlockTrailer.ENCODED_LENGTH;
+ MemorySlice compressed = fullBlock.slice(0, blockSize);
+ BlockTrailer blockTrailer =
+ BlockTrailer.readBlockTrailer(
+ fullBlock.slice(blockSize, BlockTrailer.ENCODED_LENGTH).toInput());
int crc32cCode = crc32c(compressed, blockTrailer.getCompressionType());
- checkArgument(
- blockTrailer.getCrc32c() == crc32cCode,
- String.format(
- "Expected CRC32C(%d) but found CRC32C(%d)",
- blockTrailer.getCrc32c(), crc32cCode));
+ if (blockTrailer.getCrc32c() != crc32cCode) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected CRC32C(%d) but found CRC32C(%d)",
+ blockTrailer.getCrc32c(), crc32cCode));
+ }
// decompress data
BlockCompressionFactory compressionFactory =
BlockCompressionFactory.create(blockTrailer.getCompressionType());
if (compressionFactory == null) {
- return compressedBytes;
+ return compressed;
} else {
- MemorySliceInput compressedInput = MemorySlice.wrap(compressed).toInput();
+ MemorySliceInput compressedInput = compressed.toInput();
byte[] uncompressed = new byte[compressedInput.readVarLenInt()];
BlockDecompressor decompressor = compressionFactory.getDecompressor();
int uncompressedLength =
@@ -164,7 +224,7 @@
uncompressed,
0);
checkArgument(uncompressedLength == uncompressed.length);
- return uncompressed;
+ return MemorySlice.wrap(uncompressed);
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java b/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java
index 020df5e..23fac02 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java
@@ -64,6 +64,10 @@
return this.memorySegment;
}
+ int memoryOffset() {
+ return offset;
+ }
+
MemorySlice getMemorySlice() {
return new MemorySlice(
checkNotNull(memorySegment, "MemorySegment is not set."), offset, byteLength);
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
index 0a96471..c9d0a34 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
@@ -113,6 +113,15 @@
}
public boolean testHash(int hash1) {
+ return testHash(hash1, bitSet.getMemorySegment(), bitSet.memoryOffset());
+ }
+
+ /** Probes a cached filter without retaining or mutating its backing segment. */
+ boolean testHash(int hash1, MemorySegment segment) {
+ return testHash(hash1, segment, 0);
+ }
+
+ private boolean testHash(int hash1, MemorySegment segment, int offset) {
int hash2 = hash1 >>> 16;
for (int i = 1; i <= numHashFunctions; i++) {
@@ -122,7 +131,7 @@
combinedHash = ~combinedHash;
}
int pos = combinedHash % bitSet.bitSize();
- if (!bitSet.get(pos)) {
+ if ((segment.get(offset + (pos >>> 3)) & (1 << (pos & 7))) == 0) {
return false;
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java b/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
index 4e8bc2b..500d018 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
@@ -22,7 +22,6 @@
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.VectoredReadable;
-import org.apache.paimon.io.cache.CacheCallback;
import org.apache.paimon.io.cache.CacheKey;
import org.apache.paimon.io.cache.CacheKey.PositionCacheKey;
import org.apache.paimon.io.cache.CacheManager;
@@ -34,7 +33,6 @@
import java.io.Closeable;
import java.io.IOException;
-import static org.apache.paimon.io.cache.CacheManager.REFRESH_COUNT;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Util to apply a built bloom filter . */
@@ -44,8 +42,7 @@
private final CacheManager cacheManager;
private final BloomFilter filter;
private final PositionCacheKey cacheKey;
-
- private int accessCount;
+ private final boolean cacheable;
public FileBasedBloomFilter(
SeekableInputStream input,
@@ -58,8 +55,8 @@
this.cacheManager = cacheManager;
checkArgument(expectedEntries >= 0);
this.filter = new BloomFilter(expectedEntries, readLength);
- this.accessCount = 0;
this.cacheKey = CacheKey.forPosition(filePath, readOffset, readLength, true);
+ this.cacheable = cacheManager.canFitPage(readLength, true);
}
@Nullable
@@ -80,18 +77,34 @@
bloomFilterHandle.size());
}
+ /** Whether the complete filter fits the configured index cache budget. */
+ public boolean isCacheable() {
+ return cacheable;
+ }
+
+ /** Whether the filter is currently resident; fitting the budget does not imply admission. */
+ public boolean isCached() {
+ return cacheManager.contains(cacheKey);
+ }
+
+ /** Number of bytes needed to read the complete filter. */
+ public int size() {
+ return cacheKey.length();
+ }
+
+ /** Tests a resident filter, or returns null without reading the file on a cache miss. */
+ @Nullable
+ public Boolean testHashIfPresent(int hash) {
+ MemorySegment segment = cacheManager.getPageIfPresent(cacheKey);
+ return segment == null ? null : filter.testHash(hash, segment);
+ }
+
public boolean testHash(int hash) {
- accessCount++;
- // we should refresh cache in LRU, but we cannot refresh everytime, it is costly.
- // so we introduce a refresh count to reduce refresh
- if (accessCount == REFRESH_COUNT || filter.getMemorySegment() == null) {
- MemorySegment segment =
- cacheManager.getPage(
- cacheKey, this::readBytes, new BloomFilterCallBack(filter));
- filter.setMemorySegment(segment, 0);
- accessCount = 0;
+ MemorySegment segment = cacheManager.getPageIfPresent(cacheKey);
+ if (segment == null) {
+ segment = cacheManager.getPage(cacheKey, this::readBytes);
}
- return filter.testHash(hash);
+ return filter.testHash(hash, segment);
}
private byte[] readBytes(CacheKey k) throws IOException {
@@ -117,19 +130,4 @@
public void close() throws IOException {
cacheManager.invalidPage(cacheKey);
}
-
- /** Call back for cache manager. */
- private static class BloomFilterCallBack implements CacheCallback {
-
- private final BloomFilter bloomFilter;
-
- private BloomFilterCallBack(BloomFilter bloomFilter) {
- this.bloomFilter = bloomFilter;
- }
-
- @Override
- public void onRemoval(CacheKey key) {
- this.bloomFilter.unsetMemorySegment();
- }
- }
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderCloseTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderCloseTest.java
index 27b10a7..e6ea96c 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderCloseTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderCloseTest.java
@@ -29,6 +29,7 @@
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.io.cache.Cache;
import org.apache.paimon.io.cache.CacheKey;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.options.MemorySize;
@@ -197,7 +198,7 @@
};
}
- /** Fails page invalidation, which is what closing the reader's bloom filter does. */
+ /** Fails page invalidation during reader close. */
private static class FailingCacheManager extends CacheManager {
private boolean failing = false;
@@ -207,6 +208,14 @@
}
@Override
+ protected void invalidPage(CacheKey key, Cache.CacheValue expected) {
+ if (failing) {
+ throw new RuntimeException("cache is down");
+ }
+ super.invalidPage(key, expected);
+ }
+
+ @Override
public void invalidPage(CacheKey key) {
if (failing) {
throw new RuntimeException("cache is down");
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
index 7828a60..3946f82 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
@@ -290,11 +290,11 @@
*
* <ul>
* <li>ConcurrentHashMap reader cache in LazyFilteredBTreeReader
- * <li>BlockCache.getBlock() check-then-act race under eviction pressure
+ * <li>Shared block loading under eviction pressure
* <li>CacheManager eviction callbacks racing with concurrent reads
- * <li>BTreeIndexReader readLock contention across query types
+ * <li>Concurrent BTreeIndexReader queries
* <li>LazyField initialization race for null bitmaps
- * <li>SegmentContainer.accessCount non-atomic increment under contention
+ * <li>Concurrent access to the shared file cache index
* </ul>
*/
@TestTemplate
diff --git a/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerFileTest.java b/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerFileTest.java
new file mode 100644
index 0000000..f4694c9
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerFileTest.java
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.io.cache;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.options.MemorySize;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the shared cache's file index and concurrent file invalidation. */
+@Timeout(30)
+class CacheManagerFileTest {
+
+ @ParameterizedTest
+ @ValueSource(longs = {0, 1})
+ void testRejectedPagesLeaveNoFileIndex(long capacity) throws Exception {
+ try (CacheManager manager = new CacheManager(MemorySize.ofBytes(capacity), 0)) {
+ for (int i = 0; i < 100; i++) {
+ CacheKey key = key(new Path("file-" + i));
+ assertThat(manager.getPage(key, ignored -> new byte[] {1, 2}).getHeapMemory())
+ .containsExactly(1, 2);
+ assertThat(manager.dataCache().asMap()).isEmpty();
+ assertThat(manager.cachedFileCount()).isZero();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDecodedRangeChargesItsBackingAllocation(boolean offHeap) throws Exception {
+ byte[] encoded = new byte[] {99, 1, 2, 3, 4, 99};
+ Path file = new Path("decoded-file");
+ CacheKey key = CacheKey.forPosition(file, 0, encoded.length, false);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofBytes(5), 0)
+ : new CacheManager(MemorySize.ofBytes(5), 0)) {
+ MemorySlice decoded =
+ manager.getPageSlice(
+ key, ignored -> encoded, bytes -> MemorySlice.wrap(bytes).slice(1, 4));
+ assertThat(decoded.copyBytes()).containsExactly(1, 2, 3, 4);
+ if (offHeap) {
+ assertThat(decoded.segment().isOffHeap()).isTrue();
+ assertThat(decoded.segment().size()).isEqualTo(4);
+ } else {
+ assertThat(decoded.segment().getHeapMemory()).isSameAs(encoded);
+ }
+ // Heap retains all six bytes; off-heap copies only the four-byte decoded range.
+ assertThat(manager.contains(key)).isEqualTo(offHeap);
+ manager.invalidFile(file);
+ assertThat(manager.cachedFileCount()).isZero();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testWholePageAccessHonorsCachedDecodedRange(boolean offHeap) throws Exception {
+ CacheKey key = key(new Path("decoded-file"));
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofKibiBytes(64), 0)
+ : new CacheManager(MemorySize.ofKibiBytes(64), 0)) {
+ manager.getPageSlice(
+ key,
+ ignored -> new byte[] {99, 1, 2, 99},
+ bytes -> MemorySlice.wrap(bytes).slice(1, 2));
+ MemorySegment page =
+ manager.getPage(
+ key,
+ ignored -> {
+ throw new AssertionError("Decoded page should be cached");
+ });
+ assertThat(manager.getPageSliceIfPresent(key).copyBytes()).containsExactly(1, 2);
+ MemorySegment cached = manager.getPageIfPresent(key);
+ byte[] cachedBytes = new byte[cached.size()];
+ cached.get(0, cachedBytes);
+ assertThat(cachedBytes).containsExactly(1, 2);
+ assertThat(cached.isOffHeap()).isEqualTo(offHeap);
+ byte[] bytes = new byte[page.size()];
+ page.get(0, bytes);
+ assertThat(bytes).containsExactly(1, 2);
+ assertThat(page.isOffHeap()).isEqualTo(offHeap);
+ manager.invalidPage(key);
+ assertThat(manager.getPageIfPresent(key)).isNull();
+ assertThat(manager.getPageSliceIfPresent(key)).isNull();
+ }
+ }
+
+ @Test
+ void testPresentPagesUseTheirOwnPriorityPool() throws Exception {
+ Path file = new Path("priority-file");
+ CacheKey dataKey = CacheKey.forPosition(file, 0, 2, false);
+ CacheKey indexKey = CacheKey.forPosition(file, 0, 2, true);
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0.5)) {
+ assertThat(manager.getPageIfPresent(dataKey)).isNull();
+ assertThat(manager.getPageSliceIfPresent(indexKey)).isNull();
+ manager.getPage(dataKey, ignored -> new byte[] {1, 2});
+ manager.getPage(indexKey, ignored -> new byte[] {3, 4});
+ assertThat(manager.getPageIfPresent(dataKey).getHeapMemory()).containsExactly(1, 2);
+ assertThat(manager.getPageSliceIfPresent(indexKey).copyBytes()).containsExactly(3, 4);
+ manager.invalidFile(file);
+ assertThat(manager.getPageIfPresent(dataKey)).isNull();
+ assertThat(manager.getPageSliceIfPresent(indexKey)).isNull();
+ }
+ }
+
+ @Test
+ void testEvictionRemovesEmptyFiles() throws Exception {
+ try (CacheManager manager = new CacheManager(MemorySize.ofBytes(16), 0)) {
+ for (int i = 0; i < 100; i++) {
+ manager.getPage(key(new Path("file-" + i)), ignored -> new byte[8]);
+ // Each file has one page. No file index may outlive its cached page.
+ assertThat(manager.cachedFileCount()).isEqualTo(manager.dataCache().asMap().size());
+ }
+ manager.close();
+ assertThat(manager.cachedFileCount()).isZero();
+ }
+ }
+
+ @Test
+ void testRemovingMiddleAndTailPagesKeepsFileIndexAccurate() throws Exception {
+ List<CacheKey> invalidated = new ArrayList<>();
+ Path file = new Path("linked-pages");
+ CacheKey tail = CacheKey.forPosition(file, 0, 2, false);
+ CacheKey middle = CacheKey.forPosition(file, 2, 2, false);
+ CacheKey head = CacheKey.forPosition(file, 4, 2, false);
+ CacheKey other = key(new Path("other-file"));
+ try (CacheManager manager =
+ new CacheManager(MemorySize.ofKibiBytes(64), 0) {
+ @Override
+ protected void invalidPage(CacheKey key, Cache.CacheValue expected) {
+ invalidated.add(key);
+ super.invalidPage(key, expected);
+ }
+ }) {
+ for (CacheKey key : new CacheKey[] {tail, middle, head, other}) {
+ manager.getPage(key, ignored -> new byte[2]);
+ }
+ manager.invalidPage(middle);
+ manager.invalidPage(tail);
+ manager.invalidFile(file);
+ assertThat(invalidated).containsExactly(head);
+ assertThat(manager.cachedFileCount()).isOne();
+ assertThat(manager.contains(other)).isTrue();
+ manager.invalidFile(new Path("other-file"));
+ assertThat(manager.cachedFileCount()).isZero();
+ }
+ }
+
+ @Test
+ void testFailedLoadCanRetryWithoutLeavingFileIndex() throws Exception {
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0)) {
+ Path file = new Path("retry-file");
+ CacheKey key = key(file);
+ assertThatThrownBy(
+ () ->
+ manager.getPage(
+ key,
+ ignored -> {
+ throw new IOException("read failed");
+ }))
+ .hasRootCauseMessage("read failed");
+ assertThat(manager.cachedFileCount()).isZero();
+ assertThat(manager.getPage(key, ignored -> new byte[] {7, 8}).getHeapMemory())
+ .containsExactly(7, 8);
+ manager.invalidFile(file);
+ assertThat(manager.cachedFileCount()).isZero();
+ assertThat(manager.contains(key)).isFalse();
+ }
+ }
+
+ @Test
+ void testOldInvalidationDoesNotRemoveReplacement() throws Exception {
+ Path file = new Path("shared-file");
+ CacheKey key = key(file);
+ CountDownLatch detached = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ AtomicBoolean pause = new AtomicBoolean(true);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (CacheManager manager =
+ new CacheManager(MemorySize.ofKibiBytes(64), 0) {
+ @Override
+ protected void invalidPage(CacheKey k, Cache.CacheValue expected) {
+ if (pause.compareAndSet(true, false)) {
+ detached.countDown();
+ await(resume);
+ }
+ super.invalidPage(k, expected);
+ }
+ }) {
+ manager.getPage(key, ignored -> new byte[] {1, 2});
+ Future<?> invalidation =
+ executor.submit(
+ () -> {
+ manager.invalidFile(file);
+ return null;
+ });
+ try {
+ assertThat(detached.await(10, TimeUnit.SECONDS)).isTrue();
+ manager.invalidPage(key);
+ MemorySegment replacement = manager.getPage(key, ignored -> new byte[] {3, 4});
+ resume.countDown();
+ invalidation.get(10, TimeUnit.SECONDS);
+
+ assertThat(
+ manager.getPage(
+ key,
+ ignored -> {
+ throw new AssertionError("Replacement was invalidated");
+ }))
+ .isSameAs(replacement);
+ assertThat(manager.cachedFileCount()).isOne();
+ manager.invalidFile(file);
+ assertThat(manager.contains(key)).isFalse();
+ assertThat(manager.cachedFileCount()).isZero();
+ } finally {
+ resume.countDown();
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testConcurrentLoadMayRepopulateInvalidatedFile() throws Exception {
+ Path file = new Path("loading-file");
+ CacheKey key = key(file);
+ CountDownLatch loading = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0)) {
+ Future<MemorySegment> load =
+ executor.submit(
+ () ->
+ manager.getPage(
+ key,
+ ignored -> {
+ loading.countDown();
+ await(resume);
+ return new byte[] {5, 6};
+ }));
+ try {
+ assertThat(loading.await(10, TimeUnit.SECONDS)).isTrue();
+ manager.invalidFile(file);
+ resume.countDown();
+ assertThat(load.get(10, TimeUnit.SECONDS).getHeapMemory()).containsExactly(5, 6);
+ assertThat(manager.contains(key)).isTrue();
+ manager.invalidFile(file);
+ assertThat(manager.cachedFileCount()).isZero();
+ } finally {
+ resume.countDown();
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testDifferentFilesLoadConcurrently() throws Exception {
+ CyclicBarrier loading = new CyclicBarrier(2);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0)) {
+ CacheReader loader =
+ ignored -> {
+ try {
+ loading.await(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ return new byte[] {9, 10};
+ };
+ Future<MemorySegment> first =
+ executor.submit(() -> manager.getPage(key(new Path("first")), loader));
+ Future<MemorySegment> second =
+ executor.submit(() -> manager.getPage(key(new Path("second")), loader));
+ assertThat(first.get(15, TimeUnit.SECONDS).getHeapMemory()).containsExactly(9, 10);
+ assertThat(second.get(15, TimeUnit.SECONDS).getHeapMemory()).containsExactly(9, 10);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private static CacheKey key(Path file) {
+ return CacheKey.forPosition(file, 0, 2, false);
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for cache operation");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerTest.java b/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerTest.java
index 662978a..1fe1a81 100644
--- a/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/io/cache/CacheManagerTest.java
@@ -81,9 +81,9 @@
CacheManager cacheManager =
new CacheManager(MemorySize.ofKibiBytes(64), 0) {
@Override
- public void invalidPage(CacheKey key) {
+ protected void invalidPage(CacheKey key, Cache.CacheValue expected) {
invalidatedPages.incrementAndGet();
- super.invalidPage(key);
+ super.invalidPage(key, expected);
}
};
BlockCache blockCache =
@@ -100,6 +100,7 @@
blockCache.close();
assertThat(invalidatedPages).hasValue(hotPages);
+ assertThat(cacheManager.cachedFileCount()).isZero();
}
@Test
diff --git a/paimon-common/src/test/java/org/apache/paimon/io/cache/CaffeineCacheTest.java b/paimon-common/src/test/java/org/apache/paimon/io/cache/CaffeineCacheTest.java
new file mode 100644
index 0000000..6464efe
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/io/cache/CaffeineCacheTest.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.io.cache;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.memory.MemorySlice;
+
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the Caffeine adapter's non-loading lookup contract. */
+class CaffeineCacheTest {
+
+ @Test
+ void testPresentLookupRefreshesAccessExpiration() {
+ AtomicLong time = new AtomicLong();
+ Cache cache =
+ new CaffeineCache(
+ Caffeine.newBuilder()
+ .expireAfterAccess(10, TimeUnit.SECONDS)
+ .ticker(time::get)
+ .executor(Runnable::run)
+ .build());
+ CacheKey key = CacheKey.forPosition(new Path("file"), 0, 2, false);
+ Cache.CacheValue value =
+ new Cache.CacheValue(MemorySlice.wrap(new byte[] {1, 2}), ignored -> {});
+ cache.put(key, value);
+
+ time.set(TimeUnit.SECONDS.toNanos(9));
+ assertThat(cache.getIfPresent(key)).isSameAs(value);
+ time.set(TimeUnit.SECONDS.toNanos(18));
+ assertThat(cache.getIfPresent(key)).isSameAs(value);
+ time.set(TimeUnit.SECONDS.toNanos(29));
+ assertThat(cache.getIfPresent(key)).isNull();
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/SortLookupStoreCloseTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/SortLookupStoreCloseTest.java
index 06649f6..37e3587 100644
--- a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/SortLookupStoreCloseTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/SortLookupStoreCloseTest.java
@@ -24,6 +24,7 @@
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.cache.Cache;
import org.apache.paimon.io.cache.CacheKey;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.memory.MemorySliceOutput;
@@ -103,7 +104,7 @@
for (int key = 0; key < 400; key += 40) {
lookup(reader, key * 2);
}
- int cachedPages = cacheManager.pagesTaken.get();
+ int cachedPages = cacheManager.dataCache().asMap().size();
assertThat(cachedPages).isGreaterThan(1);
cacheManager.failFrom(1);
@@ -164,7 +165,6 @@
/** A cache manager whose page invalidation can be made to fail. */
private static class ThrowingCacheManager extends CacheManager {
- private final AtomicInteger pagesTaken = new AtomicInteger();
private final AtomicInteger invalidated = new AtomicInteger();
private volatile int failFromCall;
@@ -178,19 +178,19 @@
}
@Override
- public org.apache.paimon.memory.MemorySegment getPage(
- CacheKey key,
- org.apache.paimon.io.cache.CacheReader reader,
- org.apache.paimon.io.cache.CacheCallback callback) {
- pagesTaken.incrementAndGet();
- return super.getPage(key, reader, callback);
+ public void invalidPage(CacheKey key) {
+ super.invalidPage(key);
+ invalidated();
}
@Override
- public void invalidPage(CacheKey key) {
- int call = invalidated.getAndIncrement();
- super.invalidPage(key);
- if (call >= failFromCall) {
+ protected void invalidPage(CacheKey key, Cache.CacheValue expected) {
+ super.invalidPage(key, expected);
+ invalidated();
+ }
+
+ private void invalidated() {
+ if (invalidated.getAndIncrement() >= failFromCall) {
throw new RuntimeException("invalidPage failed");
}
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/sst/BlockCacheTest.java b/paimon-common/src/test/java/org/apache/paimon/sst/BlockCacheTest.java
new file mode 100644
index 0000000..e2221d0
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/sst/BlockCacheTest.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.sst;
+
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.cache.CacheKey;
+import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.options.MemorySize;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for shared block ownership and file invalidation. */
+class BlockCacheTest {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ @Timeout(30)
+ void testConcurrentLoadAfterFastLookupMissIsReused(boolean slice) throws Exception {
+ Path path = new Path("shared-file");
+ CountDownLatch missed = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ AtomicBoolean pause = new AtomicBoolean(true);
+ AtomicInteger loads = new AtomicInteger();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (CacheManager manager =
+ new CacheManager(MemorySize.ofKibiBytes(64), 0) {
+ @Override
+ public MemorySlice getPageSliceIfPresent(CacheKey key) {
+ MemorySlice cached = super.getPageSliceIfPresent(key);
+ if (cached == null && pause.compareAndSet(true, false)) {
+ missed.countDown();
+ try {
+ if (!resume.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out after cache miss");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ return cached;
+ }
+ };
+ BlockCache first = new BlockCache(path, input(), manager);
+ BlockCache second = new BlockCache(path, input(), manager)) {
+ Future<MemorySlice> pending = executor.submit(() -> read(first, slice, loads));
+ try {
+ assertThat(missed.await(10, TimeUnit.SECONDS)).isTrue();
+ MemorySlice loaded = read(second, slice, loads);
+ resume.countDown();
+ MemorySlice reused = pending.get(10, TimeUnit.SECONDS);
+ assertThat(reused.copyBytes()).containsExactly(1, 2, 3, 4);
+ assertThat(reused.segment()).isSameAs(loaded.segment());
+ assertThat(loads).hasValue(1);
+ } finally {
+ resume.countDown();
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testSharedReaderReloadsInvalidatedBlock() throws Exception {
+ Path path = new Path("shared-file");
+ AtomicInteger loads = new AtomicInteger();
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0);
+ BlockCache first = new BlockCache(path, input(), manager);
+ BlockCache second = new BlockCache(path, input(), manager)) {
+ MemorySegment original =
+ first.getBlock(
+ 0,
+ 4,
+ bytes -> {
+ loads.incrementAndGet();
+ return bytes;
+ },
+ false);
+ assertThat(
+ second.getBlock(
+ 0,
+ 4,
+ bytes -> {
+ loads.incrementAndGet();
+ return bytes;
+ },
+ false))
+ .isSameAs(original);
+ manager.invalidPage(CacheKey.forPosition(path, 0, 4, false));
+
+ MemorySegment reloaded =
+ second.getBlock(
+ 0,
+ 4,
+ bytes -> {
+ loads.incrementAndGet();
+ return bytes;
+ },
+ false);
+ assertThat(reloaded.getHeapMemory()).containsExactly(1, 2, 3, 4);
+ assertThat(loads).hasValue(2);
+ }
+ }
+
+ @Test
+ void testCloseInvalidatesAllReadersPagesForFile() throws Exception {
+ Path path = new Path("shared-file");
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0.5);
+ BlockCache first = new BlockCache(path, input(), manager);
+ BlockCache second = new BlockCache(path, input(), manager);
+ BlockCache other = new BlockCache(new Path("other-file"), input(), manager)) {
+ first.getBlock(0, 4, bytes -> bytes, false);
+ second.getBlock(4, 4, bytes -> bytes, true);
+ MemorySegment otherBlock = other.getBlock(0, 4, bytes -> bytes, false);
+
+ first.close();
+
+ assertThat(manager.contains(CacheKey.forPosition(path, 0, 4, false))).isFalse();
+ assertThat(manager.contains(CacheKey.forPosition(path, 4, 4, true))).isFalse();
+ assertThat(
+ other.getBlock(
+ 0,
+ 4,
+ bytes -> {
+ throw new AssertionError("Unexpected reload");
+ },
+ false))
+ .isSameAs(otherBlock);
+ }
+ }
+
+ private static MemorySlice read(BlockCache cache, boolean slice, AtomicInteger loads) {
+ if (slice) {
+ return cache.getBlockSlice(
+ 0,
+ 4,
+ bytes -> {
+ loads.incrementAndGet();
+ return MemorySlice.wrap(bytes);
+ },
+ false);
+ }
+ return MemorySlice.wrap(
+ cache.getBlock(
+ 0,
+ 4,
+ bytes -> {
+ loads.incrementAndGet();
+ return bytes;
+ },
+ false));
+ }
+
+ private static ByteArraySeekableStream input() {
+ return new ByteArraySeekableStream(new byte[] {1, 2, 3, 4, 5, 6, 7, 8});
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/sst/SstFileReaderCacheTest.java b/paimon-common/src/test/java/org/apache/paimon/sst/SstFileReaderCacheTest.java
new file mode 100644
index 0000000..86e2298
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/sst/SstFileReaderCacheTest.java
@@ -0,0 +1,498 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.sst;
+
+import org.apache.paimon.compression.BlockCompressionFactory;
+import org.apache.paimon.compression.BlockCompressionType;
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.io.cache.CacheKey;
+import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.utils.BloomFilter;
+import org.apache.paimon.utils.FileBasedBloomFilter;
+import org.apache.paimon.utils.MurmurHashUtils;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Comparator;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests the combined block/trailer cache without changing the SST format. */
+class SstFileReaderCacheTest {
+
+ @ParameterizedTest
+ @CsvSource({"NONE,false", "NONE,true", "LZ4,false", "LZ4,true"})
+ void testRejectedBloomUsesCachedData(String compression, boolean offHeap) throws Exception {
+ BloomFixture fixture = new BloomFixture(compression, 16 * 1024);
+ CountingInput input = new CountingInput(fixture.bytes);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofMebiBytes(1), 0.5)
+ : new CacheManager(MemorySize.ofMebiBytes(1), 0.5);
+ SstFileReader reader = fixture.reader(input, manager)) {
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ assertThat(manager.contains(fixture.bloomKey())).isFalse();
+ assertThat(manager.dataCache().asMap()).hasSize(1);
+
+ input.reads = 0;
+ assertThat(reader.lookup(new byte[] {4})).containsExactly(fixture.value);
+ assertThat(reader.lookup(new byte[] {1})).isNull();
+ assertThat(input.reads).isZero();
+
+ // Another reader can admit the filter later. Use it without loading the data block.
+ manager.invalidPage(fixture.dataKey());
+ manager.getPage(
+ fixture.bloomKey(),
+ ignored ->
+ Arrays.copyOfRange(
+ fixture.bytes,
+ (int) fixture.bloom.offset(),
+ (int) fixture.bloom.offset() + fixture.bloom.size()));
+ assertThat(reader.lookup(new byte[] {1})).isNull();
+ assertThat(input.reads).isZero();
+ assertThat(manager.dataCache().asMap()).isEmpty();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testRejectedBloomCanRecoverInTheSameReader(boolean offHeap) throws Exception {
+ BloomFixture fixture = new BloomFixture("NONE", 16 * 1024);
+ CountingInput input = new CountingInput(fixture.bytes);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofMebiBytes(1), 0.5)
+ : new CacheManager(MemorySize.ofMebiBytes(1), 0.5);
+ SstFileReader reader = fixture.reader(input, manager)) {
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ assertThat(manager.contains(fixture.bloomKey())).isFalse();
+ fixture.rejectBloom = false;
+ input.bytesRead = 0;
+ for (int i = 0; i < 64 && !manager.contains(fixture.bloomKey()); i++) {
+ manager.invalidPage(fixture.dataKey());
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ }
+ assertThat(manager.contains(fixture.bloomKey())).isTrue();
+ assertThat(input.bytesRead)
+ .isLessThanOrEqualTo(2 * fixture.bloom.size() + 2 * fixture.dataSize);
+ manager.invalidPage(fixture.dataKey());
+ input.reads = 0;
+ assertThat(reader.lookup(new byte[] {1})).isNull();
+ assertThat(input.reads).isZero();
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "NONE,false,16384",
+ "NONE,true,16384",
+ "NONE,false,256",
+ "NONE,true,256",
+ "LZ4,false,16384",
+ "LZ4,true,16384"
+ })
+ void testRejectedBloomReadCost(String compression, boolean offHeap, int bloomSize)
+ throws Exception {
+ BloomFixture fixture = new BloomFixture(compression, bloomSize);
+ CountingInput input = new CountingInput(fixture.bytes);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofMebiBytes(1), 0.5)
+ : new CacheManager(MemorySize.ofMebiBytes(1), 0.5);
+ SstFileReader reader = fixture.reader(input, manager)) {
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ manager.invalidPage(fixture.dataKey());
+ input.reads = 0;
+ input.bytesRead = 0;
+ assertThat(reader.lookup(new byte[] {1})).isNull();
+ assertThat(input.reads).isOne();
+ boolean cheaperBloom = bloomSize < fixture.dataSize || "LZ4".equals(compression);
+ assertThat(input.bytesRead).isEqualTo(cheaperBloom ? bloomSize : fixture.dataSize);
+
+ input.reads = 0;
+ assertThat(reader.lookup(new byte[] {127})).isNull();
+ assertThat(input.reads).isZero();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCompressedIndexKeepsUncacheableBloomOnReopen(boolean offHeap) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ PositionOutputStream out = output(bytes);
+ SstFileWriter writer =
+ new SstFileWriter(
+ out, 256, null, BlockCompressionFactory.create(BlockCompressionType.LZ4));
+ byte[] value = new byte[64];
+ for (int i = 0; i < 1000; i++) {
+ MemorySegment key = MemorySegment.allocateHeapMemory(32);
+ key.putInt(0, 2 * i);
+ writer.put(key.getHeapMemory(), value);
+ }
+ writer.flush();
+ BloomFilter bloom = new BloomFilter(1000, 128 * 1024);
+ bloom.setMemorySegment(MemorySegment.allocateHeapMemory(128 * 1024), 0);
+ for (int i = 0; i < 1000; i++) {
+ MemorySegment key = MemorySegment.allocateHeapMemory(32);
+ key.putInt(0, 2 * i);
+ bloom.addHash(MurmurHashUtils.hashBytes(key.getHeapMemory()));
+ }
+ BloomFilterHandle bloomHandle = bloom.write(out);
+ BlockHandle index = writer.writeIndexBlock();
+ byte[] file = bytes.toByteArray();
+ assertThat(file[(int) index.offset() + index.size()]).isNotZero();
+ MemorySegment missing = MemorySegment.allocateHeapMemory(32);
+ missing.putInt(0, 1);
+ assertThat(bloom.testHash(MurmurHashUtils.hashBytes(missing.getHeapMemory()))).isFalse();
+ Path path = new Path("compressed-index-bloom");
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofKibiBytes(64), 0.5)
+ : new CacheManager(MemorySize.ofKibiBytes(64), 0.5)) {
+ // Keep both readers open so the second borrows the already decoded index page.
+ CountingInput first = new CountingInput(file);
+ CountingInput second = new CountingInput(file);
+ try (SstFileReader reader =
+ new SstFileReader(
+ Comparator.comparingInt(slice -> slice.readInt(0)),
+ new BlockCache(path, first, manager),
+ index,
+ FileBasedBloomFilter.create(
+ first, path, manager, bloomHandle));
+ SstFileReader reopened =
+ new SstFileReader(
+ Comparator.comparingInt(slice -> slice.readInt(0)),
+ new BlockCache(path, second, manager),
+ index,
+ FileBasedBloomFilter.create(
+ second, path, manager, bloomHandle))) {
+ assertThat(first.reads).isOne();
+ assertThat(second.reads).isZero();
+ first.reads = 0;
+ first.bytesRead = 0;
+ assertThat(reader.lookup(missing.getHeapMemory())).isNull();
+ assertThat(reopened.lookup(missing.getHeapMemory())).isNull();
+ assertThat(first.reads).isOne();
+ assertThat(second.reads).isOne();
+ assertThat(first.bytesRead).isEqualTo(bloomHandle.size());
+ assertThat(second.bytesRead).isEqualTo(bloomHandle.size());
+ assertThat(manager.dataCache().asMap()).isEmpty();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "NONE,false,-1,0",
+ "NONE,false,0,0",
+ "NONE,false,1,0",
+ "NONE,true,1,0",
+ "LZ4,false,1,0",
+ "LZ4,true,1,0",
+ "LZ4,false,4096,0",
+ "LZ4,true,4096,0",
+ "NONE,false,1,1",
+ "NONE,true,1,1",
+ "NONE,false,1,2"
+ })
+ void testUncacheableBloomReadCost(
+ String compression, boolean offHeap, int sizeDelta, int cacheBudget) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ PositionOutputStream out = output(bytes);
+ SstFileWriter writer =
+ new SstFileWriter(
+ out,
+ 64 * 1024,
+ null,
+ BlockCompressionFactory.create(BlockCompressionType.valueOf(compression)));
+ byte[] value = new byte[64];
+ Arrays.fill(value, (byte) 42);
+ for (int i = 0; i < 16; i++) {
+ writer.put(new byte[] {(byte) (2 * i)}, value);
+ }
+ writer.flush();
+ int dataBlockSize = bytes.size();
+ BloomFilter bloom = new BloomFilter(1000, dataBlockSize + sizeDelta);
+ bloom.setMemorySegment(MemorySegment.allocateHeapMemory(dataBlockSize + sizeDelta), 0);
+ for (int i = 0; i < 16; i++) {
+ bloom.addHash(MurmurHashUtils.hashBytes(new byte[] {(byte) (2 * i)}));
+ }
+ byte[] missing = new byte[] {1};
+ while (missing[0] < 31 && bloom.testHash(MurmurHashUtils.hashBytes(missing))) {
+ missing[0] += 2;
+ }
+ assertThat(bloom.testHash(MurmurHashUtils.hashBytes(missing))).isFalse();
+ BloomFilterHandle bloomHandle = bloom.write(out);
+ BlockHandle index = writer.writeIndexBlock();
+ CountingInput input = new CountingInput(bytes.toByteArray());
+ Path path = new Path("bloom-read-cost");
+ boolean cached = cacheBudget == 1;
+ // Budget 2 admits the data block but cannot fit the larger Bloom in the index pool.
+ MemorySize capacity =
+ cached
+ ? MemorySize.ofMebiBytes(1)
+ : MemorySize.ofBytes(cacheBudget == 2 ? 2L * (bloomHandle.size() - 1) : 0);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(capacity, 0.5)
+ : new CacheManager(capacity, 0.5);
+ SstFileReader reader =
+ new SstFileReader(
+ Comparator.comparingInt(slice -> slice.readByte(0)),
+ new BlockCache(path, input, manager),
+ index,
+ FileBasedBloomFilter.create(input, path, manager, bloomHandle))) {
+ input.reads = 0;
+ input.bytesRead = 0;
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(value);
+ boolean probeBloom = cached || sizeDelta < 0;
+ assertThat(input.reads).isEqualTo(probeBloom ? 2 : 1);
+ assertThat(input.bytesRead)
+ .isEqualTo(dataBlockSize + (probeBloom ? bloomHandle.size() : 0));
+
+ // Force the absent-key lookup to choose between reading Bloom and reading data.
+ manager.invalidPage(CacheKey.forPosition(path, 0, dataBlockSize, false));
+ input.reads = 0;
+ input.bytesRead = 0;
+ assertThat(reader.lookup(missing)).isNull();
+ assertThat(input.reads).isEqualTo(cached ? 0 : 1);
+ // A compressed block can be much larger to decode than its serialized size.
+ boolean probeOnMissing = probeBloom || "LZ4".equals(compression);
+ assertThat(input.bytesRead)
+ .isEqualTo(cached ? 0 : probeOnMissing ? bloomHandle.size() : dataBlockSize);
+ input.reads = 0;
+ assertThat(reader.lookup(new byte[] {127})).isNull();
+ assertThat(input.reads).isZero();
+ if (cacheBudget == 0) {
+ assertThat(manager.dataCache().asMap()).isEmpty();
+ assertThat(manager.indexCache().asMap()).isEmpty();
+ } else if (cacheBudget == 2) {
+ assertThat(manager.dataCache().asMap()).hasSize(1);
+ assertThat(
+ manager.contains(
+ CacheKey.forPosition(
+ path,
+ bloomHandle.offset(),
+ bloomHandle.size(),
+ true)))
+ .isFalse();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"NONE,false", "NONE,true", "LZ4,false", "LZ4,true"})
+ void testOneReadAndCacheEntryPerBlock(String compression, boolean offHeap) throws Exception {
+ Fixture fixture = new Fixture(compression);
+ CountingInput input = new CountingInput(fixture.bytes);
+ try (CacheManager manager =
+ offHeap
+ ? CacheManager.createOffHeap(MemorySize.ofMebiBytes(1), 0.5)
+ : new CacheManager(MemorySize.ofMebiBytes(1), 0.5);
+ SstFileReader reader = fixture.reader(input, manager)) {
+ assertThat(input.reads).isOne();
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ assertThat(input.reads).isEqualTo(2);
+ assertThat(manager.dataCache().asMap()).hasSize(1);
+ assertThat(manager.indexCache().asMap()).hasSize(1);
+ assertThat(reader.lookup(new byte[] {3})).containsExactly(fixture.value);
+ assertThat(input.reads).isEqualTo(2);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"NONE", "LZ4"})
+ void testCorruptBlockIsRejectedAndCanBeRetried(String compression) throws Exception {
+ Fixture fixture = new Fixture(compression);
+ CountingInput input = new CountingInput(fixture.bytes);
+ try (CacheManager manager = new CacheManager(MemorySize.ofMebiBytes(1), 0.5);
+ SstFileReader reader = fixture.reader(input, manager)) {
+ fixture.bytes[0] ^= 1;
+ assertThatThrownBy(() -> reader.lookup(new byte[] {2}))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Expected CRC32C");
+ assertThat(manager.dataCache().asMap()).isEmpty();
+ fixture.bytes[0] ^= 1;
+ assertThat(reader.lookup(new byte[] {2})).containsExactly(fixture.value);
+ assertThat(manager.dataCache().asMap()).hasSize(1);
+ }
+ }
+
+ private static class BloomFixture {
+ private final Path path = new Path("rejected-bloom");
+ private final byte[] value = new byte[64];
+ private final byte[] bytes;
+ private final int dataSize;
+ private final BlockHandle index;
+ private final BloomFilterHandle bloom;
+ private boolean rejectBloom = true;
+
+ private BloomFixture(String compression, int bloomSize) throws IOException {
+ Arrays.fill(value, (byte) 42);
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ PositionOutputStream out = output(bytes);
+ SstFileWriter writer =
+ new SstFileWriter(
+ out,
+ 64 * 1024,
+ null,
+ BlockCompressionFactory.create(
+ BlockCompressionType.valueOf(compression)));
+ BloomFilter filter = new BloomFilter(16, bloomSize);
+ filter.setMemorySegment(MemorySegment.allocateHeapMemory(bloomSize), 0);
+ for (int i = 0; i < 16; i++) {
+ byte[] key = new byte[] {(byte) (2 * i)};
+ writer.put(key, value);
+ filter.addHash(MurmurHashUtils.hashBytes(key));
+ }
+ writer.flush();
+ this.dataSize = bytes.size();
+ this.bloom = filter.write(out);
+ this.index = writer.writeIndexBlock();
+ this.bytes = bytes.toByteArray();
+ assertThat(filter.testHash(MurmurHashUtils.hashBytes(new byte[] {1}))).isFalse();
+ assertThat(this.bytes[dataSize - BlockTrailer.ENCODED_LENGTH] == 0)
+ .isEqualTo("NONE".equals(compression));
+ }
+
+ private CacheKey bloomKey() {
+ return CacheKey.forPosition(path, bloom.offset(), bloom.size(), true);
+ }
+
+ private CacheKey dataKey() {
+ return CacheKey.forPosition(path, 0, dataSize, false);
+ }
+
+ private SstFileReader reader(CountingInput input, CacheManager manager) {
+ FileBasedBloomFilter filter =
+ new FileBasedBloomFilter(
+ input, path, manager, 16, bloom.offset(), bloom.size()) {
+ @Override
+ public boolean testHash(int hash) {
+ boolean result = super.testHash(hash);
+ // Admission is optional even when the page fits. Force this legal
+ // outcome without depending on Caffeine's frequency sketch or timing.
+ if (rejectBloom) {
+ manager.invalidPage(bloomKey());
+ }
+ return result;
+ }
+ };
+ assertThat(filter.isCacheable()).isTrue();
+ return new SstFileReader(
+ Comparator.comparingInt(slice -> slice.readByte(0)),
+ new BlockCache(path, input, manager),
+ index,
+ filter);
+ }
+ }
+
+ private static class Fixture {
+ private final byte[] value = new byte[2048];
+ private final byte[] bytes;
+ private final BlockHandle index;
+
+ private Fixture(String compression) throws IOException {
+ Arrays.fill(value, (byte) 42);
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ PositionOutputStream out = output(bytes);
+ SstFileWriter writer =
+ new SstFileWriter(
+ out,
+ 64 * 1024,
+ null,
+ BlockCompressionFactory.create(
+ BlockCompressionType.valueOf(compression)));
+ for (int i = 1; i <= 3; i++) {
+ writer.put(new byte[] {(byte) i}, value);
+ }
+ writer.flush();
+ this.index = writer.writeIndexBlock();
+ this.bytes = bytes.toByteArray();
+ // Verify the fixture really exercises compressed data, not the writer's raw fallback.
+ int compressionId = this.bytes[(int) index.offset() - BlockTrailer.ENCODED_LENGTH];
+ assertThat(compressionId == 0).isEqualTo("NONE".equals(compression));
+ }
+
+ private SstFileReader reader(CountingInput input, CacheManager manager) {
+ return new SstFileReader(
+ Comparator.comparingInt(slice -> slice.readByte(0)),
+ new BlockCache(new Path("sst-file"), input, manager),
+ index,
+ null);
+ }
+ }
+
+ private static PositionOutputStream output(ByteArrayOutputStream bytes) {
+ return new PositionOutputStream() {
+ @Override
+ public void close() {}
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public long getPos() {
+ return bytes.size();
+ }
+
+ @Override
+ public void write(int b) {
+ bytes.write(b);
+ }
+
+ @Override
+ public void write(byte[] b) {
+ bytes.write(b, 0, b.length);
+ }
+
+ @Override
+ public void write(byte[] b, int offset, int length) {
+ bytes.write(b, offset, length);
+ }
+ };
+ }
+
+ private static class CountingInput extends ByteArraySeekableStream {
+ private int reads;
+ private int bytesRead;
+
+ private CountingInput(byte[] bytes) {
+ super(bytes);
+ }
+
+ @Override
+ public int read(byte[] b, int offset, int length) throws IOException {
+ reads++;
+ int read = super.read(b, offset, length);
+ bytesRead += Math.max(read, 0);
+ return read;
+ }
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java
index e4c3058..379b807 100644
--- a/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java
@@ -22,6 +22,8 @@
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
@@ -155,6 +157,34 @@
i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isFalse());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testProbeWithOffsetAndBorrowedSegment(boolean offHeap) {
+ MemorySegment original =
+ offHeap
+ ? MemorySegment.allocateOffHeapMemory(75)
+ : MemorySegment.allocateHeapMemory(75);
+ for (int i = 0; i < 11; i++) {
+ original.put(i, (byte) 0xff);
+ }
+ BloomFilter filter = new BloomFilter(10, 64);
+ filter.setMemorySegment(original, 11);
+ filter.addHash(42);
+ Assertions.assertThat(filter.testHash(42)).isTrue();
+ Assertions.assertThat(filter.testHash(43)).isFalse();
+
+ byte[] bytes = new byte[64];
+ original.get(11, bytes);
+ MemorySegment borrowed =
+ offHeap
+ ? MemorySegment.allocateOffHeapMemory(64)
+ : MemorySegment.allocateHeapMemory(64);
+ borrowed.put(0, bytes);
+ Assertions.assertThat(filter.testHash(42, borrowed)).isTrue();
+ Assertions.assertThat(filter.testHash(43, borrowed)).isFalse();
+ Assertions.assertThat(filter.getMemorySegment()).isSameAs(original);
+ }
+
private static int numHashFunctions(long expectedEntries, double fpp) {
return buildFilter(BloomFilter.fixedBuilder(expectedEntries, fpp)).numHashFunctions();
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java
index 9d19d81..f25e51f 100644
--- a/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.utils;
+import org.apache.paimon.fs.ByteArraySeekableStream;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.cache.CacheManager;
@@ -27,6 +28,8 @@
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.io.File;
import java.io.IOException;
@@ -35,12 +38,76 @@
import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
/** Test for {@link FileBasedBloomFilter}. */
public class FileBasedBloomFilterTest {
@TempDir Path tempDir;
+ @ParameterizedTest
+ @CsvSource({
+ "64,0,64,true",
+ "64,0.5,32,true",
+ "64,0.5,33,false",
+ "64,0.75,32,true",
+ "0,0,1,false"
+ })
+ void testFilterFitsItsActualIndexPool(long capacity, double indexRatio, int size, boolean fits)
+ throws Exception {
+ byte[] bytes = new byte[size];
+ Arrays.fill(bytes, (byte) 0xff);
+ AtomicInteger reads = new AtomicInteger();
+ try (CacheManager manager = new CacheManager(MemorySize.ofBytes(capacity), indexRatio);
+ FileBasedBloomFilter filter =
+ new FileBasedBloomFilter(
+ countingInput(bytes, reads),
+ new org.apache.paimon.fs.Path("pool-boundary"),
+ manager,
+ 10,
+ 0,
+ size)) {
+ Assertions.assertThat(filter.isCacheable()).isEqualTo(fits);
+ Assertions.assertThat(filter.testHash(123)).isTrue();
+ Assertions.assertThat(filter.testHash(123)).isTrue();
+ Assertions.assertThat(reads).hasValue(fits ? 1 : 2);
+ }
+ }
+
+ @Test
+ void testResidentProbeDoesNotReadOrRetainAnEvictedFilter() throws Exception {
+ MemorySegment segment = MemorySegment.allocateHeapMemory(64);
+ BloomFilter bloom = new BloomFilter(10, segment.size());
+ bloom.setMemorySegment(segment, 0);
+ bloom.addHash(123);
+ Assertions.assertThat(bloom.testHash(124)).isFalse();
+ AtomicInteger reads = new AtomicInteger();
+ org.apache.paimon.fs.Path path = new org.apache.paimon.fs.Path("resident-filter");
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0.5);
+ FileBasedBloomFilter filter =
+ new FileBasedBloomFilter(
+ countingInput(segment.getHeapMemory(), reads),
+ path,
+ manager,
+ 10,
+ 0,
+ segment.size())) {
+ Assertions.assertThat(filter.testHashIfPresent(123)).isNull();
+ Assertions.assertThat(reads).hasValue(0);
+ Assertions.assertThat(filter.testHash(123)).isTrue();
+ Assertions.assertThat(filter.testHashIfPresent(123)).isTrue();
+ Assertions.assertThat(filter.testHashIfPresent(124)).isFalse();
+ Assertions.assertThat(reads).hasValue(1);
+
+ manager.invalidFile(path);
+ Assertions.assertThat(filter.testHashIfPresent(123)).isNull();
+ Assertions.assertThat(reads).hasValue(1);
+ Assertions.assertThat(filter.bloomFilter().getMemorySegment()).isNull();
+ Assertions.assertThat(filter.testHash(123)).isTrue();
+ Assertions.assertThat(reads).hasValue(2);
+ }
+ }
+
@Test
public void testProbe() throws IOException {
MemorySegment segment = MemorySegment.wrap(new byte[1000]);
@@ -65,6 +132,40 @@
Assertions.assertThat(filter.bloomFilter().getMemorySegment()).isNull();
}
+ @Test
+ void testSharedFilterReloadsAfterFileInvalidation() throws Exception {
+ byte[] bytes = new byte[64];
+ Arrays.fill(bytes, (byte) 0xff);
+ AtomicInteger reads = new AtomicInteger();
+ org.apache.paimon.fs.Path path = new org.apache.paimon.fs.Path("shared-filter");
+ try (CacheManager manager = new CacheManager(MemorySize.ofKibiBytes(64), 0.5)) {
+ FileBasedBloomFilter first =
+ new FileBasedBloomFilter(
+ countingInput(bytes, reads), path, manager, 10, 0, bytes.length);
+ FileBasedBloomFilter second =
+ new FileBasedBloomFilter(
+ countingInput(bytes, reads), path, manager, 10, 0, bytes.length);
+ Assertions.assertThat(first.testHash(123)).isTrue();
+ Assertions.assertThat(second.testHash(123)).isTrue();
+ Assertions.assertThat(reads).hasValue(1);
+ manager.invalidFile(path);
+ Assertions.assertThat(second.testHash(123)).isTrue();
+ Assertions.assertThat(reads).hasValue(2);
+ Assertions.assertThat(first.bloomFilter().getMemorySegment()).isNull();
+ Assertions.assertThat(second.bloomFilter().getMemorySegment()).isNull();
+ }
+ }
+
+ private static ByteArraySeekableStream countingInput(byte[] bytes, AtomicInteger reads) {
+ return new ByteArraySeekableStream(bytes) {
+ @Override
+ public int read(byte[] b, int offset, int length) throws IOException {
+ reads.incrementAndGet();
+ return super.read(b, offset, length);
+ }
+ };
+ }
+
private File writeFile(byte[] bytes) throws IOException {
File file = new File(tempDir.toFile(), UUID.randomUUID().toString());
if (!file.createNewFile()) {
