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 547093b6b8cb697a73628692d7e5bd309df5224c Author: Jingsong <[email protected]> AuthorDate: Wed Jan 24 17:08:55 2024 +0800 [core] Refactor BloomFilter and FileBasedBloomFilter --- .../benchmark/LookupBloomFilterBenchmark.java | 9 +- .../paimon/io/cache/CachedRandomInputView.java | 80 +++----------- .../apache/paimon/lookup/LookupStoreFactory.java | 25 ++++- .../paimon/lookup/bloom/BloomFilterBuilder.java | 71 ------------ .../paimon/lookup/bloom/BloomFilterTester.java | 71 ------------ .../paimon/lookup/hash/HashLookupStoreFactory.java | 9 +- .../paimon/lookup/hash/HashLookupStoreReader.java | 8 +- .../paimon/lookup/hash/HashLookupStoreWriter.java | 31 +++--- .../paimon/{lookup/bloom => utils}/BitSet.java | 48 ++++++--- .../{lookup/bloom => utils}/BloomFilter.java | 120 +++++++++++---------- .../apache/paimon/utils/FileBasedBloomFilter.java | 71 ++++++++++++ .../paimon/io/cache/CachedRandomInputViewTest.java | 18 +--- .../lookup/bloom/BloomFilterBuilderTest.java | 67 ------------ .../lookup/hash/HashLookupStoreFactoryTest.java | 13 ++- .../paimon/{lookup/bloom => utils}/BitSetTest.java | 4 +- .../{lookup/bloom => utils}/BloomFilterTest.java | 43 ++++++-- .../FileBasedBloomFilterTest.java} | 24 ++--- .../apache/paimon/mergetree/ContainsLevels.java | 8 +- .../org/apache/paimon/mergetree/LookupLevels.java | 9 +- .../paimon/operation/KeyValueFileStoreWrite.java | 43 +++----- .../apache/paimon/table/query/LocalTableQuery.java | 20 ++-- .../paimon/mergetree/ContainsLevelsTest.java | 4 +- .../apache/paimon/mergetree/LookupLevelsTest.java | 4 +- 23 files changed, 320 insertions(+), 480 deletions(-) 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 index 92c00c545..5b5ea766e 100644 --- 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 @@ -21,11 +21,11 @@ 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.apache.paimon.utils.BloomFilter; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -105,8 +105,7 @@ public class LookupBloomFilterBenchmark { }); HashLookupStoreReader reader2 = - writeData( - BloomFilterBuilder.bfBuilder(inputs.length, 0.05), inputs, valueLength); + writeData(BloomFilter.builder(inputs.length, 0.05), inputs, valueLength); benchmark.addCase( String.format("bf-enabled-%dB-value", valueLength), @@ -126,7 +125,7 @@ public class LookupBloomFilterBenchmark { } private HashLookupStoreReader writeData( - BloomFilterBuilder filter, byte[][] inputs, int valueLength) throws IOException { + BloomFilter.Builder filter, byte[][] inputs, int valueLength) throws IOException { byte[] value = new byte[valueLength]; Arrays.fill(value, (byte) 1); HashLookupStoreFactory factory = @@ -134,7 +133,7 @@ public class LookupBloomFilterBenchmark { new CacheManager(MemorySize.ofMebiBytes(10)), 16 * 1024, 0.75); File file = new File(tempDir.toFile(), UUID.randomUUID().toString()); - HashLookupStoreWriter writer = factory.createWriter(file, () -> filter); + HashLookupStoreWriter writer = factory.createWriter(file, filter); for (byte[] input : inputs) { writer.put(input, value); } 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 739f1ce38..53acad5b8 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,8 +33,6 @@ 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 @@ -46,7 +44,7 @@ public class CachedRandomInputView extends AbstractPagedInputView private final RandomAccessFile file; private final long fileLength; private final CacheManager cacheManager; - private final Map<CacheKey, MemorySegment> segments; + private final Map<Integer, MemorySegment> segments; private final int segmentSize; private final int segmentSizeBits; private final int segmentSizeMask; @@ -75,13 +73,12 @@ public class CachedRandomInputView extends AbstractPagedInputView } private MemorySegment getCurrentPage() { - long position = segmentIndexToPosition(currentSegmentIndex); - int length = (int) Math.min(segmentSize, fileLength - position); - CacheKey cacheKey = new CacheKey(position, length); - MemorySegment segment = segments.get(cacheKey); + MemorySegment segment = segments.get(currentSegmentIndex); if (segment == null) { - segment = cacheManager.getPage(file, position, length, this::invalidPage); - segments.put(cacheKey, segment); + long offset = segmentIndexToPosition(currentSegmentIndex); + int length = (int) Math.min(segmentSize, fileLength - offset); + segment = cacheManager.getPage(file, offset, length, this::invalidPage); + segments.put(currentSegmentIndex, segment); } return segment; } @@ -102,46 +99,18 @@ public class CachedRandomInputView extends AbstractPagedInputView } private void invalidPage(long position, int length) { - 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; + segments.remove(positionToSegmentIndex(position)); } @Override public void close() throws IOException { // copy out to avoid ConcurrentModificationException - List<CacheKey> pages = new ArrayList<>(segments.keySet()); + List<Integer> pages = new ArrayList<>(segments.keySet()); pages.forEach( page -> { - cacheManager.invalidPage(file, page.readOffset, page.readLength); + long offset = segmentIndexToPosition(page); + int length = (int) Math.min(segmentSize, fileLength - offset); + cacheManager.invalidPage(file, offset, length); }); file.close(); @@ -155,30 +124,7 @@ public class CachedRandomInputView extends AbstractPagedInputView 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); - } + public RandomAccessFile file() { + return file; } } 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 2143bdd9f..cde93fef0 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,11 +18,15 @@ package org.apache.paimon.lookup; -import org.apache.paimon.lookup.bloom.BloomFilterBuilder; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.utils.BloomFilter; + +import javax.annotation.Nullable; import java.io.File; import java.io.IOException; -import java.util.function.Supplier; +import java.util.function.Function; /** * A key-value store for lookup, key-value store should be single binary file written once and ready @@ -35,8 +39,23 @@ import java.util.function.Supplier; */ public interface LookupStoreFactory { - LookupStoreWriter createWriter(File file, Supplier<BloomFilterBuilder> bfProvider) + LookupStoreWriter createWriter(File file, @Nullable BloomFilter.Builder bloomFilter) throws IOException; LookupStoreReader createReader(File file) throws IOException; + + static Function<Long, BloomFilter.Builder> bfGenerator(Options options) { + Function<Long, BloomFilter.Builder> bfGenerator = rowCount -> null; + if (options.get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_ENABLED)) { + double bfFpp = options.get(CoreOptions.LOOKUP_CACHE_BLOOM_FILTER_FPP); + bfGenerator = + rowCount -> { + if (rowCount > 0) { + return BloomFilter.builder(rowCount, bfFpp); + } + return null; + }; + } + return bfGenerator; + } } 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 deleted file mode 100644 index 856010e4d..000000000 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterBuilder.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * - * 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 deleted file mode 100644 index 3f4e67d13..000000000 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilterTester.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * - * 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 31828a8dd..e34e97a4e 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,11 +20,12 @@ 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 org.apache.paimon.utils.BloomFilter; + +import javax.annotation.Nullable; 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 { @@ -45,8 +46,8 @@ public class HashLookupStoreFactory implements LookupStoreFactory { } @Override - public HashLookupStoreWriter createWriter(File file, Supplier<BloomFilterBuilder> bfProvider) + public HashLookupStoreWriter createWriter(File file, @Nullable BloomFilter.Builder bloomFilter) throws IOException { - return new HashLookupStoreWriter(loadFactor, file, bfProvider); + return new HashLookupStoreWriter(loadFactor, file, bloomFilter); } } 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 160648eed..6fc380691 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,7 +21,7 @@ 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.FileBasedBloomFilter; import org.apache.paimon.utils.MurmurHashUtils; import org.apache.paimon.utils.VarLengthIntUtils; @@ -64,7 +64,7 @@ public class HashLookupStoreReader private CachedRandomInputView inputView; // Buffers private final byte[] slotBuffer; - @Nullable private BloomFilterTester bloomFilter; + @Nullable private FileBasedBloomFilter bloomFilter; HashLookupStoreReader(CacheManager cacheManager, int cachePageSize, File file) throws IOException { @@ -104,7 +104,9 @@ public class HashLookupStoreReader int bfStartIndex = dataInputStream.readInt(); // Skip bloom filter buffers dataInputStream.skipBytes(bfBytes); - bloomFilter = new BloomFilterTester(recordCount, bfStartIndex, bfBytes, inputView); + bloomFilter = + new FileBasedBloomFilter( + inputView.file(), cacheManager, recordCount, bfStartIndex, bfBytes); } // Read offset counts and keys 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 7fcd9f696..972066643 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,7 +19,7 @@ package org.apache.paimon.lookup.hash; import org.apache.paimon.lookup.LookupStoreWriter; -import org.apache.paimon.lookup.bloom.BloomFilterBuilder; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.MurmurHashUtils; import org.apache.paimon.utils.VarLengthIntUtils; @@ -45,7 +45,6 @@ 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 { @@ -80,9 +79,9 @@ public class HashLookupStoreWriter implements LookupStoreWriter { private int valueCount; // Number of collisions private int collisions; - @Nullable private final BloomFilterBuilder bloomFilter; + @Nullable private final BloomFilter.Builder bloomFilter; - HashLookupStoreWriter(double loadFactor, File file, Supplier<BloomFilterBuilder> bfProvider) + HashLookupStoreWriter(double loadFactor, File file, @Nullable BloomFilter.Builder bloomFilter) throws IOException { this.loadFactor = loadFactor; if (loadFactor <= 0.0 || loadFactor >= 1.0) { @@ -90,21 +89,21 @@ public class HashLookupStoreWriter implements LookupStoreWriter { "Illegal load factor = " + loadFactor + ", should be between 0.0 and 1.0."); } - tempFolder = new File(file.getParentFile(), UUID.randomUUID().toString()); + this.tempFolder = new File(file.getParentFile(), UUID.randomUUID().toString()); if (!tempFolder.mkdir()) { throw new IOException("Can not create temp folder: " + tempFolder); } - outputStream = new BufferedOutputStream(new FileOutputStream(file)); - indexStreams = new DataOutputStream[0]; - dataStreams = new DataOutputStream[0]; - indexFiles = new File[0]; - dataFiles = new File[0]; - lastValues = new byte[0][]; - lastValuesLength = new int[0]; - dataLengths = new long[0]; - maxOffsetLengths = new int[0]; - keyCounts = new int[0]; - bloomFilter = bfProvider.get(); + this.outputStream = new BufferedOutputStream(new FileOutputStream(file)); + this.indexStreams = new DataOutputStream[0]; + this.dataStreams = new DataOutputStream[0]; + this.indexFiles = new File[0]; + this.dataFiles = new File[0]; + this.lastValues = new byte[0][]; + this.lastValuesLength = new int[0]; + this.dataLengths = new long[0]; + this.maxOffsetLengths = new int[0]; + this.keyCounts = new int[0]; + this.bloomFilter = bloomFilter; } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java b/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java similarity index 72% rename from paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java rename to paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java index 581158b4d..1ba658ec8 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BitSet.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/BitSet.java @@ -18,13 +18,15 @@ * */ -package org.apache.paimon.lookup.bloom; +package org.apache.paimon.utils; import org.apache.paimon.memory.MemorySegment; -import org.apache.paimon.utils.Preconditions; -/** BitSet util. Copied from Apache Flink Project. */ +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** BitSet based on {@link MemorySegment}. */ public class BitSet { + private static final int BYTE_INDEX_MASK = 0x00000007; private MemorySegment memorySegment; @@ -33,34 +35,42 @@ public class BitSet { private int offset; // The BitSet byte size. - private int byteLength; + private final int byteLength; // The BitSet bit size. - private int bitLength; + private final int bitLength; public BitSet(int byteSize) { - Preconditions.checkArgument(byteSize > 0, "bits size should be greater than 0."); + 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( + checkArgument(memorySegment != null, "MemorySegment can not be null."); + checkArgument(offset >= 0, "Offset should be positive integer."); + checkArgument( offset + byteLength <= memorySegment.size(), "Could not set MemorySegment, the remain buffers is not enough."); this.memorySegment = memorySegment; this.offset = offset; } + public void unsetMemorySegment() { + this.memorySegment = null; + } + + public MemorySegment getMemorySegment() { + return this.memorySegment; + } + /** * Sets the bit at specified index. * * @param index - position */ public void set(int index) { - Preconditions.checkArgument(index < bitLength && index >= 0); + checkArgument(index < bitLength && index >= 0); int byteIndex = index >>> 3; byte current = memorySegment.get(offset + byteIndex); @@ -75,7 +85,7 @@ public class BitSet { * @return - value at the bit position */ public boolean get(int index) { - Preconditions.checkArgument(index < bitLength && index >= 0); + checkArgument(index < bitLength && index >= 0); int byteIndex = index >>> 3; byte current = memorySegment.get(offset + byteIndex); @@ -102,11 +112,15 @@ public class BitSet { @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(); + return "BitSet:\n" + + "\tMemorySegment:" + + memorySegment.size() + + "\n" + + "\tOffset:" + + offset + + "\n" + + "\tLength:" + + byteLength + + "\n"; } } diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java similarity index 51% rename from paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java rename to paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java index cfcb7e67a..d4fee8ee5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/bloom/BloomFilter.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java @@ -18,31 +18,14 @@ * */ -package org.apache.paimon.lookup.bloom; +package org.apache.paimon.utils; +import org.apache.paimon.annotation.VisibleForTesting; 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 - */ +/** Bloom filter based on one memory segment. */ public class BloomFilter { protected BitSet bitSet; @@ -52,57 +35,43 @@ public class BloomFilter { public BloomFilter(long expectedEntries, int byteSize) { checkArgument(expectedEntries > 0, "expectedEntries should be > 0"); this.expectedEntries = expectedEntries; - this.numHashFunctions = optimalNumOfHashFunctions(expectedEntries, byteSize << 3); + this.numHashFunctions = optimalNumOfHashFunctions(expectedEntries, (long) byteSize << 3); this.bitSet = new BitSet(byteSize); } - public void setBitsLocation(MemorySegment memorySegment, int offset) { + public void setMemorySegment(MemorySegment memorySegment, int offset) { this.bitSet.setMemorySegment(memorySegment, offset); } + public void unsetMemorySegment() { + this.bitSet.unsetMemorySegment(); + } + + public MemorySegment getMemorySegment() { + return this.bitSet.getMemorySegment(); + } + /** * 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; + return (int) (-inputEntries * Math.log(fpp) / (Math.log(2) * Math.log(2))); } /** * 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; + public void addHash(int hash1) { + int hash2 = hash1 >>> 16; for (int i = 1; i <= numHashFunctions; i++) { int combinedHash = hash1 + (i * hash2); @@ -115,9 +84,8 @@ public class BloomFilter { } } - public boolean testHash(int hash32) { - int hash1 = hash32; - int hash2 = hash32 >>> 16; + public boolean testHash(int hash1) { + int hash2 = hash1 >>> 16; for (int i = 1; i <= numHashFunctions; i++) { int combinedHash = hash1 + (i * hash2); @@ -139,10 +107,52 @@ public class BloomFilter { @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(); + return "BloomFilter:\n" + "\thash function number:" + numHashFunctions + "\n" + bitSet; + } + + public static Builder builder(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 Builder(MemorySegment.wrap(new byte[numBytes]), expectedRow); + } + + /** Bloom filter based on one memory segment. */ + public static class Builder { + + private final MemorySegment buffer; + private final BloomFilter filter; + private final long numRecords; + + Builder(MemorySegment buffer, long numRecords) { + this.buffer = buffer; + this.filter = new BloomFilter(numRecords, buffer.size()); + filter.setMemorySegment(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; + } + + @VisibleForTesting + public BloomFilter getFilter() { + return filter; + } } } 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 new file mode 100644 index 000000000..541185e7b --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.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.utils; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.memory.MemorySegment; + +import java.io.RandomAccessFile; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Util to apply a built bloom filter . */ +public class FileBasedBloomFilter { + + private final RandomAccessFile file; + private final CacheManager cacheManager; + private final BloomFilter filter; + private final long readOffset; + private final int readLength; + + public FileBasedBloomFilter( + RandomAccessFile file, + CacheManager cacheManager, + long numRecords, + long readOffset, + int readLength) { + this.file = file; + this.cacheManager = cacheManager; + checkArgument(numRecords >= 0); + this.filter = new BloomFilter(numRecords, readLength); + this.readOffset = readOffset; + this.readLength = readLength; + } + + public boolean testHash(int hash) { + if (filter.getMemorySegment() == null) { + MemorySegment segment = + cacheManager.getPage( + file, + readOffset, + readLength, + (position, length) -> filter.unsetMemorySegment()); + filter.setMemorySegment(segment, 0); + } + return filter.testHash(hash); + } + + @VisibleForTesting + BloomFilter bloomFilter() { + return filter; + } +} 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 68cf28674..c00dfdf27 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,9 +65,8 @@ public class CachedRandomInputViewTest { } File file = writeFile(bytes); - int pageSize = 1024; CacheManager cacheManager = new CacheManager(MemorySize.ofKibiBytes(128)); - CachedRandomInputView view = new CachedRandomInputView(file, cacheManager, pageSize); + CachedRandomInputView view = new CachedRandomInputView(file, cacheManager, 1024); // read first one // this assertThatCode check the ConcurrentModificationException is not threw. @@ -94,21 +93,6 @@ 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/BloomFilterBuilderTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterBuilderTest.java deleted file mode 100644 index f4c649ae1..000000000 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterBuilderTest.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * - * 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/hash/HashLookupStoreFactoryTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java index d1e48a4da..a2633ff36 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,8 +24,8 @@ 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.BloomFilter; import org.apache.paimon.utils.MathUtils; import org.apache.paimon.utils.VarLengthIntUtils; @@ -76,11 +76,11 @@ public class HashLookupStoreFactoryTest { return new Object[] {false, true}; } - private BloomFilterBuilder createBloomFiler(boolean enabled) { + private BloomFilter.Builder createBloomFiler(boolean enabled) { if (!enabled) { return null; } - return BloomFilterBuilder.bfBuilder(100, 0.01); + return BloomFilter.builder(100, 0.01); } private byte[] toBytes(Object o) { @@ -95,7 +95,7 @@ public class HashLookupStoreFactoryTest { @MethodSource(value = "enableBloomFilter") public void testEmpty(boolean enableBloomFilter) throws IOException { HashLookupStoreWriter writer = - factory.createWriter(file, () -> createBloomFiler(enableBloomFilter)); + factory.createWriter(file, createBloomFiler(enableBloomFilter)); writer.close(); assertThat(file.exists()).isTrue(); @@ -111,7 +111,7 @@ public class HashLookupStoreFactoryTest { @MethodSource(value = "enableBloomFilter") public void testOneKey(boolean enableBloomFilter) throws IOException { HashLookupStoreWriter writer = - factory.createWriter(file, () -> createBloomFiler(enableBloomFilter)); + factory.createWriter(file, createBloomFiler(enableBloomFilter)); writer.put(toBytes(1), toBytes("foo")); writer.close(); @@ -393,8 +393,7 @@ public class HashLookupStoreFactoryTest { Object[] keys, Object[] values) throws IOException { - HashLookupStoreWriter writer = - factory.createWriter(location, () -> createBloomFiler(enableBf)); + 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/lookup/bloom/BitSetTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/BitSetTest.java similarity index 95% rename from paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BitSetTest.java rename to paimon-common/src/test/java/org/apache/paimon/utils/BitSetTest.java index 8acb93cde..51b36936a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BitSetTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/BitSetTest.java @@ -18,7 +18,7 @@ * */ -package org.apache.paimon.lookup.bloom; +package org.apache.paimon.utils; import org.apache.paimon.memory.MemorySegment; @@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test; import java.util.stream.IntStream; -/** Test for {@link BitSet}. */ +/** Test for {@link org.apache.paimon.utils.BitSet}. */ public class BitSetTest { @Test diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java similarity index 76% rename from paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java rename to paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java index 7b3e7afc0..518b52b30 100644 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/BloomFilterTest.java @@ -18,7 +18,7 @@ * */ -package org.apache.paimon.lookup.bloom; +package org.apache.paimon.utils; import org.apache.paimon.memory.MemorySegment; @@ -30,9 +30,40 @@ import java.util.Arrays; import static org.apache.paimon.utils.CommonTestUtils.generateRandomInts; import static org.junit.jupiter.api.Assertions.assertEquals; -/** Test for {@link BloomFilter}. */ +/** Test for {@link org.apache.paimon.utils.BloomFilter}. */ public class BloomFilterTest { + @Test + public void testOneSegmentBuilder() { + BloomFilter.Builder builder = BloomFilter.builder(100, 0.01); + int[] inputs = generateRandomInts(100); + for (int input : inputs) { + builder.addHash(Integer.hashCode(input)); + } + + for (int input : inputs) { + Assertions.assertThat(builder.testHash(Integer.hashCode(input))).isTrue(); + } + } + + @Test + public void testEstimatedHashFunctions() { + Assertions.assertThat(BloomFilter.builder(1000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat(BloomFilter.builder(10_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat(BloomFilter.builder(100_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat(BloomFilter.builder(100_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat(BloomFilter.builder(100_000, 0.05).getFilter().numHashFunctions) + .isEqualTo(4); + Assertions.assertThat(BloomFilter.builder(1_000_000, 0.01).getFilter().numHashFunctions) + .isEqualTo(7); + Assertions.assertThat(BloomFilter.builder(1_000_000, 0.05).getFilter().numHashFunctions) + .isEqualTo(4); + } + @Test public void testBloomNumBits() { Assertions.assertThat(BloomFilter.optimalNumOfBits(0, 0)).isEqualTo(0); @@ -79,7 +110,7 @@ public class BloomFilterTest { // segment1 MemorySegment segment1 = MemorySegment.wrap(new byte[1024]); - filter.setBitsLocation(segment1, 0); + filter.setMemorySegment(segment1, 0); int[] inputs1 = generateRandomInts(100); Arrays.stream(inputs1).forEach(i -> filter.addHash(Integer.hashCode(i))); Arrays.stream(inputs1) @@ -87,14 +118,14 @@ public class BloomFilterTest { // segments 2 MemorySegment segment2 = MemorySegment.wrap(new byte[1024]); - filter.setBitsLocation(segment2, 0); + filter.setMemorySegment(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); + filter.setMemorySegment(segment1, 0); Arrays.stream(inputs1) .forEach(i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue()); @@ -104,7 +135,7 @@ public class BloomFilterTest { .forEach( i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isFalse()); // clear segment2 - filter.setBitsLocation(segment2, 0); + filter.setMemorySegment(segment2, 0); filter.reset(); Arrays.stream(inputs2) .forEach( diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java similarity index 75% rename from paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java rename to paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java index 15a4dd6ce..9a71544c0 100644 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/bloom/BloomFilterTesterTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/FileBasedBloomFilterTest.java @@ -18,13 +18,11 @@ * */ -package org.apache.paimon.lookup.bloom; +package org.apache.paimon.utils; 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; @@ -32,35 +30,35 @@ import org.junit.jupiter.api.io.TempDir; import java.io.File; import java.io.IOException; +import java.io.RandomAccessFile; 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 { +/** Test for {@link FileBasedBloomFilter}. */ +public class FileBasedBloomFilterTest { @TempDir Path tempDir; @Test public void testProbe() throws IOException { - MemorySegment segment = MemorySegment.wrap(new byte[1000]); - BloomFilterBuilder filter = new BloomFilterBuilder(segment, 100); + BloomFilter.Builder builder = new BloomFilter.Builder(segment, 100); int[] inputs = CommonTestUtils.generateRandomInts(100); - Arrays.stream(inputs).forEach(i -> filter.addHash(Integer.hashCode(i))); + Arrays.stream(inputs).forEach(i -> builder.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)); + FileBasedBloomFilter filter = + new FileBasedBloomFilter( + new RandomAccessFile(file, "r"), cacheManager, 100, 0, 1000); Arrays.stream(inputs) - .forEach(i -> Assertions.assertThat(tester.testHash(Integer.hashCode(i))).isTrue()); + .forEach(i -> Assertions.assertThat(filter.testHash(Integer.hashCode(i))).isTrue()); cacheManager.cache().invalidateAll(); - Assertions.assertThat(tester.getSegments()[0]).isNull(); + Assertions.assertThat(filter.bloomFilter().getMemorySegment()).isNull(); } private File writeFile(byte[] bytes) throws IOException { 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 f9c9fa672..34f9947ba 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,10 +26,10 @@ 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; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.FileIOUtils; import org.apache.paimon.utils.IOFunction; @@ -65,7 +65,7 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { private final Supplier<File> localFileFactory; private final LookupStoreFactory lookupStoreFactory; private final Cache<String, ContainsFile> containsFiles; - private final Function<Long, BloomFilterBuilder> bfGenerator; + private final Function<Long, BloomFilter.Builder> bfGenerator; public ContainsLevels( Levels levels, @@ -76,7 +76,7 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { LookupStoreFactory lookupStoreFactory, Duration fileRetention, MemorySize maxDiskSize, - Function<Long, BloomFilterBuilder> bfGenerator) { + Function<Long, BloomFilter.Builder> bfGenerator) { this.levels = levels; this.keyComparator = keyComparator; this.keySerializer = new RowCompactedSerializer(keyType); @@ -156,7 +156,7 @@ public class ContainsLevels implements Levels.DropFileCallback, Closeable { } try (LookupStoreWriter kvWriter = lookupStoreFactory.createWriter( - localFile, () -> bfGenerator.apply(file.rowCount())); + 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 5e2731f3b..9f8b7a196 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,12 +27,12 @@ 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; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.FileIOUtils; import org.apache.paimon.utils.IOFunction; @@ -67,7 +67,7 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { private final Supplier<File> localFileFactory; private final LookupStoreFactory lookupStoreFactory; private final Cache<String, LookupFile> lookupFiles; - private final Function<Long, BloomFilterBuilder> bfGenerator; + private final Function<Long, BloomFilter.Builder> bfGenerator; public LookupLevels( Levels levels, @@ -79,7 +79,7 @@ public class LookupLevels implements Levels.DropFileCallback, Closeable { LookupStoreFactory lookupStoreFactory, Duration fileRetention, MemorySize maxDiskSize, - Function<Long, BloomFilterBuilder> bfGenerator) { + Function<Long, BloomFilter.Builder> bfGenerator) { this.levels = levels; this.keyComparator = keyComparator; this.keySerializer = new RowCompactedSerializer(keyType); @@ -167,14 +167,13 @@ 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, () -> bfGenerator.apply(file.rowCount())); + 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 4b0c57958..53ae886a4 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,7 +34,6 @@ 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; @@ -51,6 +50,7 @@ import org.apache.paimon.mergetree.compact.MergeFunctionFactory; import org.apache.paimon.mergetree.compact.MergeTreeCompactManager; import org.apache.paimon.mergetree.compact.MergeTreeCompactRewriter; import org.apache.paimon.mergetree.compact.UniversalCompaction; +import org.apache.paimon.options.Options; import org.apache.paimon.schema.KeyValueFieldsExtractor; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.types.RowType; @@ -70,6 +70,7 @@ import java.util.concurrent.ExecutorService; import java.util.function.Supplier; import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber; +import static org.apache.paimon.lookup.LookupStoreFactory.bfGenerator; /** {@link FileStoreWrite} for {@link KeyValueFileStore}. */ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { @@ -273,6 +274,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { throw new RuntimeException( "Can not use lookup, there is no temp disk directory to use."); } + Options options = this.options.toConfiguration(); return new LookupLevels( levels, keyComparatorSupplier.get(), @@ -284,20 +286,11 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { () -> ioManager.createChannel().getPathFile(), new HashLookupStoreFactory( cacheManager, - 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), - 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; - }); + this.options.cachePageSize(), + options.get(CoreOptions.LOOKUP_HASH_LOAD_FACTOR)), + options.get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), + options.get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + bfGenerator(options)); } private ContainsLevels createContainsLevels( @@ -306,6 +299,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { throw new RuntimeException( "Can not use lookup, there is no temp disk directory to use."); } + Options options = this.options.toConfiguration(); return new ContainsLevels( levels, keyComparatorSupplier.get(), @@ -316,19 +310,10 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> { () -> ioManager.createChannel().getPathFile(), new HashLookupStoreFactory( cacheManager, - 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), - 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; - }); + this.options.cachePageSize(), + options.get(CoreOptions.LOOKUP_HASH_LOAD_FACTOR)), + options.get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), + options.get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + bfGenerator(options)); } } 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 40cacc786..2009932c5 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,10 +32,10 @@ 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; +import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.utils.KeyComparatorSupplier; import org.apache.paimon.utils.Preconditions; @@ -50,6 +50,7 @@ import java.util.Map; import java.util.function.Supplier; import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE; +import static org.apache.paimon.lookup.LookupStoreFactory.bfGenerator; /** Implementation for {@link TableQuery} for caching data and file in local. */ public class LocalTableQuery implements TableQuery { @@ -124,6 +125,7 @@ public class LocalTableQuery implements TableQuery { private void newLookupLevels(BinaryRow partition, int bucket, List<DataFileMeta> dataFiles) { Levels levels = new Levels(keyComparatorSupplier.get(), dataFiles, options.numLevels()); KeyValueFileReaderFactory factory = readerFactoryBuilder.build(partition, bucket); + Options options = this.options.toConfiguration(); LookupLevels lookupLevels = new LookupLevels( levels, @@ -141,19 +143,9 @@ public class LocalTableQuery implements TableQuery { .createChannel() .getPathFile(), hashLookupStoreFactory, - options.toConfiguration().get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), - 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; - }); + options.get(CoreOptions.LOOKUP_CACHE_FILE_RETENTION), + options.get(CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE), + bfGenerator(options)); 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 cbe3d65f8..e4d4e83b0 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,7 +31,6 @@ 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; @@ -43,6 +42,7 @@ import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.FileStorePathFactory; import org.junit.jupiter.api.Test; @@ -189,7 +189,7 @@ public class ContainsLevelsTest { new HashLookupStoreFactory(new CacheManager(MemorySize.ofMebiBytes(1)), 2048, 0.75), Duration.ofHours(1), maxDiskSize, - rowCount -> BloomFilterBuilder.bfBuilder(rowCount, 0.01)); + rowCount -> BloomFilter.builder(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 1acb26e42..836f8f874 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,7 +31,6 @@ 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; @@ -43,6 +42,7 @@ import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.FileStorePathFactory; import org.junit.jupiter.api.Test; @@ -229,7 +229,7 @@ public class LookupLevelsTest { new HashLookupStoreFactory(new CacheManager(MemorySize.ofMebiBytes(1)), 2048, 0.75), Duration.ofHours(1), maxDiskSize, - rowCount -> BloomFilterBuilder.bfBuilder(rowCount, 0.05)); + rowCount -> BloomFilter.builder(rowCount, 0.05)); } private KeyValue kv(int key, int value) {
