This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit eaf7c8744b0aa4e76c05e495d8f9bff00c936060 Author: Jingsong Lee <[email protected]> AuthorDate: Fri Sep 18 11:30:05 2026 +0800 [core][python] Add optional Bloom filters to BTree indexes (#9946) --- docs/docs/multimodal-table/global-index/btree.mdx | 1 + .../globalindex/btree/BTreeGlobalIndexer.java | 8 + .../globalindex/btree/BTreeIndexOptions.java | 7 + .../paimon/globalindex/btree/BTreeIndexReader.java | 15 +- .../paimon/globalindex/btree/BTreeIndexWriter.java | 17 +- .../globalindex/btree/BTreeBloomFilterTest.java | 239 +++++++++++++++++++++ .../globalindex/btree/BTreeIndexOptionsTest.java | 5 + .../globalindex/btree/BTreeThreadSafetyTest.java | 1 + .../pypaimon/common/options/core_options.py | 12 ++ .../pypaimon/globalindex/btree/bloom_filter.py | 151 +++++++++++++ .../globalindex/btree/btree_index_reader.py | 17 +- .../globalindex/btree/btree_index_writer.py | 34 ++- .../pypaimon/globalindex/btree/sst_file_reader.py | 42 ++++ .../pypaimon/globalindex/create_global_index.py | 2 + .../pypaimon/tests/btree_bloom_filter_test.py | 176 +++++++++++++++ .../pypaimon/tests/global_index_build_test.py | 19 +- 16 files changed, 732 insertions(+), 14 deletions(-) diff --git a/docs/docs/multimodal-table/global-index/btree.mdx b/docs/docs/multimodal-table/global-index/btree.mdx index e685e1fcb9..e16ef839d4 100644 --- a/docs/docs/multimodal-table/global-index/btree.mdx +++ b/docs/docs/multimodal-table/global-index/btree.mdx @@ -152,6 +152,7 @@ print(pa_table) | `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree, Bitmap, and Multivalue builds. | | `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building sorted global indexes. | | `btree-index.block-size` | `64 kb` | Block size used by BTree index files. | +| `btree-index.bloom-filter.enabled` | `false` | Whether to write a Bloom filter to accelerate BTree equality and `IN` lookups. | | `btree-index.cache-size` | `128 mb` | Cache size used by BTree index readers. | | `btree-index.fallback-scan-max-size` | `256 mb` | Maximum total size of candidate BTree global index files to allow fallback index scans. Set to `0 b` to disable fallback scans. | | `btree-index.compression` | `none` | Compression algorithm used by BTree index blocks. | diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java index 660423c0cc..0b4c5b4763 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java @@ -31,6 +31,7 @@ import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.options.Options; import org.apache.paimon.types.DataField; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.LazyField; import java.io.IOException; @@ -62,6 +63,8 @@ import java.util.concurrent.ExecutorService; */ public class BTreeGlobalIndexer implements SortedGlobalIndexer { + private static final double BLOOM_FILTER_FPP = 0.05; + private final KeySerializer keySerializer; private final GlobalIndexKeyExtractor keyExtractor; private final Options options; @@ -96,10 +99,15 @@ public class BTreeGlobalIndexer implements SortedGlobalIndexer { new CompressOptions( options.get(BTreeIndexOptions.BTREE_INDEX_COMPRESSION), options.get(BTreeIndexOptions.BTREE_INDEX_COMPRESSION_LEVEL)); + BloomFilter.Builder bloomFilterBuilder = + options.get(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED) + ? BloomFilter.dynamicBuilder(BLOOM_FILTER_FPP) + : null; return new BTreeIndexWriter( fileWriter, keySerializer, (int) blockSize, + bloomFilterBuilder, BlockCompressionFactory.create(compressOptions)); } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java index d46ef0d92a..e06704c561 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java @@ -43,6 +43,13 @@ public class BTreeIndexOptions { .defaultValue(MemorySize.ofKibiBytes(64)) .withDescription("The block size to use for BTreeIndex"); + public static final ConfigOption<Boolean> BTREE_INDEX_BLOOM_FILTER_ENABLED = + ConfigOptions.key("btree-index.bloom-filter.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether to enable the Bloom filter for BTree index point lookups."); + public static final ConfigOption<MemorySize> BTREE_INDEX_CACHE_SIZE = ConfigOptions.key("btree-index.cache-size") .memoryType() diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java index b0c272321c..d7d6017b5c 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java @@ -322,7 +322,7 @@ public class BTreeIndexReader implements Closeable { } public Optional<GlobalIndexResult> visitEqual(Object literal) { - return createResult(() -> rangeQuery(literal, literal, true, true)); + return createResult(() -> pointQuery(literal)); } public Optional<GlobalIndexResult> visitGreaterThan(Object literal) { @@ -339,7 +339,7 @@ public class BTreeIndexReader implements Closeable { if (literal == null) { continue; } - result.or(rangeQuery(literal, literal, true, true)); + result.or(pointQuery(literal)); } return result; }); @@ -399,6 +399,17 @@ public class BTreeIndexReader implements Closeable { return rangeQuery(minKey, maxKey, true, true); } + private RoaringNavigableMap64 pointQuery(Object key) throws IOException { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + byte[] rowIds = reader.lookup(keySerializer.serialize(key)); + if (rowIds != null) { + for (long rowId : deserializeRowIds(MemorySlice.wrap(rowIds))) { + result.add(rowId); + } + } + return result; + } + private TopNGlobalIndexResult topN( int limit, SortValue.SortDirection direction, SortValue.NullOrdering nullOrdering) throws IOException { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java index ccef1ad471..e1916d11a4 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java @@ -30,6 +30,7 @@ import org.apache.paimon.memory.MemorySliceOutput; import org.apache.paimon.sst.BlockHandle; import org.apache.paimon.sst.BloomFilterHandle; import org.apache.paimon.sst.SstFileWriter; +import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.LazyField; import org.apache.paimon.utils.RoaringNavigableMap64; @@ -90,12 +91,21 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { int blockSize, BlockCompressionFactory compressionFactory) throws IOException { + this(indexFileWriter, keySerializer, blockSize, null, compressionFactory); + } + + public BTreeIndexWriter( + GlobalIndexFileWriter indexFileWriter, + KeySerializer keySerializer, + int blockSize, + @Nullable BloomFilter.Builder bloomFilterBuilder, + BlockCompressionFactory compressionFactory) + throws IOException { this.fileName = indexFileWriter.newFileName(BTreeGlobalIndexerFactory.IDENTIFIER); this.out = indexFileWriter.newOutputStream(this.fileName); this.keySerializer = keySerializer; this.comparator = keySerializer.createComparator(); - // todo: we may enable bf to accelerate equal and in predicate in the future - this.writer = new SstFileWriter(out, blockSize, null, compressionFactory); + this.writer = new SstFileWriter(out, blockSize, bloomFilterBuilder, compressionFactory); } @Override @@ -150,8 +160,7 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { // write null bitmap BlockHandle nullBitmapHandle = writeNullBitmap(); - // write bloom filter (currently is always null, but we could add it for equal - // and in condition.) + // write bloom filter BloomFilterHandle bloomFilterHandle = writer.writeBloomFilter(); // write index block diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeBloomFilterTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeBloomFilterTest.java new file mode 100644 index 0000000000..7b1b8b8f0b --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeBloomFilterTest.java @@ -0,0 +1,239 @@ +/* + * 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.globalindex.btree; + +import org.apache.paimon.fs.ByteArraySeekableStream; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; +import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.memory.MemorySlice; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.sst.BloomFilterHandle; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.IntType; +import org.apache.paimon.utils.BloomFilter; +import org.apache.paimon.utils.MurmurHashUtils; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the optional Bloom filter in BTree index files. */ +class BTreeBloomFilterTest { + + private static final int ENTRY_COUNT = 1_000; + + @Test + void testBloomFormatCompatibleWithPython() { + assertThat(MurmurHashUtils.hashBytes("a".getBytes())).isEqualTo(1485273170); + assertThat(MurmurHashUtils.hashBytes("hello".getBytes())).isEqualTo(-1008564952); + assertThat(MurmurHashUtils.hashBytes("world".getBytes())).isEqualTo(-623458850); + + BloomFilter.Builder builder = BloomFilter.fixedBuilder(3, 0.05); + builder.addHash(MurmurHashUtils.hashBytes("a".getBytes())); + builder.addHash(MurmurHashUtils.hashBytes("hello".getBytes())); + builder.addHash(MurmurHashUtils.hashBytes("world".getBytes())); + + assertThat(builder.build().getMemorySegment().getArray()) + .isEqualTo(new byte[] {(byte) 0xea, 0x22, (byte) 0x99}); + } + + @Test + void testBloomFilterIsNotWrittenByDefault() throws Exception { + Fixture fixture = writeIndex(new Options()); + + assertThat(fixture.footer.getBloomFilterHandle()).isNull(); + } + + @Test + void testBloomFilterAvoidsDataReadForMissingLookups() throws Exception { + Options options = new Options(); + options.set(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED, true); + Fixture fixture = writeIndex(options); + BloomFilterHandle bloomHandle = fixture.footer.getBloomFilterHandle(); + + assertThat(bloomHandle).isNotNull(); + assertThat(bloomHandle.expectedEntries()).isEqualTo(ENTRY_COUNT); + + int missingKey = findRejectedMissingKey(fixture.bytes, bloomHandle); + assertMissingLookupReadsOnlyBloom( + fixture, bloomHandle, reader -> reader.visitEqual(missingKey)); + assertMissingLookupReadsOnlyBloom( + fixture, + bloomHandle, + reader -> reader.visitIn(Collections.singletonList(missingKey))); + } + + private void assertMissingLookupReadsOnlyBloom( + Fixture fixture, BloomFilterHandle bloomHandle, Lookup lookup) throws Exception { + CountingInput input = new CountingInput(fixture.bytes); + try (CacheManager cacheManager = new CacheManager(MemorySize.ofMebiBytes(1), 0.5); + BTreeIndexReader reader = + new BTreeIndexReader( + KeySerializer.create(new IntType()), + ignored -> input, + fixture.meta, + cacheManager)) { + input.resetCount(); + + assertThat(lookup.apply(reader).get().results()).isEmpty(); + assertThat(input.bytesRead).isEqualTo(bloomHandle.size()); + } + } + + private Fixture writeIndex(Options options) throws IOException { + ByteArrayGlobalIndexFileWriter fileWriter = new ByteArrayGlobalIndexFileWriter(); + BTreeGlobalIndexer indexer = + new BTreeGlobalIndexer(new DataField(0, "k", new IntType()), options); + GlobalIndexSingleColumnWriter writer = indexer.createWriter(fileWriter); + for (int i = 0; i < ENTRY_COUNT; i++) { + writer.write(i * 2, i); + } + List<ResultEntry> entries = writer.finish(); + assertThat(entries).hasSize(1); + + byte[] bytes = fileWriter.bytes(); + ResultEntry entry = entries.get(0); + BTreeFileFooter footer = + BTreeFileFooter.readFooter( + MemorySlice.wrap(bytes) + .slice( + bytes.length - BTreeFileFooter.ENCODED_LENGTH, + BTreeFileFooter.ENCODED_LENGTH) + .toInput()); + GlobalIndexIOMeta meta = + new GlobalIndexIOMeta( + new Path(entry.fileName()), bytes.length, entry.rowCount(), entry.meta()); + return new Fixture(bytes, footer, meta); + } + + private int findRejectedMissingKey(byte[] bytes, BloomFilterHandle handle) { + BloomFilter filter = new BloomFilter(handle.expectedEntries(), handle.size()); + filter.setMemorySegment(MemorySegment.wrap(bytes), (int) handle.offset()); + KeySerializer serializer = KeySerializer.create(new IntType()); + for (int candidate = 1; candidate < ENTRY_COUNT * 2; candidate += 2) { + if (!filter.testHash(MurmurHashUtils.hashBytes(serializer.serialize(candidate)))) { + return candidate; + } + } + throw new AssertionError("Unable to find a Bloom-filter-negative missing key."); + } + + private static class Fixture { + private final byte[] bytes; + private final BTreeFileFooter footer; + private final GlobalIndexIOMeta meta; + + private Fixture(byte[] bytes, BTreeFileFooter footer, GlobalIndexIOMeta meta) { + this.bytes = bytes; + this.footer = footer; + this.meta = meta; + } + } + + @FunctionalInterface + private interface Lookup { + Optional<GlobalIndexResult> apply(BTreeIndexReader reader); + } + + private static class ByteArrayGlobalIndexFileWriter implements GlobalIndexFileWriter { + private final ByteArrayPositionOutputStream output = new ByteArrayPositionOutputStream(); + + @Override + public String newFileName(String prefix) { + return prefix + "-test"; + } + + @Override + public PositionOutputStream newOutputStream(String fileName) { + return output; + } + + private byte[] bytes() { + return output.bytes(); + } + } + + private static class ByteArrayPositionOutputStream extends PositionOutputStream { + private final ByteArrayOutputStream output = new ByteArrayOutputStream(); + + @Override + public long getPos() { + return output.size(); + } + + @Override + public void write(int value) { + output.write(value); + } + + @Override + public void write(byte[] bytes) { + output.write(bytes, 0, bytes.length); + } + + @Override + public void write(byte[] bytes, int offset, int length) { + output.write(bytes, offset, length); + } + + @Override + public void flush() {} + + @Override + public void close() {} + + private byte[] bytes() { + return output.toByteArray(); + } + } + + private static class CountingInput extends ByteArraySeekableStream { + private int bytesRead; + + private CountingInput(byte[] bytes) { + super(bytes); + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + int read = super.read(bytes, offset, length); + bytesRead += Math.max(read, 0); + return read; + } + + private void resetCount() { + bytesRead = 0; + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java index 27ae90f98e..72ceec750b 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java @@ -30,4 +30,9 @@ class BTreeIndexOptionsTest { assertThat(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE.defaultValue()) .isEqualTo(10_000_000L); } + + @Test + void testBloomFilterDisabledByDefault() { + assertThat(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED.defaultValue()).isFalse(); + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java index 2f22974c43..04421e707a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java @@ -106,6 +106,7 @@ public class BTreeThreadSafetyTest { Options options = new Options(); options.set(BTreeIndexOptions.BTREE_INDEX_CACHE_SIZE, MemorySize.ofMebiBytes(8)); + options.set(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED, true); DataField dataField = new DataField(1, "id", new IntType()); globalIndexer = new BTreeGlobalIndexer(dataField, options); keySerializer = KeySerializer.create(new IntType()); diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 0de00abdc1..7df8ee0ef2 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1018,6 +1018,15 @@ class CoreOptions: .with_description("The block size to use for BTree global indexes.") ) + BTREE_INDEX_BLOOM_FILTER_ENABLED: ConfigOption[bool] = ( + ConfigOptions.key("btree-index.bloom-filter.enabled") + .boolean_type() + .default_value(False) + .with_description( + "Whether to enable the Bloom filter for BTree index point lookups." + ) + ) + SORTED_INDEX_RECORDS_PER_RANGE: ConfigOption[int] = ( ConfigOptions.key("sorted-index.records-per-range") .long_type() @@ -1827,6 +1836,9 @@ class CoreOptions: def btree_index_block_size(self) -> int: return self.options.get(CoreOptions.BTREE_INDEX_BLOCK_SIZE).get_bytes() + def btree_index_bloom_filter_enabled(self) -> bool: + return self.options.get(CoreOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED) + def sorted_index_records_per_range(self) -> int: if self.options.contains(CoreOptions.SORTED_INDEX_RECORDS_PER_RANGE): return self.options.get(CoreOptions.SORTED_INDEX_RECORDS_PER_RANGE) diff --git a/paimon-python/pypaimon/globalindex/btree/bloom_filter.py b/paimon-python/pypaimon/globalindex/btree/bloom_filter.py new file mode 100644 index 0000000000..b6e1dd030a --- /dev/null +++ b/paimon-python/pypaimon/globalindex/btree/bloom_filter.py @@ -0,0 +1,151 @@ +# 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. + +"""Bloom filter compatible with Paimon's Java SST implementation.""" + +import math +from typing import Collection, Optional + + +_INT_MASK = 0xFFFFFFFF +_INT_SIGN_BIT = 0x80000000 +_MURMUR_C1 = 0xCC9E2D51 +_MURMUR_C2 = 0x1B873593 +_MURMUR_DEFAULT_SEED = 42 + + +def _int32(value: int) -> int: + value &= _INT_MASK + return value if value < _INT_SIGN_BIT else value - (_INT_MASK + 1) + + +def _rotate_left(value: int, distance: int) -> int: + unsigned = value & _INT_MASK + return _int32( + (unsigned << distance) | (unsigned >> (32 - distance))) + + +def _mix_k1(value: int) -> int: + value = _int32(value * _MURMUR_C1) + value = _rotate_left(value, 15) + return _int32(value * _MURMUR_C2) + + +def _mix_h1(hash_value: int, mixed_value: int) -> int: + hash_value = _int32( + (hash_value & _INT_MASK) ^ (mixed_value & _INT_MASK)) + hash_value = _rotate_left(hash_value, 13) + return _int32(hash_value * 5 + 0xE6546B64) + + +def _fmix(hash_value: int, length: int) -> int: + hash_value = _int32((hash_value & _INT_MASK) ^ length) + hash_value = _int32( + (hash_value & _INT_MASK) ^ ((hash_value & _INT_MASK) >> 16)) + hash_value = _int32(hash_value * 0x85EBCA6B) + hash_value = _int32( + (hash_value & _INT_MASK) ^ ((hash_value & _INT_MASK) >> 13)) + hash_value = _int32(hash_value * 0xC2B2AE35) + return _int32( + (hash_value & _INT_MASK) ^ ((hash_value & _INT_MASK) >> 16)) + + +def murmur_hash_bytes(data: bytes) -> int: + """Return the signed hash produced by Java ``MurmurHashUtils.hashBytes``.""" + aligned_length = len(data) - len(data) % 4 + hash_value = _MURMUR_DEFAULT_SEED + + for offset in range(0, aligned_length, 4): + word = int.from_bytes(data[offset:offset + 4], "little", signed=True) + hash_value = _mix_h1(hash_value, _mix_k1(word)) + + # Paimon's Java implementation mixes every remaining signed byte as a + # separate word instead of combining the Murmur3 tail bytes. + for byte in data[aligned_length:]: + signed_byte = byte if byte < 128 else byte - 256 + hash_value = _mix_h1(hash_value, _mix_k1(signed_byte)) + + return _fmix(hash_value, len(data)) + + +class BloomFilter: + """A compact Bloom filter using the Java Paimon bit layout.""" + + def __init__( + self, + expected_entries: int, + byte_size: int, + data: Optional[bytes] = None, + ): + if expected_entries <= 0: + raise ValueError("expected_entries must be positive") + if byte_size <= 0: + raise ValueError("byte_size must be positive") + if data is not None and len(data) != byte_size: + raise ValueError( + "Bloom filter size mismatch: expected %s bytes, got %s" + % (byte_size, len(data))) + + self.expected_entries = expected_entries + self._bits = bytearray(byte_size if data is None else data) + hashes = byte_size * 8 / expected_entries * math.log(2) + # Java Math.round rounds positive halfway values toward positive infinity. + self._num_hash_functions = max(1, int(math.floor(hashes + 0.5))) + + @classmethod + def from_hashes( + cls, + hashes: Collection[int], + false_positive_probability: float, + ) -> Optional['BloomFilter']: + expected_entries = len(hashes) + if expected_entries == 0: + return None + if not 0 < false_positive_probability < 1: + raise ValueError( + "false_positive_probability must be between 0 and 1") + + optimal_bits = int( + -expected_entries * math.log(false_positive_probability) + / (math.log(2) * math.log(2))) + byte_size = int(math.ceil(optimal_bits / 8.0)) + bloom_filter = cls(expected_entries, byte_size) + for hash_value in hashes: + bloom_filter.add_hash(hash_value) + return bloom_filter + + def add_hash(self, hash_value: int) -> None: + for position in self._positions(hash_value): + self._bits[position >> 3] |= 1 << (position & 7) + + def test_hash(self, hash_value: int) -> bool: + return all( + self._bits[position >> 3] & (1 << (position & 7)) + for position in self._positions(hash_value)) + + def to_bytes(self) -> bytes: + return bytes(self._bits) + + def _positions(self, hash_value: int): + hash_value = _int32(hash_value) + hash2 = (hash_value & _INT_MASK) >> 16 + bit_size = len(self._bits) * 8 + for index in range(1, self._num_hash_functions + 1): + combined_hash = _int32(hash_value + index * hash2) + if combined_hash < 0: + combined_hash = _int32(~combined_hash) + yield combined_hash % bit_size diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py index 61decbefef..c1d0b6007e 100644 --- a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py +++ b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py @@ -120,6 +120,7 @@ class BTreeIndexReader: return SstFileReader( self.input_stream, comparator, self.footer.index_block_handle, + bloom_filter_handle=self.footer.bloom_filter_handle, use_pread=self._supports_pread, io_lock=self._io_lock) def _read_null_bitmap(self) -> RoaringBitmap64: @@ -217,8 +218,7 @@ class BTreeIndexReader: self._range_query(self.min_key, literal, True, True)) def visit_equal(self, literal: object) -> Optional[GlobalIndexResult]: - return GlobalIndexResult.create( - self._range_query(literal, literal, True, True)) + return GlobalIndexResult.create(self._point_query(literal)) def visit_greater_than(self, literal: object) -> Optional[GlobalIndexResult]: return GlobalIndexResult.create(self.greater_than(literal)) @@ -226,8 +226,9 @@ class BTreeIndexReader: def visit_in(self, literals: List[object]) -> Optional[GlobalIndexResult]: result = RoaringBitmap64() for literal in literals: - range_result = self._range_query(literal, literal, True, True) - result = RoaringBitmap64.or_(result, range_result) + if literal is None: + continue + result = RoaringBitmap64.or_(result, self._point_query(literal)) return GlobalIndexResult.create(result) def visit_not_in(self, literals: List[object]) -> Optional[GlobalIndexResult]: @@ -259,6 +260,14 @@ class BTreeIndexReader: def greater_than(self, literal: object) -> RoaringBitmap64: return self._range_query(literal, self.max_key, False, True) + def _point_query(self, key: object) -> RoaringBitmap64: + result = RoaringBitmap64() + row_ids = self.reader.lookup(self.key_serializer.serialize(key)) + if row_ids is not None: + for row_id in _deserialize_row_ids(row_ids): + result.add(row_id) + return result + def close(self) -> None: if self.input_stream is not None: self.input_stream.close() diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py index be3dd7014a..ad244d0ce5 100644 --- a/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py +++ b/paimon-python/pypaimon/globalindex/btree/btree_index_writer.py @@ -17,6 +17,7 @@ """BTree global index writer compatible with Java's SST-backed format.""" +from array import array import struct import zlib from typing import List, Optional @@ -28,6 +29,10 @@ from pypaimon.globalindex.index_file_utils import ( write_var_len_int, write_var_len_long, ) +from pypaimon.globalindex.btree.bloom_filter import ( + BloomFilter, + murmur_hash_bytes, +) from pypaimon.globalindex.result_entry import ResultEntry from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta from pypaimon.utils.roaring_bitmap import RoaringBitmap64 @@ -39,6 +44,7 @@ _BTREE_MAGIC_NUMBER = 0x50425449 _BTREE_CURRENT_VERSION = 1 _BLOCK_ALIGNED = 0 _BLOCK_UNALIGNED = 1 +_BLOOM_FILTER_FPP = 0.05 def _write_block_handle(offset: int, size: int) -> bytes: @@ -100,15 +106,23 @@ class _BlockWriter: class _SstFileWriter: - def __init__(self, out: PositionOutput, block_size: int): + def __init__( + self, + out: PositionOutput, + block_size: int, + bloom_filter_enabled: bool, + ): self._out = out self._block_size = block_size self._data_block_writer = _BlockWriter() self._index_block_writer = _BlockWriter() self._last_key: Optional[bytes] = None + self._bloom_hashes = array('i') if bloom_filter_enabled else None def put(self, key: bytes, value: bytes) -> None: self._data_block_writer.add(key, value) + if self._bloom_hashes is not None: + self._bloom_hashes.append(murmur_hash_bytes(key)) self._last_key = key if self._data_block_writer.memory() > self._block_size: @@ -122,7 +136,19 @@ class _SstFileWriter: self._last_key, _write_block_handle(block_offset, block_size)) def write_bloom_filter(self): - return None + if self._bloom_hashes is None: + return None + + bloom_filter = BloomFilter.from_hashes( + self._bloom_hashes, _BLOOM_FILTER_FPP) + self._bloom_hashes = None + if bloom_filter is None: + return None + + data = bloom_filter.to_bytes() + offset = self._out.pos + self._out.write(data) + return offset, len(data), bloom_filter.expected_entries def write_index_block(self): return self._write_block(self._index_block_writer) @@ -150,6 +176,7 @@ class BTreeIndexWriter: index_path: str, key_serializer, block_size: int = 64 * 1024, + bloom_filter_enabled: bool = False, ): self.file_name = new_global_index_file_name(BTREE_IDENTIFIER) self._file_io = file_io @@ -166,7 +193,8 @@ class BTreeIndexWriter: self._file_io.check_or_mkdirs(self._index_path) self._output = PositionOutput( self._file_io.new_output_stream(self._file_path())) - self._sst = _SstFileWriter(self._output, block_size) + self._sst = _SstFileWriter( + self._output, block_size, bloom_filter_enabled) def write(self, key, row_id: int) -> None: self._row_count += 1 diff --git a/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py b/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py index e766946918..44a27dbaf9 100644 --- a/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py +++ b/paimon-python/pypaimon/globalindex/btree/sst_file_reader.py @@ -39,6 +39,11 @@ from pypaimon.globalindex.block_compression import ( from pypaimon.globalindex.btree.block_handle import BlockHandle from pypaimon.globalindex.btree.block_entry import BlockEntry from pypaimon.globalindex.btree.block_reader import BlockReader, BlockIterator +from pypaimon.globalindex.btree.bloom_filter import ( + BloomFilter, + murmur_hash_bytes, +) +from pypaimon.globalindex.btree.btree_file_footer import BloomFilterHandle from pypaimon.globalindex.memory_slice_input import MemorySliceInput @@ -123,6 +128,7 @@ class SstFileReader: input_stream: BinaryIO, comparator: Callable[[bytes, bytes], int], index_block_handle: BlockHandle, + bloom_filter_handle: Optional[BloomFilterHandle] = None, use_pread: bool = False, io_lock: Optional[threading.Lock] = None, ): @@ -130,6 +136,9 @@ class SstFileReader: self.input_stream = input_stream self._supports_pread = use_pread self._lock = io_lock or threading.Lock() + self._bloom_filter_handle = bloom_filter_handle + self._bloom_filter: Optional[BloomFilter] = None + self._bloom_filter_lock = threading.Lock() self.index_block = self._read_block(index_block_handle) def _read_from(self, offset: int, length: int) -> bytes: @@ -173,6 +182,39 @@ class SstFileReader: read_block, self.index_block.iterator()) + def lookup(self, key: bytes) -> Optional[bytes]: + """Return the value for an exact key, or ``None`` when absent.""" + if ( + self._bloom_filter_handle is not None + and not self._test_bloom_filter(murmur_hash_bytes(key)) + ): + return None + + index_iterator = self.index_block.iterator() + index_iterator.seek_to(key) + if not index_iterator.has_next(): + return None + + index_entry = next(index_iterator) + handle = SstFileIterator._parse_block_handle(index_entry.value) + data_iterator = self._read_block(handle).iterator() + if not data_iterator.seek_to(key): + return None + return next(data_iterator).value + + def _test_bloom_filter(self, hash_value: int) -> bool: + bloom_filter = self._bloom_filter + if bloom_filter is None: + with self._bloom_filter_lock: + bloom_filter = self._bloom_filter + if bloom_filter is None: + handle = self._bloom_filter_handle + data = self._read_from(handle.offset, handle.size) + bloom_filter = BloomFilter( + handle.expected_entries, handle.size, data) + self._bloom_filter = bloom_filter + return bloom_filter.test_hash(hash_value) + def close(self) -> None: """Close the reader and release resources.""" # No resources to release in this implementation diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py b/paimon-python/pypaimon/globalindex/create_global_index.py index 93473bb7b3..60caf4e8bd 100644 --- a/paimon-python/pypaimon/globalindex/create_global_index.py +++ b/paimon-python/pypaimon/globalindex/create_global_index.py @@ -275,6 +275,8 @@ class GlobalIndexBuilder: index_path, key_serializer, block_size=self._core_options.btree_index_block_size(), + bloom_filter_enabled=( + self._core_options.btree_index_bloom_filter_enabled()), ) if self._index_type == BITMAP_IDENTIFIER: return BitmapIndexWriter( diff --git a/paimon-python/pypaimon/tests/btree_bloom_filter_test.py b/paimon-python/pypaimon/tests/btree_bloom_filter_test.py new file mode 100644 index 0000000000..8fc09c58e5 --- /dev/null +++ b/paimon-python/pypaimon/tests/btree_bloom_filter_test.py @@ -0,0 +1,176 @@ +# 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. + +"""Tests for the optional Bloom filter in Python BTree indexes.""" + +import io +import os +import tempfile +import unittest + +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.common.options.options import Options +from pypaimon.filesystem.local_file_io import LocalFileIO +from pypaimon.globalindex.btree.bloom_filter import ( + BloomFilter, + murmur_hash_bytes, +) +from pypaimon.globalindex.btree.btree_file_footer import BTreeFileFooter +from pypaimon.globalindex.btree.btree_index_reader import BTreeIndexReader +from pypaimon.globalindex.btree.btree_index_writer import BTreeIndexWriter +from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta +from pypaimon.globalindex.key_serializer import create_serializer +from pypaimon.schema.data_types import AtomicType + + +class _CountingInput(io.BytesIO): + + def __init__(self, data): + super().__init__(data) + self.bytes_read = 0 + + def read(self, size=-1): + data = super().read(size) + self.bytes_read += len(data) + return data + + +class _MemoryFileIO: + + def __init__(self, data): + self.data = data + self.last_input = None + + def new_input_stream(self, path): + self.last_input = _CountingInput(self.data) + return self.last_input + + +class BTreeBloomFilterTest(unittest.TestCase): + + def setUp(self): + self.serializer = create_serializer(AtomicType("INT")) + + def test_java_compatible_hash_and_bit_layout(self): + values = [b"a", b"hello", b"world"] + + self.assertEqual(1485273170, murmur_hash_bytes(values[0])) + self.assertEqual(-1008564952, murmur_hash_bytes(values[1])) + self.assertEqual(-623458850, murmur_hash_bytes(values[2])) + + bloom_filter = BloomFilter.from_hashes( + [murmur_hash_bytes(value) for value in values], 0.05) + self.assertEqual("ea2299", bloom_filter.to_bytes().hex()) + + def test_bloom_filter_option_defaults_to_disabled(self): + self.assertFalse( + CoreOptions(Options({})).btree_index_bloom_filter_enabled()) + self.assertTrue( + CoreOptions(Options({ + "btree-index.bloom-filter.enabled": "true", + })).btree_index_bloom_filter_enabled()) + + data, _ = self._write_index(False, 10) + footer = BTreeFileFooter.read_footer( + data[-BTreeFileFooter.ENCODED_LENGTH:]) + self.assertIsNone(footer.bloom_filter_handle) + + def test_missing_point_lookups_read_only_bloom_filter(self): + entry_count = 1000 + data, entry = self._write_index(True, entry_count) + footer = BTreeFileFooter.read_footer( + data[-BTreeFileFooter.ENCODED_LENGTH:]) + handle = footer.bloom_filter_handle + + self.assertIsNotNone(handle) + self.assertEqual(entry_count, handle.expected_entries) + + bloom_filter = BloomFilter( + handle.expected_entries, + handle.size, + data[handle.offset:handle.offset + handle.size], + ) + missing_key = next( + candidate + for candidate in range(1, entry_count * 2, 2) + if not bloom_filter.test_hash( + murmur_hash_bytes(self.serializer.serialize(candidate))) + ) + + self._assert_missing_lookup_reads_only_bloom( + data, entry, handle.size, + lambda reader: reader.visit_equal(missing_key)) + self._assert_missing_lookup_reads_only_bloom( + data, entry, handle.size, + lambda reader: reader.visit_in([None, missing_key])) + + def test_present_point_lookup_returns_all_row_ids(self): + data, entry = self._write_index(True, 10, duplicate_key=8) + file_io = _MemoryFileIO(data) + reader = self._reader(file_io, data, entry) + try: + result = reader.visit_equal(8) + self.assertEqual([4, 10], result.results().to_list()) + finally: + reader.close() + + def _write_index(self, enabled, entry_count, duplicate_key=None): + with tempfile.TemporaryDirectory() as directory: + file_io = LocalFileIO() + writer = BTreeIndexWriter( + file_io, + directory, + self.serializer, + block_size=128, + bloom_filter_enabled=enabled, + ) + for index in range(entry_count): + writer.write(index * 2, index) + if index * 2 == duplicate_key: + writer.write(index * 2, entry_count) + entry = writer.finish()[0] + path = os.path.join(directory, entry.file_name) + with open(path, "rb") as stream: + return stream.read(), entry + + def _reader(self, file_io, data, entry): + return BTreeIndexReader( + self.serializer, + file_io, + "/unused", + GlobalIndexIOMeta( + file_name=entry.file_name, + file_size=len(data), + metadata=entry.meta, + ), + ) + + def _assert_missing_lookup_reads_only_bloom( + self, data, entry, bloom_size, lookup, + ): + file_io = _MemoryFileIO(data) + reader = self._reader(file_io, data, entry) + try: + file_io.last_input.bytes_read = 0 + self.assertTrue(lookup(reader).results().is_empty()) + self.assertEqual(bloom_size, file_io.last_input.bytes_read) + finally: + reader.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index 70985b95fa..9735d5b1c0 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -35,6 +35,7 @@ from pypaimon.globalindex.build_plan import ( split_one_by_contiguous_row_range as _split_one_by_contiguous_row_range, ) from pypaimon.globalindex.create_global_index import GlobalIndexBuilder +from pypaimon.globalindex.btree.btree_file_footer import BTreeFileFooter from pypaimon.globalindex.key_serializer import create_serializer from pypaimon.globalindex.full_text.native_full_text_global_index_reader import ( FULL_TEXT_IDENTIFIER, @@ -235,7 +236,10 @@ class GlobalIndexBuildTest( added = table.create_global_index( 'id', - options={'sorted-index.records-per-range': '2'}, + options={ + 'sorted-index.records-per-range': '2', + 'btree-index.bloom-filter.enabled': 'true', + }, ) self.assertEqual(2, added) @@ -248,6 +252,19 @@ class GlobalIndexBuildTest( self.assertEqual({0}, {e.index_file.global_index_meta.row_range_start for e in entries}) self.assertEqual({3}, {e.index_file.global_index_meta.row_range_end for e in entries}) + index_path_factory = table.path_factory().global_index_path_factory() + for entry in entries: + index_file = entry.index_file + index_path = index_file.external_path or index_path_factory.to_path( + index_file.file_name) + footer_bytes = table.file_io.read_file_range( + index_path, + index_file.file_size - BTreeFileFooter.ENCODED_LENGTH, + BTreeFileFooter.ENCODED_LENGTH, + ) + footer = BTreeFileFooter.read_footer(footer_bytes) + self.assertIsNotNone(footer.bloom_filter_handle) + read_builder = table.new_read_builder() predicate = read_builder.new_predicate_builder().equal('id', 2) with DataEvolutionGlobalIndexScanner.create(
