This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 7505c97585 [core][python] Add optional Bloom filters to BTree indexes
(#9946)
7505c97585 is described below
commit 7505c97585126ec1552ec435607772a60a1d2c5d
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(
