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/incubator-paimon.git
commit d732b5ff6b05f6e6b48cb5d2bcee1702c49126eb Author: Aitozi <[email protected]> AuthorDate: Sun Jan 14 20:37:19 2024 +0800 [core] Introduce bloom filter for local lookup cache file This closes #2696 --- .../shortcodes/generated/core_configuration.html | 12 ++ .../benchmark/LookupBloomFilterBenchmark.java | 144 ++++++++++++++++ .../main/java/org/apache/paimon/CoreOptions.java | 13 ++ .../org/apache/paimon/io/cache/CacheManager.java | 2 +- .../paimon/io/cache/CachedRandomInputView.java | 80 +++++++-- .../apache/paimon/lookup/LookupStoreFactory.java | 6 +- .../org/apache/paimon/lookup/bloom/BitSet.java | 112 ++++++++++++ .../apache/paimon/lookup/bloom/BloomFilter.java | 148 ++++++++++++++++ .../paimon/lookup/bloom/BloomFilterBuilder.java | 71 ++++++++ .../paimon/lookup/bloom/BloomFilterTester.java | 71 ++++++++ .../paimon/lookup/hash/HashLookupStoreFactory.java | 11 +- .../paimon/lookup/hash/HashLookupStoreReader.java | 27 ++- .../paimon/lookup/hash/HashLookupStoreWriter.java | 26 ++- .../java/org/apache/paimon/utils/MathUtils.java | 10 ++ .../paimon/io/cache/CachedRandomInputViewTest.java | 18 +- .../org/apache/paimon/lookup/bloom/BitSetTest.java | 50 ++++++ .../lookup/bloom/BloomFilterBuilderTest.java | 67 +++++++ .../paimon/lookup/bloom/BloomFilterTest.java | 113 ++++++++++++ .../paimon/lookup/bloom/BloomFilterTesterTest.java | 74 ++++++++ .../lookup/hash/HashLookupStoreFactoryTest.java | 192 +++++++++++++-------- .../org/apache/paimon/utils/CommonTestUtils.java | 11 ++ .../apache/paimon/mergetree/ContainsLevels.java | 12 +- .../org/apache/paimon/mergetree/LookupLevels.java | 17 +- .../paimon/operation/KeyValueFileStoreWrite.java | 25 ++- .../apache/paimon/table/query/LocalTableQuery.java | 15 +- .../paimon/mergetree/ContainsLevelsTest.java | 6 +- .../apache/paimon/mergetree/LookupLevelsTest.java | 4 +- 27 files changed, 1229 insertions(+), 108 deletions(-) diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html index e3924df73..c8fc47052 100644 --- a/docs/layouts/shortcodes/generated/core_configuration.html +++ b/docs/layouts/shortcodes/generated/core_configuration.html @@ -273,6 +273,18 @@ Mainly to resolve data skew on primary keys. We recommend starting with 64 mb wh <td>MemorySize</td> <td>Max memory size for lookup cache.</td> </tr> + <tr> + <td><h5>lookup.cache.bloom.filter.enabled</h5></td> + <td style="word-wrap: break-word;">true</td> + <td>Boolean</td> + <td>Whether to enable the bloom filter for lookup cache.</td> + </tr> + <tr> + <td><h5>lookup.cache.bloom.filter.fpp</h5></td> + <td style="word-wrap: break-word;">0.05</td> + <td>Double</td> + <td>Define the default false positive probability for lookup cache bloom filters.</td> + </tr> <tr> <td><h5>lookup.hash-load-factor</h5></td> <td style="word-wrap: break-word;">0.75</td> diff --git a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java new file mode 100644 index 000000000..92c00c545 --- /dev/null +++ b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java @@ -0,0 +1,144 @@ +/* + * + * 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.benchmark; + +import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; +import org.apache.paimon.lookup.hash.HashLookupStoreFactory; +import org.apache.paimon.lookup.hash.HashLookupStoreReader; +import org.apache.paimon.lookup.hash.HashLookupStoreWriter; +import org.apache.paimon.options.MemorySize; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.UUID; +import java.util.concurrent.ThreadLocalRandom; + +/** Benchmark for measure the bloom filter for lookup. */ +public class LookupBloomFilterBenchmark { + + @TempDir Path tempDir; + ThreadLocalRandom rnd = ThreadLocalRandom.current(); + + @Test + public void testHighMatch() throws Exception { + innerTest("lookup", generateSequenceInputs(0, 100000), generateRandomInputs(0, 100000)); + } + + @Test + public void testHalfMatch() throws Exception { + innerTest("lookup", generateSequenceInputs(0, 100000), generateRandomInputs(50000, 150000)); + } + + @Test + public void testLowMatch() throws Exception { + innerTest( + "lookup", generateSequenceInputs(0, 100000), generateRandomInputs(100000, 200000)); + } + + private byte[][] generateSequenceInputs(int start, int end) { + int count = end - start; + byte[][] result = new byte[count][4]; + for (int i = 0; i < count; i++) { + result[i] = intToByteArray(i); + } + return result; + } + + private byte[][] generateRandomInputs(int start, int end) { + int count = end - start; + byte[][] result = new byte[count][4]; + for (int i = 0; i < count; i++) { + result[i] = intToByteArray(rnd.nextInt(start, end)); + } + return result; + } + + public byte[] intToByteArray(int value) { + return new byte[] { + (byte) (value >>> 24), (byte) (value >>> 16), (byte) (value >>> 8), (byte) value + }; + } + + public void innerTest(String name, byte[][] inputs, byte[][] probe) throws Exception { + Benchmark benchmark = + new Benchmark(name, probe.length).setNumWarmupIters(1).setOutputPerIteration(true); + + int[] valueLengths = {0, 500, 1000, 2000}; + for (int valueLength : valueLengths) { + HashLookupStoreReader reader = writeData(null, inputs, valueLength); + + benchmark.addCase( + String.format("bf-disabled-%dB-value", valueLength), + 5, + () -> { + try { + for (byte[] key : probe) { + reader.lookup(key); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + HashLookupStoreReader reader2 = + writeData( + BloomFilterBuilder.bfBuilder(inputs.length, 0.05), inputs, valueLength); + + benchmark.addCase( + String.format("bf-enabled-%dB-value", valueLength), + 5, + () -> { + try { + for (byte[] key : probe) { + reader2.lookup(key); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + benchmark.run(); + } + + private HashLookupStoreReader writeData( + BloomFilterBuilder filter, byte[][] inputs, int valueLength) throws IOException { + byte[] value = new byte[valueLength]; + Arrays.fill(value, (byte) 1); + HashLookupStoreFactory factory = + new HashLookupStoreFactory( + new CacheManager(MemorySize.ofMebiBytes(10)), 16 * 1024, 0.75); + + File file = new File(tempDir.toFile(), UUID.randomUUID().toString()); + HashLookupStoreWriter writer = factory.createWriter(file, () -> filter); + for (byte[] input : inputs) { + writer.put(input, value); + } + writer.close(); + return factory.createReader(file); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java index 6a3fbaeee..7c8cfeaa4 100644 --- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java @@ -728,6 +728,19 @@ public class CoreOptions implements Serializable { .defaultValue(MemorySize.parse("256 mb")) .withDescription("Max memory size for lookup cache."); + public static final ConfigOption<Boolean> LOOKUP_CACHE_BLOOM_FILTER_ENABLED = + key("lookup.cache.bloom.filter.enabled") + .booleanType() + .defaultValue(true) + .withDescription("Whether to enable the bloom filter for lookup cache."); + + public static final ConfigOption<Double> LOOKUP_CACHE_BLOOM_FILTER_FPP = + key("lookup.cache.bloom.filter.fpp") + .doubleType() + .defaultValue(0.05) + .withDescription( + "Define the default false positive probability for lookup cache bloom filters."); + public static final ConfigOption<Integer> READ_BATCH_SIZE = key("read.batch-size") .intType() 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 af5ab7397..69dd9461f 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 @@ -48,7 +48,7 @@ public class CacheManager { } @VisibleForTesting - Cache<CacheKey, CacheValue> cache() { + public Cache<CacheKey, CacheValue> cache() { return cache; } diff --git a/paimon-common/src/main/java/org/apache/paimon/io/cache/CachedRandomInputView.java b/paimon-common/src/main/java/org/apache/paimon/io/cache/CachedRandomInputView.java index c06ffbd6c..739f1ce38 100644 --- a/paimon-common/src/main/java/org/apache/paimon/io/cache/CachedRandomInputView.java +++ b/paimon-common/src/main/java/org/apache/paimon/io/cache/CachedRandomInputView.java @@ -33,6 +33,8 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; +import java.util.function.BiConsumer; /** * A {@link SeekableDataInputView} to read bytes from {@link RandomAccessFile}, the bytes can be @@ -44,7 +46,7 @@ public class CachedRandomInputView extends AbstractPagedInputView private final RandomAccessFile file; private final long fileLength; private final CacheManager cacheManager; - private final Map<Integer, MemorySegment> segments; + private final Map<CacheKey, MemorySegment> segments; private final int segmentSize; private final int segmentSizeBits; private final int segmentSizeMask; @@ -73,12 +75,13 @@ public class CachedRandomInputView extends AbstractPagedInputView } private MemorySegment getCurrentPage() { - MemorySegment segment = segments.get(currentSegmentIndex); + long position = segmentIndexToPosition(currentSegmentIndex); + int length = (int) Math.min(segmentSize, fileLength - position); + CacheKey cacheKey = new CacheKey(position, length); + MemorySegment segment = segments.get(cacheKey); if (segment == null) { - long offset = segmentIndexToPosition(currentSegmentIndex); - int length = (int) Math.min(segmentSize, fileLength - offset); - segment = cacheManager.getPage(file, offset, length, this::invalidPage); - segments.put(currentSegmentIndex, segment); + segment = cacheManager.getPage(file, position, length, this::invalidPage); + segments.put(cacheKey, segment); } return segment; } @@ -99,18 +102,46 @@ public class CachedRandomInputView extends AbstractPagedInputView } private void invalidPage(long position, int length) { - segments.remove(positionToSegmentIndex(position)); + segments.remove(new CacheKey(position, length)); + } + + /** + * Read a continuous segment of the specified length at a given position. + * + * @param position The start reading position. + * @param length The length to read. + * @return The wrapped segment. NOTE: this segment should not be cached outside this class. + */ + public MemorySegment read(int position, int length, BiConsumer<Long, Integer> cleanCallback) { + final int offset = position & this.segmentSizeMask; + length = (int) Math.min(length, fileLength - offset); + CacheKey cacheKey = new CacheKey(position, length); + MemorySegment segment = segments.get(cacheKey); + if (segment == null) { + segment = + cacheManager.getPage( + file, + position, + length, + new BiConsumer<Long, Integer>() { + @Override + public void accept(Long position, Integer length) { + invalidPage(position, length); + cleanCallback.accept(position, length); + } + }); + segments.put(cacheKey, segment); + } + return segment; } @Override public void close() throws IOException { // copy out to avoid ConcurrentModificationException - List<Integer> pages = new ArrayList<>(segments.keySet()); + List<CacheKey> pages = new ArrayList<>(segments.keySet()); pages.forEach( page -> { - long offset = segmentIndexToPosition(page); - int length = (int) Math.min(segmentSize, fileLength - offset); - cacheManager.invalidPage(file, offset, length); + cacheManager.invalidPage(file, page.readOffset, page.readLength); }); file.close(); @@ -123,4 +154,31 @@ public class CachedRandomInputView extends AbstractPagedInputView private long segmentIndexToPosition(int segmentIndex) { return (long) segmentIndex << segmentSizeBits; } + + private static class CacheKey { + private final long readOffset; + private final int readLength; + + public CacheKey(long readOffset, int readLength) { + this.readOffset = readOffset; + this.readLength = readLength; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof CacheKey)) { + return false; + } + CacheKey cacheKey = (CacheKey) o; + return readOffset == cacheKey.readOffset && readLength == cacheKey.readLength; + } + + @Override + public int hashCode() { + return Objects.hash(readOffset, readLength); + } + } } diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java index a09788d7f..2143bdd9f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java @@ -18,8 +18,11 @@ package org.apache.paimon.lookup; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; + import java.io.File; import java.io.IOException; +import java.util.function.Supplier; /** * A key-value store for lookup, key-value store should be single binary file written once and ready @@ -32,7 +35,8 @@ import java.io.IOException; */ public interface LookupStoreFactory { - LookupStoreWriter createWriter(File file) throws IOException; + LookupStoreWriter createWriter(File file, Supplier<BloomFilterBuilder> bfProvider) + throws IOException; LookupStoreReader createReader(File file) throws IOException; } diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java new file mode 100644 index 000000000..581158b4d --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java @@ -0,0 +1,112 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.utils.Preconditions; + +/** BitSet util. Copied from Apache Flink Project. */ +public class BitSet { + private static final int BYTE_INDEX_MASK = 0x00000007; + + private MemorySegment memorySegment; + + // MemorySegment byte array offset. + private int offset; + + // The BitSet byte size. + private int byteLength; + + // The BitSet bit size. + private int bitLength; + + public BitSet(int byteSize) { + Preconditions.checkArgument(byteSize > 0, "bits size should be greater than 0."); + this.byteLength = byteSize; + this.bitLength = byteSize << 3; + } + + public void setMemorySegment(MemorySegment memorySegment, int offset) { + Preconditions.checkArgument(memorySegment != null, "MemorySegment can not be null."); + Preconditions.checkArgument(offset >= 0, "Offset should be positive integer."); + Preconditions.checkArgument( + offset + byteLength <= memorySegment.size(), + "Could not set MemorySegment, the remain buffers is not enough."); + this.memorySegment = memorySegment; + this.offset = offset; + } + + /** + * Sets the bit at specified index. + * + * @param index - position + */ + public void set(int index) { + Preconditions.checkArgument(index < bitLength && index >= 0); + + int byteIndex = index >>> 3; + byte current = memorySegment.get(offset + byteIndex); + current |= (1 << (index & BYTE_INDEX_MASK)); + memorySegment.put(offset + byteIndex, current); + } + + /** + * Returns true if the bit is set in the specified index. + * + * @param index - position + * @return - value at the bit position + */ + public boolean get(int index) { + Preconditions.checkArgument(index < bitLength && index >= 0); + + int byteIndex = index >>> 3; + byte current = memorySegment.get(offset + byteIndex); + return (current & (1 << (index & BYTE_INDEX_MASK))) != 0; + } + + /** Number of bits. */ + public int bitSize() { + return bitLength; + } + + /** Clear the bit set. */ + public void clear() { + int index = 0; + while (index + 8 <= byteLength) { + memorySegment.putLong(offset + index, 0L); + index += 8; + } + while (index < byteLength) { + memorySegment.put(offset + index, (byte) 0); + index += 1; + } + } + + @Override + public String toString() { + StringBuilder output = new StringBuilder(); + output.append("BitSet:\n"); + output.append("\tMemorySegment:").append(memorySegment.size()).append("\n"); + output.append("\tOffset:").append(offset).append("\n"); + output.append("\tLength:").append(byteLength).append("\n"); + return output.toString(); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java new file mode 100644 index 000000000..cfcb7e67a --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java @@ -0,0 +1,148 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.memory.MemorySegment; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * BloomFilter is a probabilistic data structure for set membership check. BloomFilters are highly + * space efficient when compared to using a HashSet. Because of the probabilistic nature of bloom + * filter false positive (element not present in bloom filter but test() says true) are possible but + * false negatives are not possible (if element is present then test() will never say false). The + * false positive probability is configurable depending on which storage requirement may increase or + * decrease. Lower the false positive probability greater is the space requirement. Bloom filters + * are sensitive to number of elements that will be inserted in the bloom filter. During the + * creation of bloom filter expected number of entries must be specified. If the number of + * insertions exceed the specified initial number of entries then false positive probability will + * increase accordingly. + * + * <p>Internally, this implementation of bloom filter uses MemorySegment to store BitSet, + * BloomFilter and BitSet are designed to be able to switch between different MemorySegments, so + * that Flink can share the same BloomFilter/BitSet object instance for different bloom filters. + * + * <p>Part of this class refers to the implementation from Apache Hive project + * https://github.com/apache/hive/blob/master/common/src/java/org/apache/hive/common/util/BloomFilter.java + */ +public class BloomFilter { + + protected BitSet bitSet; + protected long expectedEntries; + protected int numHashFunctions; + + public BloomFilter(long expectedEntries, int byteSize) { + checkArgument(expectedEntries > 0, "expectedEntries should be > 0"); + this.expectedEntries = expectedEntries; + this.numHashFunctions = optimalNumOfHashFunctions(expectedEntries, byteSize << 3); + this.bitSet = new BitSet(byteSize); + } + + public void setBitsLocation(MemorySegment memorySegment, int offset) { + this.bitSet.setMemorySegment(memorySegment, offset); + } + + /** + * Compute optimal bits number with given input entries and expected false positive probability. + * + * @param inputEntries + * @param fpp + * @return optimal bits number + */ + public static int optimalNumOfBits(long inputEntries, double fpp) { + int numBits = (int) (-inputEntries * Math.log(fpp) / (Math.log(2) * Math.log(2))); + return numBits; + } + + /** + * Compute the false positive probability based on given input entries and bits size. Note: this + * is just the math expected value, you should not expect the fpp in real case would under the + * return value for certain. + * + * @param inputEntries + * @param bitSize + * @return + */ + public static double estimateFalsePositiveProbability(long inputEntries, int bitSize) { + int numFunction = optimalNumOfHashFunctions(inputEntries, bitSize); + double p = Math.pow(Math.E, -(double) numFunction * inputEntries / bitSize); + double estimatedFPP = Math.pow(1 - p, numFunction); + return estimatedFPP; + } + + /** + * compute the optimal hash function number with given input entries and bits size, which would + * make the false positive probability lowest. + * + * @param expectEntries + * @param bitSize + * @return hash function number + */ + static int optimalNumOfHashFunctions(long expectEntries, long bitSize) { + return Math.max(1, (int) Math.round((double) bitSize / expectEntries * Math.log(2))); + } + + public void addHash(int hash32) { + int hash1 = hash32; + int hash2 = hash32 >>> 16; + + for (int i = 1; i <= numHashFunctions; i++) { + int combinedHash = hash1 + (i * hash2); + // hashcode should be positive, flip all the bits if it's negative + if (combinedHash < 0) { + combinedHash = ~combinedHash; + } + int pos = combinedHash % bitSet.bitSize(); + bitSet.set(pos); + } + } + + public boolean testHash(int hash32) { + int hash1 = hash32; + int hash2 = hash32 >>> 16; + + for (int i = 1; i <= numHashFunctions; i++) { + int combinedHash = hash1 + (i * hash2); + // hashcode should be positive, flip all the bits if it's negative + if (combinedHash < 0) { + combinedHash = ~combinedHash; + } + int pos = combinedHash % bitSet.bitSize(); + if (!bitSet.get(pos)) { + return false; + } + } + return true; + } + + public void reset() { + this.bitSet.clear(); + } + + @Override + public String toString() { + StringBuilder output = new StringBuilder(); + output.append("BloomFilter:\n"); + output.append("\thash function number:").append(numHashFunctions).append("\n"); + output.append(bitSet); + return output.toString(); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterBuilder.java b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterBuilder.java new file mode 100644 index 000000000..856010e4d --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterBuilder.java @@ -0,0 +1,71 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.utils.Preconditions; + +/** Bloom filter based on one memory segment. */ +public class BloomFilterBuilder { + + private final MemorySegment buffer; + private final BloomFilter filter; + private final long numRecords; + + public BloomFilterBuilder(MemorySegment buffer, long numRecords) { + this.buffer = buffer; + this.filter = new BloomFilter(numRecords, buffer.size()); + filter.setBitsLocation(buffer, 0); + this.numRecords = numRecords; + } + + public boolean testHash(int hash) { + return filter.testHash(hash); + } + + public void addHash(int hash) { + filter.addHash(hash); + } + + public MemorySegment getBuffer() { + return buffer; + } + + public long getNumRecords() { + return numRecords; + } + + public static BloomFilterBuilder bfBuilder(long expectedRow, double fpp) { + int numBytes = (int) Math.ceil(BloomFilter.optimalNumOfBits(expectedRow, fpp) / 8D); + Preconditions.checkArgument( + numBytes > 0, + "The optimal bits should > 0. expectedRow: %s, fpp: %s", + expectedRow, + fpp); + return new BloomFilterBuilder(MemorySegment.wrap(new byte[numBytes]), expectedRow); + } + + @VisibleForTesting + public BloomFilter getFilter() { + return filter; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterTester.java b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterTester.java new file mode 100644 index 000000000..3f4e67d13 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterTester.java @@ -0,0 +1,71 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.io.cache.CachedRandomInputView; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.utils.Preconditions; + +import java.util.function.BiConsumer; + +/** Util to apply a built bloom filter . */ +public class BloomFilterTester { + + private final BloomFilter filter; + private final CachedRandomInputView inputView; + private final int bfIndex; + private final int bfLength; + private final MemorySegment[] segments; + + public BloomFilterTester( + long numRecords, int bfIndex, int bfLength, CachedRandomInputView inputView) { + Preconditions.checkArgument(numRecords >= 0); + this.filter = new BloomFilter(numRecords, bfLength); + this.segments = new MemorySegment[1]; + this.inputView = inputView; + this.bfIndex = bfIndex; + this.bfLength = bfLength; + } + + public boolean testHash(int hash) { + if (segments[0] == null) { + MemorySegment segment = + inputView.read( + bfIndex, + bfLength, + new BiConsumer<Long, Integer>() { + @Override + public void accept(Long readOffset, Integer readLength) { + segments[0] = null; + } + }); + segments[0] = segment; + } + filter.setBitsLocation(segments[0], 0); + return filter.testHash(hash); + } + + @VisibleForTesting + public MemorySegment[] getSegments() { + return segments; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java index 9b02b8099..31828a8dd 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java @@ -20,9 +20,11 @@ package org.apache.paimon.lookup.hash; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.lookup.LookupStoreFactory; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import java.io.File; import java.io.IOException; +import java.util.function.Supplier; /** A {@link LookupStoreFactory} which uses hash to lookup records on disk. */ public class HashLookupStoreFactory implements LookupStoreFactory { @@ -38,12 +40,13 @@ public class HashLookupStoreFactory implements LookupStoreFactory { } @Override - public HashLookupStoreWriter createWriter(File file) throws IOException { - return new HashLookupStoreWriter(loadFactor, file); + public HashLookupStoreReader createReader(File file) throws IOException { + return new HashLookupStoreReader(cacheManager, cachePageSize, file); } @Override - public HashLookupStoreReader createReader(File file) throws IOException { - return new HashLookupStoreReader(cacheManager, cachePageSize, file); + public HashLookupStoreWriter createWriter(File file, Supplier<BloomFilterBuilder> bfProvider) + throws IOException { + return new HashLookupStoreWriter(loadFactor, file, bfProvider); } } 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 fd98ac5fd..160648eed 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 @@ -21,12 +21,15 @@ package org.apache.paimon.lookup.hash; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.io.cache.CachedRandomInputView; import org.apache.paimon.lookup.LookupStoreReader; +import org.apache.paimon.lookup.bloom.BloomFilterTester; import org.apache.paimon.utils.MurmurHashUtils; import org.apache.paimon.utils.VarLengthIntUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.BufferedInputStream; import java.io.DataInputStream; import java.io.File; @@ -61,6 +64,7 @@ public class HashLookupStoreReader private CachedRandomInputView inputView; // Buffers private final byte[] slotBuffer; + @Nullable private BloomFilterTester bloomFilter; HashLookupStoreReader(CacheManager cacheManager, int cachePageSize, File file) throws IOException { @@ -69,6 +73,8 @@ public class HashLookupStoreReader throw new FileNotFoundException("File " + file.getAbsolutePath() + " not found"); } LOG.info("Opening file {}", file.getName()); + // Create Mapped file in read-only mode + inputView = new CachedRandomInputView(file, cacheManager, cachePageSize); // Open file and read metadata long createdAt; @@ -78,6 +84,7 @@ public class HashLookupStoreReader int keyCount; int indexOffset; long dataOffset; + boolean bloomFilterEnabled; try { // Time createdAt = dataInputStream.readLong(); @@ -88,6 +95,17 @@ public class HashLookupStoreReader int keyLengthCount = dataInputStream.readInt(); // Max key length int maxKeyLength = dataInputStream.readInt(); + // BloomFilter + bloomFilterEnabled = dataInputStream.readBoolean(); + if (bloomFilterEnabled) { + long recordCount = dataInputStream.readLong(); + int bfBytes = dataInputStream.readInt(); + // Bloom filter buffer start index + int bfStartIndex = dataInputStream.readInt(); + // Skip bloom filter buffers + dataInputStream.skipBytes(bfBytes); + bloomFilter = new BloomFilterTester(recordCount, bfStartIndex, bfBytes, inputView); + } // Read offset counts and keys indexOffsets = new int[maxKeyLength + 1]; @@ -127,14 +145,12 @@ public class HashLookupStoreReader inputStream.close(); } - // Create Mapped file in read-only mode - inputView = new CachedRandomInputView(file, cacheManager, cachePageSize); - // logging DecimalFormat integerFormat = new DecimalFormat("#,##0.00"); StringBuilder statMsg = new StringBuilder("Storage metadata\n"); statMsg.append(" Created at: ").append(formatCreatedAt(createdAt)).append("\n"); statMsg.append(" Key count: ").append(keyCount).append("\n"); + statMsg.append(" Bloom filter: ").append(bloomFilterEnabled).append("\n"); for (int i = 0; i < keyCounts.length; i++) { if (keyCounts[i] > 0) { statMsg.append(" Key count for key length ") @@ -159,6 +175,11 @@ public class HashLookupStoreReader if (keyLength >= slots.length || keyCounts[keyLength] == 0) { return null; } + + if (bloomFilter != null && !bloomFilter.testHash(MurmurHashUtils.hashBytes(key))) { + return null; + } + long hash = MurmurHashUtils.hashBytesPositive(key); int numSlots = slots[keyLength]; int slotSize = slotSizes[keyLength]; diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java index aae7da127..7fcd9f696 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java @@ -19,12 +19,15 @@ package org.apache.paimon.lookup.hash; import org.apache.paimon.lookup.LookupStoreWriter; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.utils.MurmurHashUtils; import org.apache.paimon.utils.VarLengthIntUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.DataInputStream; @@ -42,6 +45,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.UUID; +import java.util.function.Supplier; /** Internal write implementation for hash kv store. */ public class HashLookupStoreWriter implements LookupStoreWriter { @@ -76,8 +80,10 @@ public class HashLookupStoreWriter implements LookupStoreWriter { private int valueCount; // Number of collisions private int collisions; + @Nullable private final BloomFilterBuilder bloomFilter; - HashLookupStoreWriter(double loadFactor, File file) throws IOException { + HashLookupStoreWriter(double loadFactor, File file, Supplier<BloomFilterBuilder> bfProvider) + throws IOException { this.loadFactor = loadFactor; if (loadFactor <= 0.0 || loadFactor >= 1.0) { throw new IllegalArgumentException( @@ -98,6 +104,7 @@ public class HashLookupStoreWriter implements LookupStoreWriter { dataLengths = new long[0]; maxOffsetLengths = new int[0]; keyCounts = new int[0]; + bloomFilter = bfProvider.get(); } @Override @@ -145,6 +152,9 @@ public class HashLookupStoreWriter implements LookupStoreWriter { keyCount++; keyCounts[keyLength]++; + if (bloomFilter != null) { + bloomFilter.addHash(MurmurHashUtils.hashBytes(key)); + } } @Override @@ -223,6 +233,20 @@ public class HashLookupStoreWriter implements LookupStoreWriter { // Write the max value for keyLength dataOutputStream.writeInt(maxKeyLength); + // Write whether the bloom filter is enabled. + dataOutputStream.writeBoolean(bloomFilter != null); + + if (bloomFilter != null) { + dataOutputStream.writeLong(bloomFilter.getNumRecords()); + dataOutputStream.writeInt(bloomFilter.getBuffer().size()); + + // Write the bloom filter start index + dataOutputStream.writeInt(dataOutputStream.size() + Integer.SIZE / Byte.SIZE); + + dataOutputStream.write(bloomFilter.getBuffer().getArray()); + LOG.info("Bloom filter size: {} bytes", bloomFilter.getBuffer().size()); + } + // For each keyLength long datasLength = 0L; for (int i = 0; i < keyCounts.length; i++) { diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/MathUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/MathUtils.java index 4bd0ba680..310ae9b7f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/MathUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/MathUtils.java @@ -32,6 +32,16 @@ public class MathUtils { return Integer.highestOneBit(value); } + /** + * Checks whether the given value is a power of two. + * + * @param value The value to check. + * @return True, if the value is a power of two, false otherwise. + */ + public static boolean isPowerOf2(long value) { + return (value & (value - 1)) == 0; + } + /** * Computes the logarithm of the given value to the base of 2. This method throws an error, if * the given argument is not a power of 2. diff --git a/paimon-common/src/test/java/org/apache/paimon/io/cache/CachedRandomInputViewTest.java b/paimon-common/src/test/java/org/apache/paimon/io/cache/CachedRandomInputViewTest.java index c00dfdf27..68cf28674 100644 --- a/paimon-common/src/test/java/org/apache/paimon/io/cache/CachedRandomInputViewTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/io/cache/CachedRandomInputViewTest.java @@ -65,8 +65,9 @@ public class CachedRandomInputViewTest { } File file = writeFile(bytes); + int pageSize = 1024; CacheManager cacheManager = new CacheManager(MemorySize.ofKibiBytes(128)); - CachedRandomInputView view = new CachedRandomInputView(file, cacheManager, 1024); + CachedRandomInputView view = new CachedRandomInputView(file, cacheManager, pageSize); // read first one // this assertThatCode check the ConcurrentModificationException is not threw. @@ -93,6 +94,21 @@ public class CachedRandomInputViewTest { assertThat(view.readLong()).isEqualTo(segment.getLongBigEndian(position)); } + byte[] expected = new byte[100]; + System.arraycopy(segment.getArray(), 0, expected, 0, 100); + assertThat(view.read(0, 100, (page, length) -> {}).getArray()).isEqualTo(expected); + + // Read not aligned + expected = new byte[100]; + System.arraycopy(segment.getArray(), 1, expected, 0, 100); + assertThat(view.read(1, 100, (page, length) -> {}).getArray()).isEqualTo(expected); + + // Read cross pagesize + expected = new byte[pageSize * 2 + 1]; + System.arraycopy(segment.getArray(), pageSize * 2, expected, 0, pageSize * 2 + 1); + assertThat(view.read(pageSize * 2, pageSize * 2 + 1, (page, length) -> {}).getArray()) + .isEqualTo(expected); + view.close(); assertThat(cacheManager.cache().asMap().size()).isEqualTo(0); } diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BitSetTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BitSetTest.java new file mode 100644 index 000000000..8acb93cde --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BitSetTest.java @@ -0,0 +1,50 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.memory.MemorySegment; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.stream.IntStream; + +/** Test for {@link BitSet}. */ +public class BitSetTest { + + @Test + public void testBitSet() { + BitSet bitSet = new BitSet(100); + Assertions.assertThatThrownBy( + () -> bitSet.setMemorySegment(MemorySegment.wrap(new byte[99]), 0)) + .isInstanceOf(IllegalArgumentException.class); + Assertions.assertThatThrownBy( + () -> bitSet.setMemorySegment(MemorySegment.wrap(new byte[100]), 1)) + .isInstanceOf(IllegalArgumentException.class); + bitSet.setMemorySegment(MemorySegment.wrap(new byte[100]), 0); + IntStream.range(0, 100).forEach(i -> Assertions.assertThat(bitSet.get(i)).isFalse()); + IntStream.range(0, 50).forEach(bitSet::set); + IntStream.range(0, 50).forEach(i -> Assertions.assertThat(bitSet.get(i)).isTrue()); + IntStream.range(50, 100).forEach(i -> Assertions.assertThat(bitSet.get(i)).isFalse()); + bitSet.clear(); + IntStream.range(0, 100).forEach(i -> Assertions.assertThat(bitSet.get(i)).isFalse()); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterBuilderTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterBuilderTest.java new file mode 100644 index 000000000..f4c649ae1 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterBuilderTest.java @@ -0,0 +1,67 @@ +/* + * + * 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.lookup.bloom; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import static org.apache.paimon.utils.CommonTestUtils.generateRandomInts; + +/** Test for {@link BloomFilterBuilder}. */ +public class BloomFilterBuilderTest { + + @Test + public void testOneSegment() { + BloomFilterBuilder filter = BloomFilterBuilder.bfBuilder(100, 0.01); + int[] inputs = generateRandomInts(100); + for (int input : inputs) { + filter.addHash(Integer.hashCode(input)); + } + + for (int input : inputs) { + Assertions.assertThat(filter.testHash(Integer.hashCode(input))).isTrue(); + } + } + + @Test + public void testEstimatedHashFunctions() { + Assertions.assertThat(BloomFilterBuilder.bfBuilder(1000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(10_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(100_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(100_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(100_000, 0.05).getFilter().numHashFunctions) + .isEqualTo(4); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(1_000_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat( + BloomFilterBuilder.bfBuilder(1_000_000, 0.05).getFilter().numHashFunctions) + .isEqualTo(4); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java new file mode 100644 index 000000000..7b3e7afc0 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java @@ -0,0 +1,113 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.memory.MemorySegment; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +import static org.apache.paimon.utils.CommonTestUtils.generateRandomInts; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Test for {@link BloomFilter}. */ +public class BloomFilterTest { + + @Test + public void testBloomNumBits() { + Assertions.assertThat(BloomFilter.optimalNumOfBits(0, 0)).isEqualTo(0); + Assertions.assertThat(BloomFilter.optimalNumOfBits(0, 1)).isEqualTo(0); + Assertions.assertThat(BloomFilter.optimalNumOfBits(1, 1)).isEqualTo(0); + Assertions.assertThat(BloomFilter.optimalNumOfBits(1, 0.03)).isEqualTo(7); + Assertions.assertThat(BloomFilter.optimalNumOfBits(10, 0.03)).isEqualTo(72); + Assertions.assertThat(BloomFilter.optimalNumOfBits(100, 0.03)).isEqualTo(729); + Assertions.assertThat(BloomFilter.optimalNumOfBits(1000, 0.03)).isEqualTo(7298); + Assertions.assertThat(BloomFilter.optimalNumOfBits(10000, 0.03)).isEqualTo(72984); + Assertions.assertThat(BloomFilter.optimalNumOfBits(100000, 0.03)).isEqualTo(729844); + Assertions.assertThat(BloomFilter.optimalNumOfBits(1000000, 0.03)).isEqualTo(7298440); + Assertions.assertThat(BloomFilter.optimalNumOfBits(1000000, 0.05)).isEqualTo(6235224); + Assertions.assertThat(BloomFilter.optimalNumOfBits(300000000, 0.05)).isEqualTo(1870567268); + Assertions.assertThat(BloomFilter.optimalNumOfBits(300000000, 0.1)).isEqualTo(1437758756); + Assertions.assertThat(BloomFilter.optimalNumOfBits(300000000, 0.5)).isEqualTo(432808512); + Assertions.assertThat(BloomFilter.optimalNumOfBits(3000000000L, 0.8)).isEqualTo(1393332198); + Assertions.assertThat(BloomFilter.optimalNumOfBits(3000000000L, 0.9)).isEqualTo(657882327); + Assertions.assertThat(BloomFilter.optimalNumOfBits(3000000000L, 1)).isEqualTo(0); + } + + @Test + public void testBloomNumHashFunctions() { + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(-1, -1)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(0, 0)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(10, 0)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(10, 10)); + assertEquals(7, BloomFilter.optimalNumOfHashFunctions(10, 100)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(100, 100)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(1000, 100)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(10000, 100)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(100000, 100)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(1000000, 100)); + assertEquals(3634, BloomFilter.optimalNumOfHashFunctions(100, 64 * 1024 * 8)); + assertEquals(363, BloomFilter.optimalNumOfHashFunctions(1000, 64 * 1024 * 8)); + assertEquals(36, BloomFilter.optimalNumOfHashFunctions(10000, 64 * 1024 * 8)); + assertEquals(4, BloomFilter.optimalNumOfHashFunctions(100000, 64 * 1024 * 8)); + assertEquals(1, BloomFilter.optimalNumOfHashFunctions(1000000, 64 * 1024 * 8)); + } + + @Test + public void testBloomFilter() { + BloomFilter filter = new BloomFilter(100, 1024); + + // segment1 + MemorySegment segment1 = MemorySegment.wrap(new byte[1024]); + filter.setBitsLocation(segment1, 0); + int[] inputs1 = generateRandomInts(100); + Arrays.stream(inputs1).forEach(i -> filter.addHash(Integer.hashCode(i))); + Arrays.stream(inputs1) + .forEach(i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue()); + + // segments 2 + MemorySegment segment2 = MemorySegment.wrap(new byte[1024]); + filter.setBitsLocation(segment2, 0); + int[] inputs2 = generateRandomInts(100); + Arrays.stream(inputs2).forEach(i -> filter.addHash(Integer.hashCode(i))); + Arrays.stream(inputs2) + .forEach(i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue()); + + // switch to segment1 + filter.setBitsLocation(segment1, 0); + Arrays.stream(inputs1) + .forEach(i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue()); + + // clear segment1 + filter.reset(); + Arrays.stream(inputs1) + .forEach( + i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isFalse()); + // clear segment2 + filter.setBitsLocation(segment2, 0); + filter.reset(); + Arrays.stream(inputs2) + .forEach( + i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isFalse()); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java new file mode 100644 index 000000000..15a4dd6ce --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java @@ -0,0 +1,74 @@ +/* + * + * 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.lookup.bloom; + +import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.io.cache.CachedRandomInputView; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.utils.CommonTestUtils; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Arrays; +import java.util.UUID; + +/** Test for {@link BloomFilterTester}. */ +public class BloomFilterTesterTest { + + @TempDir Path tempDir; + + @Test + public void testProbe() throws IOException { + + MemorySegment segment = MemorySegment.wrap(new byte[1000]); + BloomFilterBuilder filter = new BloomFilterBuilder(segment, 100); + int[] inputs = CommonTestUtils.generateRandomInts(100); + Arrays.stream(inputs).forEach(i -> filter.addHash(Integer.hashCode(i))); + File file = writeFile(segment.getArray()); + + CacheManager cacheManager = new CacheManager(MemorySize.ofMebiBytes(1)); + BloomFilterTester tester = + new BloomFilterTester( + 100, 0, 1000, new CachedRandomInputView(file, cacheManager, 1024)); + + Arrays.stream(inputs) + .forEach(i -> Assertions.assertThat(tester.testHash(Integer.hashCode(i))).isTrue()); + cacheManager.cache().invalidateAll(); + Assertions.assertThat(tester.getSegments()[0]).isNull(); + } + + private File writeFile(byte[] bytes) throws IOException { + File file = new File(tempDir.toFile(), UUID.randomUUID().toString()); + if (!file.createNewFile()) { + throw new IOException("Can not create: " + file); + } + Files.write(file.toPath(), bytes, StandardOpenOption.WRITE); + return file; + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java index 89138549a..d1e48a4da 100644 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java @@ -24,14 +24,16 @@ package org.apache.paimon.lookup.hash; import org.apache.paimon.io.DataOutputSerializer; import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.options.MemorySize; import org.apache.paimon.utils.MathUtils; import org.apache.paimon.utils.VarLengthIntUtils; import org.apache.commons.math3.random.RandomDataGenerator; import org.junit.jupiter.api.BeforeEach; -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.MethodSource; import java.io.File; import java.io.IOException; @@ -54,6 +56,7 @@ public class HashLookupStoreFactoryTest { @TempDir Path tempDir; private final RandomDataGenerator random = new RandomDataGenerator(); + private final int pageSize = 1024; private File file; private HashLookupStoreFactory factory; @@ -62,13 +65,24 @@ public class HashLookupStoreFactoryTest { public void setUp() throws IOException { this.factory = new HashLookupStoreFactory( - new CacheManager(MemorySize.ofMebiBytes(1)), 1024, 0.75d); + new CacheManager(MemorySize.ofMebiBytes(1)), pageSize, 0.75d); this.file = new File(tempDir.toFile(), UUID.randomUUID().toString()); if (!file.createNewFile()) { throw new IOException("Can not create file: " + file); } } + public static Object[] enableBloomFilter() { + return new Object[] {false, true}; + } + + private BloomFilterBuilder createBloomFiler(boolean enabled) { + if (!enabled) { + return null; + } + return BloomFilterBuilder.bfBuilder(100, 0.01); + } + private byte[] toBytes(Object o) { return toBytes(o.toString()); } @@ -77,9 +91,11 @@ public class HashLookupStoreFactoryTest { return str.getBytes(StandardCharsets.UTF_8); } - @Test - public void testEmpty() throws IOException { - HashLookupStoreWriter writer = factory.createWriter(file); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testEmpty(boolean enableBloomFilter) throws IOException { + HashLookupStoreWriter writer = + factory.createWriter(file, () -> createBloomFiler(enableBloomFilter)); writer.close(); assertThat(file.exists()).isTrue(); @@ -91,9 +107,11 @@ public class HashLookupStoreFactoryTest { reader.close(); } - @Test - public void testOneKey() throws IOException { - HashLookupStoreWriter writer = factory.createWriter(file); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testOneKey(boolean enableBloomFilter) throws IOException { + HashLookupStoreWriter writer = + factory.createWriter(file, () -> createBloomFiler(enableBloomFilter)); writer.put(toBytes(1), toBytes("foo")); writer.close(); @@ -102,13 +120,14 @@ public class HashLookupStoreFactoryTest { reader.close(); } - @Test - public void testTwoFirstKeyLength() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testTwoFirstKeyLength(boolean enableBloomFilter) throws IOException { int key1 = 1; int key2 = 245; // Write - writeStore(file, new Object[] {key1, key2}, new Object[] {1, 6}); + writeStore(enableBloomFilter, file, new Object[] {key1, key2}, new Object[] {1, 6}); // Read HashLookupStoreReader reader = factory.createReader(file); @@ -121,13 +140,14 @@ public class HashLookupStoreFactoryTest { assertThat(reader.lookup(toBytes(1245))).isNull(); } - @Test - public void testKeyLengthGap() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testKeyLengthGap(boolean enableBf) throws IOException { int key1 = 1; int key2 = 2450; // Write - writeStore(file, new Object[] {key1, key2}, new Object[] {1, 6}); + writeStore(enableBf, file, new Object[] {key1, key2}, new Object[] {1, 6}); // Read HashLookupStoreReader reader = factory.createReader(file); @@ -142,13 +162,14 @@ public class HashLookupStoreFactoryTest { assertThat(reader.lookup(toBytes(2454441))).isNull(); } - @Test - public void testKeyLengthStartTwo() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testKeyLengthStartTwo(boolean enableBf) throws IOException { int key1 = 245; int key2 = 2450; // Write - writeStore(file, new Object[] {key1, key2}, new Object[] {1, 6}); + writeStore(enableBf, file, new Object[] {key1, key2}, new Object[] {1, 6}); // Read HashLookupStoreReader reader = factory.createReader(file); @@ -162,8 +183,9 @@ public class HashLookupStoreFactoryTest { assertThat(reader.lookup(toBytes(2454441))).isNull(); } - @Test - public void testDataOnTwoBuffers() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testDataOnTwoBuffers(boolean enableBf) throws IOException { Object[] keys = new Object[] {1, 2, 3}; Object[] values = new Object[] { @@ -172,23 +194,24 @@ public class HashLookupStoreFactoryTest { int byteSize = toBytes(values[0]).length + toBytes(values[1]).length; - // Write - writeStore(file, keys, values); + int pageSize = MathUtils.roundDownToPowerOf2(byteSize - 100); - // Read factory = new HashLookupStoreFactory( - new CacheManager(MemorySize.ofMebiBytes(1)), - MathUtils.roundDownToPowerOf2(byteSize - 100), - 0.75d); + new CacheManager(MemorySize.ofMebiBytes(1)), pageSize, 0.75d); + + writeStore(factory, enableBf, file, keys, values); + + // Read HashLookupStoreReader reader = factory.createReader(file); for (int i = 0; i < keys.length; i++) { assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i])); } } - @Test - public void testDataSizeOnTwoBuffers() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testDataSizeOnTwoBuffers(boolean enableBf) throws IOException { Object[] keys = new Object[] {1, 2, 3}; Object[] values = new Object[] { @@ -202,68 +225,77 @@ public class HashLookupStoreFactoryTest { VarLengthIntUtils.encodeInt(new DataOutputSerializer(4), b1.length) + VarLengthIntUtils.encodeInt(new DataOutputSerializer(4), b2.length); - // Write - writeStore(file, keys, values); + int pageSize = MathUtils.roundDownToPowerOf2(byteSize + sizeSize + 3); - // Read factory = new HashLookupStoreFactory( - new CacheManager(MemorySize.ofMebiBytes(1)), - MathUtils.roundDownToPowerOf2(byteSize + sizeSize + 3), - 0.75d); + new CacheManager(MemorySize.ofMebiBytes(1)), pageSize, 0.75d); + + writeStore(enableBf, file, keys, values); + HashLookupStoreReader reader = factory.createReader(file); for (int i = 0; i < keys.length; i++) { assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i])); } } - @Test - public void testReadStringToString() throws IOException { - testReadKeyToString(generateStringKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadStringToString(boolean enableBf) throws IOException { + testReadKeyToString(generateStringKeys(100), enableBf); } - @Test - public void testReadIntToString() throws IOException { - testReadKeyToString(generateIntKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadIntToString(boolean enableBf) throws IOException { + testReadKeyToString(generateIntKeys(100), enableBf); } - @Test - public void testReadDoubleToString() throws IOException { - testReadKeyToString(generateDoubleKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadDoubleToString(boolean enableBf) throws IOException { + testReadKeyToString(generateDoubleKeys(100), enableBf); } - @Test - public void testReadLongToString() throws IOException { - testReadKeyToString(generateLongKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadLongToString(boolean enableBf) throws IOException { + testReadKeyToString(generateLongKeys(100), enableBf); } - @Test - public void testReadStringToInt() throws IOException { - testReadKeyToInt(generateStringKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadStringToInt(boolean enableBf) throws IOException { + testReadKeyToInt(generateStringKeys(100), enableBf); } - @Test - public void testReadByteToInt() throws IOException { - testReadKeyToInt(generateByteKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadByteToInt(boolean enableBf) throws IOException { + testReadKeyToInt(generateByteKeys(100), enableBf); } - @Test - public void testReadIntToInt() throws IOException { - testReadKeyToInt(generateIntKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadIntToInt(boolean enableBf) throws IOException { + testReadKeyToInt(generateIntKeys(100), enableBf); } - @Test - public void testReadCompoundToString() throws IOException { - testReadKeyToString(generateCompoundKeys(100)); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadCompoundToString(boolean enableBf) throws IOException { + testReadKeyToString(generateCompoundKeys(100), enableBf); } - @Test - public void testReadCompoundByteToString() throws IOException { - testReadKeyToString(new Object[] {generateCompoundByteKey()}); + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testReadCompoundByteToString(boolean enableBf) throws IOException { + testReadKeyToString(new Object[] {generateCompoundByteKey()}, enableBf); } - @Test - public void testCacheExpiration() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testCacheExpiration(boolean enableBf) throws IOException { int len = 1000; ThreadLocalRandom rnd = ThreadLocalRandom.current(); Object[] keys = new Object[len]; @@ -274,23 +306,25 @@ public class HashLookupStoreFactoryTest { } // Write - writeStore(file, keys, values); + writeStore(enableBf, file, keys, values); // Read - factory = new HashLookupStoreFactory(new CacheManager(new MemorySize(8096)), 1024, 0.75d); + factory = + new HashLookupStoreFactory(new CacheManager(new MemorySize(8096)), pageSize, 0.75d); HashLookupStoreReader reader = factory.createReader(file); for (int i = 0; i < keys.length; i++) { assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i])); } } - @Test - public void testIterate() throws IOException { + @ParameterizedTest + @MethodSource(value = "enableBloomFilter") + public void testIterate(boolean enableBf) throws IOException { Integer[] keys = generateIntKeys(100); String[] values = generateStringData(keys.length, 12); // Write - writeStore(file, keys, values); + writeStore(enableBf, file, keys, values); // Sets Set<Integer> keysSet = new HashSet<>(Arrays.asList(keys)); @@ -317,10 +351,10 @@ public class HashLookupStoreFactoryTest { // UTILITY - private void testReadKeyToString(Object[] keys) throws IOException { + private void testReadKeyToString(Object[] keys, boolean enableBf) throws IOException { // Write Object[] values = generateStringData(keys.length, 10); - writeStore(file, keys, values); + writeStore(enableBf, file, keys, values); // Read HashLookupStoreReader reader = factory.createReader(file); @@ -332,10 +366,10 @@ public class HashLookupStoreFactoryTest { reader.close(); } - private void testReadKeyToInt(Object[] keys) throws IOException { + private void testReadKeyToInt(Object[] keys, boolean enableBf) throws IOException { // Write Integer[] values = generateIntData(keys.length); - writeStore(file, keys, values); + writeStore(enableBf, file, keys, values); // Read HashLookupStoreReader reader = factory.createReader(file); @@ -347,8 +381,20 @@ public class HashLookupStoreFactoryTest { reader.close(); } - private void writeStore(File location, Object[] keys, Object[] values) throws IOException { - HashLookupStoreWriter writer = factory.createWriter(location); + private void writeStore(boolean enableBf, File location, Object[] keys, Object[] values) + throws IOException { + writeStore(factory, enableBf, location, keys, values); + } + + private void writeStore( + HashLookupStoreFactory factory, + boolean enableBf, + File location, + Object[] keys, + Object[] values) + throws IOException { + HashLookupStoreWriter writer = + factory.createWriter(location, () -> createBloomFiler(enableBf)); for (int i = 0; i < keys.length; i++) { writer.put(toBytes(keys[i]), toBytes(values[i])); } diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/CommonTestUtils.java b/paimon-common/src/test/java/org/apache/paimon/utils/CommonTestUtils.java index 2461d8ba4..fb525c26a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/CommonTestUtils.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/CommonTestUtils.java @@ -21,6 +21,7 @@ package org.apache.paimon.utils; import java.lang.reflect.Field; import java.time.Duration; import java.util.Map; +import java.util.Random; import java.util.concurrent.Callable; import java.util.concurrent.TimeoutException; import java.util.function.Supplier; @@ -114,4 +115,14 @@ public class CommonTestUtils { throw new TimeoutException(errorMsg); } } + + public static int[] generateRandomInts(int count) { + Random random = new Random(42); + + int[] nums = new int[count]; + for (int i = 0; i < count; i++) { + nums[i] = random.nextInt(count); + } + return nums; + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java index 24ea935bd..f9c9fa672 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java @@ -26,6 +26,7 @@ import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.lookup.LookupStoreFactory; import org.apache.paimon.lookup.LookupStoreReader; import org.apache.paimon.lookup.LookupStoreWriter; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.options.MemorySize; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.types.RowType; @@ -46,6 +47,7 @@ import java.io.UncheckedIOException; import java.time.Duration; import java.util.Comparator; import java.util.TreeSet; +import java.util.function.Function; import java.util.function.Supplier; import static org.apache.paimon.mergetree.LookupUtils.fileKibiBytes; @@ -62,8 +64,8 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { private final IOFunction<DataFileMeta, RecordReader<KeyValue>> fileReaderFactory; private final Supplier<File> localFileFactory; private final LookupStoreFactory lookupStoreFactory; - private final Cache<String, ContainsFile> containsFiles; + private final Function<Long, BloomFilterBuilder> bfGenerator; public ContainsLevels( Levels levels, @@ -73,7 +75,8 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { Supplier<File> localFileFactory, LookupStoreFactory lookupStoreFactory, Duration fileRetention, - MemorySize maxDiskSize) { + MemorySize maxDiskSize, + Function<Long, BloomFilterBuilder> bfGenerator) { this.levels = levels; this.keyComparator = keyComparator; this.keySerializer = new RowCompactedSerializer(keyType); @@ -89,6 +92,7 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { .executor(MoreExecutors.directExecutor()) .build(); levels.addDropFileCallback(this); + this.bfGenerator = bfGenerator; } @VisibleForTesting @@ -150,7 +154,9 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { if (!localFile.createNewFile()) { throw new IOException("Can not create new file: " + localFile); } - try (LookupStoreWriter kvWriter = lookupStoreFactory.createWriter(localFile); + try (LookupStoreWriter kvWriter = + lookupStoreFactory.createWriter( + localFile, () -> bfGenerator.apply(file.rowCount())); RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) { RecordReader.RecordIterator<KeyValue> batch; KeyValue kv; diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java index c1df76efc..5e2731f3b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java @@ -27,6 +27,7 @@ import org.apache.paimon.io.DataOutputSerializer; import org.apache.paimon.lookup.LookupStoreFactory; import org.apache.paimon.lookup.LookupStoreReader; import org.apache.paimon.lookup.LookupStoreWriter; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.memory.MemorySegment; import org.apache.paimon.options.MemorySize; import org.apache.paimon.reader.RecordReader; @@ -49,6 +50,7 @@ import java.io.UncheckedIOException; import java.time.Duration; import java.util.Comparator; import java.util.TreeSet; +import java.util.function.Function; import java.util.function.Supplier; import static org.apache.paimon.mergetree.LookupUtils.fileKibiBytes; @@ -64,8 +66,8 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { private final IOFunction<DataFileMeta, RecordReader<KeyValue>> fileReaderFactory; private final Supplier<File> localFileFactory; private final LookupStoreFactory lookupStoreFactory; - private final Cache<String, LookupFile> lookupFiles; + private final Function<Long, BloomFilterBuilder> bfGenerator; public LookupLevels( Levels levels, @@ -76,7 +78,8 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { Supplier<File> localFileFactory, LookupStoreFactory lookupStoreFactory, Duration fileRetention, - MemorySize maxDiskSize) { + MemorySize maxDiskSize, + Function<Long, BloomFilterBuilder> bfGenerator) { this.levels = levels; this.keyComparator = keyComparator; this.keySerializer = new RowCompactedSerializer(keyType); @@ -92,6 +95,7 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { .removalListener(this::removalCallback) .executor(MoreExecutors.directExecutor()) .build(); + this.bfGenerator = bfGenerator; levels.addDropFileCallback(this); } @@ -128,12 +132,14 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { @Nullable private KeyValue lookup(InternalRow key, DataFileMeta file) throws IOException { LookupFile lookupFile = lookupFiles.getIfPresent(file.fileName()); + + byte[] keyBytes = keySerializer.serializeToBytes(key); + while (lookupFile == null || lookupFile.isClosed) { lookupFile = createLookupFile(file); lookupFiles.put(file.fileName(), lookupFile); } - byte[] keyBytes = keySerializer.serializeToBytes(key); byte[] valueBytes = lookupFile.get(keyBytes); if (valueBytes == null) { return null; @@ -161,11 +167,14 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { } private LookupFile createLookupFile(DataFileMeta file) throws IOException { + File localFile = localFileFactory.get(); if (!localFile.createNewFile()) { throw new IOException("Can not create new file: " + localFile); } - try (LookupStoreWriter kvWriter = lookupStoreFactory.createWriter(localFile); + try (LookupStoreWriter kvWriter = + lookupStoreFactory.createWriter( + localFile, () -> bfGenerator.apply(file.rowCount())); RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) { DataOutputSerializer valueOut = new DataOutputSerializer(32); RecordReader.RecordIterator<KeyValue> batch; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java index 3c752472f..4b0c57958 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java @@ -34,6 +34,7 @@ import org.apache.paimon.index.IndexMaintainer; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.io.KeyValueFileWriterFactory; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.lookup.hash.HashLookupStoreFactory; import org.apache.paimon.mergetree.ContainsLevels; import org.apache.paimon.mergetree.Levels; @@ -286,7 +287,17 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { options.cachePageSize(), options.toConfiguration().get(CoreOptions.LOOKUP_HASH_LOAD_FACTOR)), options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), - options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE)); + options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + rowCount -> { + if (options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_ENABLED) + && rowCount > 0) { + return BloomFilterBuilder.bfBuilder( + rowCount, + options.toConfiguration() + .get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_FPP)); + } + return null; + }); } private ContainsLevels createContainsLevels( @@ -308,6 +319,16 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { options.cachePageSize(), options.toConfiguration().get(CoreOptions.LOOKUP_HASH_LOAD_FACTOR)), options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), - options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE)); + options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + rowCount -> { + if (options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_ENABLED) + && rowCount > 0) { + return BloomFilterBuilder.bfBuilder( + rowCount, + options.toConfiguration() + .get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_FPP)); + } + return null; + }); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java index e7324b69b..40cacc786 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/query/LocalTableQuery.java @@ -32,6 +32,7 @@ import org.apache.paimon.disk.IOManager; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.lookup.hash.HashLookupStoreFactory; import org.apache.paimon.mergetree.Levels; import org.apache.paimon.mergetree.LookupLevels; @@ -141,7 +142,19 @@ public class LocalTableQuery implements TableQuery { .getPathFile(), hashLookupStoreFactory, options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), - options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE)); + options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + rowCount -> { + if (options.toConfiguration() + .get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_ENABLED) + && rowCount > 0) { + return BloomFilterBuilder.bfBuilder( + rowCount, + options.toConfiguration() + .get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_FPP)); + } + return null; + }); + tableView.computeIfAbsent(partition, k -> new HashMap<>()).put(bucket, lookupLevels); } diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/ContainsLevelsTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/ContainsLevelsTest.java index 4013146ef..cbe3d65f8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/ContainsLevelsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/ContainsLevelsTest.java @@ -31,6 +31,7 @@ import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.io.KeyValueFileWriterFactory; import org.apache.paimon.io.RollingFileWriter; import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.lookup.hash.HashLookupStoreFactory; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; @@ -158,7 +159,7 @@ public class ContainsLevelsTest { files.add(newFile(1, kvs.toArray(new KeyValue[0]))); } Levels levels = new Levels(comparator, files, 1); - ContainsLevels lookupLevels = createContainsLevels(levels, MemorySize.ofKibiBytes(50)); + ContainsLevels lookupLevels = createContainsLevels(levels, MemorySize.ofKibiBytes(60)); for (int i = 0; i < fileNum * recordInFile; i++) { assertThat(lookupLevels.contains(row(i), 1)).isTrue(); @@ -187,7 +188,8 @@ public class ContainsLevelsTest { () -> new File(tempDir.toFile(), LOOKUP_FILE_PREFIX + UUID.randomUUID()), new HashLookupStoreFactory(new CacheManager(MemorySize.ofMebiBytes(1)), 2048, 0.75), Duration.ofHours(1), - maxDiskSize); + maxDiskSize, + rowCount -> BloomFilterBuilder.bfBuilder(rowCount, 0.01)); } private KeyValue kv(int key, int value) { diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java index 3a8881656..1acb26e42 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/LookupLevelsTest.java @@ -31,6 +31,7 @@ import org.apache.paimon.io.KeyValueFileReaderFactory; import org.apache.paimon.io.KeyValueFileWriterFactory; import org.apache.paimon.io.RollingFileWriter; import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.bloom.BloomFilterBuilder; import org.apache.paimon.lookup.hash.HashLookupStoreFactory; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; @@ -227,7 +228,8 @@ public class LookupLevelsTest { () -> new File(tempDir.toFile(), LOOKUP_FILE_PREFIX + UUID.randomUUID()), new HashLookupStoreFactory(new CacheManager(MemorySize.ofMebiBytes(1)), 2048, 0.75), Duration.ofHours(1), - maxDiskSize); + maxDiskSize, + rowCount -> BloomFilterBuilder.bfBuilder(rowCount, 0.05)); } private KeyValue kv(int key, int value) {
