This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 8e4de02ae [core] Manage the bloomfilter buffer in cache manager (#4403)
8e4de02ae is described below
commit 8e4de02aef0259ed612492fb41a6f5d01dfe42c7
Author: WenjunMin <[email protected]>
AuthorDate: Wed Oct 30 12:59:17 2024 +0800
[core] Manage the bloomfilter buffer in cache manager (#4403)
---
.../paimon/lookup/hash/HashLookupStoreReader.java | 3 ++
.../paimon/lookup/sort/SortLookupStoreFactory.java | 3 +-
.../paimon/lookup/sort/SortLookupStoreReader.java | 34 +++++++++++++---------
.../apache/paimon/utils/FileBasedBloomFilter.java | 15 ++++++++--
.../paimon/utils/FileBasedBloomFilterTest.java | 3 +-
5 files changed, 40 insertions(+), 18 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
index c984dafdd..742cc1a60 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
@@ -169,6 +169,9 @@ public class HashLookupStoreReader
@Override
public void close() throws IOException {
+ if (bloomFilter != null) {
+ bloomFilter.close();
+ }
inputView.close();
inputView = null;
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreFactory.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreFactory.java
index 62c4131e5..244d7f9dd 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreFactory.java
@@ -52,7 +52,8 @@ public class SortLookupStoreFactory implements
LookupStoreFactory {
@Override
public SortLookupStoreReader createReader(File file, Context context)
throws IOException {
- return new SortLookupStoreReader(comparator, file, (SortContext)
context, cacheManager);
+ return new SortLookupStoreReader(
+ comparator, file, blockSize, (SortContext) context,
cacheManager);
}
@Override
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
index 5dec3c06e..04c9d287c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/SortLookupStoreReader.java
@@ -20,12 +20,13 @@ package org.apache.paimon.lookup.sort;
import org.apache.paimon.compression.BlockCompressionFactory;
import org.apache.paimon.compression.BlockDecompressor;
+import org.apache.paimon.io.PageFileInput;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.lookup.LookupStoreReader;
import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.memory.MemorySliceInput;
-import org.apache.paimon.utils.BloomFilter;
+import org.apache.paimon.utils.FileBasedBloomFilter;
import org.apache.paimon.utils.MurmurHashUtils;
import javax.annotation.Nullable;
@@ -53,13 +54,16 @@ public class SortLookupStoreReader implements
LookupStoreReader {
private final FileChannel fileChannel;
private final String filePath;
private final long fileSize;
+ private final CacheManager cacheManager;
private final BlockIterator indexBlockIterator;
- @Nullable private final BloomFilter bloomFilter;
+ @Nullable private FileBasedBloomFilter bloomFilter;
+ @Nullable private PageFileInput fileInput;
public SortLookupStoreReader(
Comparator<MemorySlice> comparator,
File file,
+ int blockSize,
SortContext context,
CacheManager cacheManager)
throws IOException {
@@ -68,21 +72,21 @@ public class SortLookupStoreReader implements
LookupStoreReader {
this.fileChannel = new FileInputStream(file).getChannel();
this.filePath = file.getAbsolutePath();
this.fileSize = context.fileSize();
+ this.cacheManager = cacheManager;
Footer footer = readFooter();
this.indexBlockIterator =
readBlock(footer.getIndexBlockHandle()).iterator();
- this.bloomFilter = readBloomFilter(footer.getBloomFilterHandle());
- }
-
- private BloomFilter readBloomFilter(@Nullable BloomFilterHandle
bloomFilterHandle)
- throws IOException {
- BloomFilter bloomFilter = null;
- if (bloomFilterHandle != null) {
- MemorySegment segment = read(bloomFilterHandle.offset(),
bloomFilterHandle.size());
- bloomFilter = new BloomFilter(bloomFilterHandle.expectedEntries(),
segment.size());
- bloomFilter.setMemorySegment(segment, 0);
+ BloomFilterHandle handle = footer.getBloomFilterHandle();
+ if (handle != null) {
+ this.fileInput = PageFileInput.create(file, blockSize, null,
fileSize, null);
+ this.bloomFilter =
+ new FileBasedBloomFilter(
+ fileInput,
+ cacheManager,
+ handle.expectedEntries(),
+ handle.offset(),
+ handle.size());
}
- return bloomFilter;
}
private Footer readFooter() throws IOException {
@@ -177,6 +181,10 @@ public class SortLookupStoreReader implements
LookupStoreReader {
@Override
public void close() throws IOException {
this.fileChannel.close();
+ if (bloomFilter != null) {
+ bloomFilter.close();
+ fileInput.close();
+ }
// TODO clear cache too
}
}
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 7b885a5e6..3d8751774 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
@@ -25,18 +25,21 @@ import org.apache.paimon.io.cache.CacheKey;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.memory.MemorySegment;
+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 . */
-public class FileBasedBloomFilter {
+public class FileBasedBloomFilter implements Closeable {
private final PageFileInput input;
private final CacheManager cacheManager;
private final BloomFilter filter;
private final long readOffset;
private final int readLength;
-
+ private final CacheKey cacheKey;
private int accessCount;
public FileBasedBloomFilter(
@@ -52,6 +55,7 @@ public class FileBasedBloomFilter {
this.readOffset = readOffset;
this.readLength = readLength;
this.accessCount = 0;
+ this.cacheKey = CacheKey.forPosition(input.file(), readOffset,
readLength);
}
public boolean testHash(int hash) {
@@ -61,7 +65,7 @@ public class FileBasedBloomFilter {
if (accessCount == REFRESH_COUNT || filter.getMemorySegment() == null)
{
MemorySegment segment =
cacheManager.getPage(
- CacheKey.forPosition(input.file(), readOffset,
readLength),
+ cacheKey,
key -> input.readPosition(readOffset, readLength),
new BloomFilterCallBack(filter));
filter.setMemorySegment(segment, 0);
@@ -75,6 +79,11 @@ public class FileBasedBloomFilter {
return filter;
}
+ @Override
+ public void close() throws IOException {
+ cacheManager.invalidPage(cacheKey);
+ }
+
/** Call back for cache manager. */
private static class BloomFilterCallBack implements CacheCallback {
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 970219e7b..8bda49005 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
@@ -59,7 +59,8 @@ public class FileBasedBloomFilterTest {
Arrays.stream(inputs)
.forEach(i ->
Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue());
- cacheManager.cache().invalidateAll();
+ filter.close();
+ Assertions.assertThat(cacheManager.cache().asMap()).isEmpty();
Assertions.assertThat(filter.bloomFilter().getMemorySegment()).isNull();
}
