This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d09cb1ac feat(btree): add optional bloom filters (#869)
d09cb1ac is described below
commit d09cb1ac3411d53b7259ef333cfc7f363afb3c39
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 18 11:41:36 2026 +0800
feat(btree): add optional bloom filters (#869)
---
crates/paimon/src/btree/bloom_filter.rs | 186 +++++++++++++++++++++
crates/paimon/src/btree/footer.rs | 2 +
crates/paimon/src/btree/mod.rs | 1 +
crates/paimon/src/btree/reader.rs | 102 ++++++++++-
crates/paimon/src/btree/sst_file.rs | 41 +++++
crates/paimon/src/btree/tests.rs | 117 +++++++++++++
crates/paimon/src/btree/writer.rs | 28 +++-
crates/paimon/src/spec/murmur_hash.rs | 26 +++
.../sorted_global_index_build_builder/tests.rs | 26 ++-
.../sorted_global_index_build_builder/writer.rs | 3 +-
.../src/table/sorted_global_index_options.rs | 39 +++++
docs/src/sql.md | 8 +-
12 files changed, 567 insertions(+), 12 deletions(-)
diff --git a/crates/paimon/src/btree/bloom_filter.rs
b/crates/paimon/src/btree/bloom_filter.rs
new file mode 100644
index 00000000..3652d269
--- /dev/null
+++ b/crates/paimon/src/btree/bloom_filter.rs
@@ -0,0 +1,186 @@
+// 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 used by Java-compatible BTree SST files.
+
+use bytes::Bytes;
+use std::io;
+
+const BITS_PER_BYTE: usize = u8::BITS as usize;
+
+/// Bloom filter with its hash count derived from the expected entry count and
serialized size.
+pub(super) struct BloomFilter {
+ expected_entries: u64,
+ bits: BitSet,
+ num_hash_functions: i32,
+}
+
+enum BitSet {
+ Mutable(Vec<u8>),
+ Shared(Bytes),
+}
+
+impl BloomFilter {
+ pub(super) fn from_hashes(hashes: &[i32], fpp: f64) ->
io::Result<Option<Self>> {
+ if hashes.is_empty() {
+ return Ok(None);
+ }
+ if !fpp.is_finite() || fpp <= 0.0 || fpp >= 1.0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!("Bloom filter fpp must be finite and in (0, 1), but
was {fpp}"),
+ ));
+ }
+
+ let expected_entries = hashes.len() as u64;
+ let log_two = 2.0_f64.ln();
+ // Java first narrows the optimal bit count to int, then rounds the
byte count up.
+ let optimal_bits = (-(expected_entries as f64) * fpp.ln() / (log_two *
log_two)) as i32;
+ let byte_size = (f64::from(optimal_bits) / BITS_PER_BYTE as
f64).ceil() as usize;
+ if byte_size == 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!(
+ "Bloom filter size must be positive for {expected_entries}
entries and fpp {fpp}"
+ ),
+ ));
+ }
+
+ let mut bits = Vec::new();
+ bits.try_reserve_exact(byte_size).map_err(|error| {
+ io::Error::other(format!("Failed to allocate Bloom filter:
{error}"))
+ })?;
+ bits.resize(byte_size, 0);
+ let num_hash_functions = num_hash_functions(expected_entries,
bits.len())?;
+ let mut filter = Self {
+ expected_entries,
+ bits: BitSet::Mutable(bits),
+ num_hash_functions,
+ };
+ for &hash in hashes {
+ filter.add_hash(hash);
+ }
+ Ok(Some(filter))
+ }
+
+ pub(super) fn from_bytes(expected_entries: u64, bits: Bytes) ->
io::Result<Self> {
+ if expected_entries == 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Bloom filter expected entry count must be positive",
+ ));
+ }
+ if bits.is_empty() {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Bloom filter bitset must not be empty",
+ ));
+ }
+
+ let num_hash_functions = num_hash_functions(expected_entries,
bits.len())?;
+ Ok(Self {
+ expected_entries,
+ bits: BitSet::Shared(bits),
+ num_hash_functions,
+ })
+ }
+
+ pub(super) fn expected_entries(&self) -> u64 {
+ self.expected_entries
+ }
+
+ pub(super) fn bytes(&self) -> &[u8] {
+ self.bits.bytes()
+ }
+
+ pub(super) fn test_hash(&self, hash: i32) -> bool {
+ (1..=self.num_hash_functions).all(|iteration|
self.is_set(self.position(hash, iteration)))
+ }
+
+ fn add_hash(&mut self, hash: i32) {
+ for iteration in 1..=self.num_hash_functions {
+ let position = self.position(hash, iteration);
+ match &mut self.bits {
+ BitSet::Mutable(bits) => bits[position >> 3] |= 1 << (position
& 7),
+ BitSet::Shared(_) => unreachable!("serialized Bloom filter is
immutable"),
+ }
+ }
+ }
+
+ fn position(&self, hash: i32, iteration: i32) -> usize {
+ let hash2 = ((hash as u32) >> 16) as i32;
+ let mut combined = hash.wrapping_add(iteration.wrapping_mul(hash2));
+ if combined < 0 {
+ combined = !combined;
+ }
+ combined as usize % (self.bits.bytes().len() * BITS_PER_BYTE)
+ }
+
+ fn is_set(&self, position: usize) -> bool {
+ self.bits.bytes()[position >> 3] & (1 << (position & 7)) != 0
+ }
+}
+
+impl BitSet {
+ fn bytes(&self) -> &[u8] {
+ match self {
+ Self::Mutable(bits) => bits,
+ Self::Shared(bits) => bits,
+ }
+ }
+}
+
+fn num_hash_functions(expected_entries: u64, byte_size: usize) ->
io::Result<i32> {
+ let bit_size = byte_size.checked_mul(BITS_PER_BYTE).ok_or_else(|| {
+ io::Error::new(
+ io::ErrorKind::InvalidData,
+ "Bloom filter bit size overflows",
+ )
+ })?;
+ Ok(
+ (((bit_size as f64 / expected_entries as f64) * 2.0_f64.ln())
+ .round()
+ .max(1.0)) as i32,
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::murmur_hash::hash_bytes;
+
+ #[test]
+ fn test_java_compatible_bit_layout() {
+ let hashes = [hash_bytes(b"a"), hash_bytes(b"hello"),
hash_bytes(b"world")];
+ let filter = BloomFilter::from_hashes(&hashes, 0.05).unwrap().unwrap();
+
+ assert_eq!(filter.expected_entries(), 3);
+ assert_eq!(filter.bytes(), &[0xea, 0x22, 0x99]);
+ assert!(hashes.into_iter().all(|hash| filter.test_hash(hash)));
+ }
+
+ #[test]
+ fn test_empty_hashes_do_not_create_filter() {
+ assert!(BloomFilter::from_hashes(&[], 0.05).unwrap().is_none());
+ }
+
+ #[test]
+ fn test_rejects_invalid_serialized_filter() {
+ assert!(BloomFilter::from_bytes(0, Bytes::from_static(&[1])).is_err());
+ assert!(BloomFilter::from_bytes(1, Bytes::new()).is_err());
+ }
+}
diff --git a/crates/paimon/src/btree/footer.rs
b/crates/paimon/src/btree/footer.rs
index 449e95d6..027802a7 100644
--- a/crates/paimon/src/btree/footer.rs
+++ b/crates/paimon/src/btree/footer.rs
@@ -50,6 +50,8 @@ pub struct BTreeFileFooter {
}
impl BTreeFileFooter {
+ pub const ENCODED_LENGTH: usize = BTREE_FOOTER_ENCODED_LENGTH;
+
pub fn new(
bloom_filter_handle: Option<BloomFilterHandle>,
index_block_handle: BlockHandle,
diff --git a/crates/paimon/src/btree/mod.rs b/crates/paimon/src/btree/mod.rs
index ad66ddfe..6cc4ea01 100644
--- a/crates/paimon/src/btree/mod.rs
+++ b/crates/paimon/src/btree/mod.rs
@@ -37,6 +37,7 @@
//! ```
mod block;
+mod bloom_filter;
mod footer;
pub(crate) mod key_serde;
mod meta;
diff --git a/crates/paimon/src/btree/reader.rs
b/crates/paimon/src/btree/reader.rs
index 11d48ebd..7f8f1ba1 100644
--- a/crates/paimon/src/btree/reader.rs
+++ b/crates/paimon/src/btree/reader.rs
@@ -24,14 +24,22 @@
//! - IN / NOT IN queries
use crate::btree::block::{BlockHandle, BlockReader};
-use crate::btree::footer::{BTreeFileFooter, BTREE_FOOTER_ENCODED_LENGTH};
+use crate::btree::bloom_filter::BloomFilter;
+use crate::btree::footer::{BTreeFileFooter, BloomFilterHandle,
BTREE_FOOTER_ENCODED_LENGTH};
use crate::btree::meta::BTreeIndexMeta;
use crate::btree::sst_file::{read_block_from_bytes, SstFileReader};
use crate::btree::var_len::{decode_var_int, decode_var_long};
use crate::io::FileRead;
+use crate::spec::murmur_hash::hash_bytes;
use roaring::RoaringTreemap;
use std::cmp::Ordering;
use std::io::{self, Cursor};
+use tokio::sync::OnceCell;
+
+struct LazyBloomFilter {
+ handle: BloomFilterHandle,
+ filter: OnceCell<BloomFilter>,
+}
/// BTree index reader with on-demand async data block loading.
pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
@@ -41,6 +49,7 @@ pub struct BTreeIndexReader<F: Fn(&[u8], &[u8]) -> Ordering> {
min_key: Option<Vec<u8>>,
max_key: Option<Vec<u8>>,
key_comparator: F,
+ bloom_filter: Option<LazyBloomFilter>,
}
impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
@@ -67,6 +76,10 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
.await
.map_err(|e| io::Error::other(e.to_string()))?;
let footer = BTreeFileFooter::read_footer(&footer_bytes)?;
+ let bloom_filter = footer.bloom_filter_handle.map(|handle|
LazyBloomFilter {
+ handle,
+ filter: OnceCell::new(),
+ });
// 2. Read index block
let idx = &footer.index_block_handle;
@@ -91,6 +104,7 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
min_key: meta.first_key.clone(),
max_key: meta.last_key.clone(),
key_comparator,
+ bloom_filter,
})
}
@@ -248,9 +262,81 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F> {
read_block_from_bytes(&bytes, handle.size)
}
+ async fn bloom_might_contain(&self, key: &[u8]) -> io::Result<bool> {
+ let Some(lazy) = &self.bloom_filter else {
+ return Ok(true);
+ };
+ let filter = lazy
+ .filter
+ .get_or_try_init(|| async {
+ if lazy.handle.size == 0 || lazy.handle.expected_entries == 0 {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "BTree Bloom filter handle must have positive size and
expected entries",
+ ));
+ }
+ let end = lazy
+ .handle
+ .offset
+ .checked_add(u64::from(lazy.handle.size))
+ .ok_or_else(|| {
+ io::Error::new(
+ io::ErrorKind::InvalidData,
+ "BTree Bloom filter range overflows",
+ )
+ })?;
+ let bits = self
+ .reader
+ .read(lazy.handle.offset..end)
+ .await
+ .map_err(|error| io::Error::other(error.to_string()))?;
+ if bits.len() != lazy.handle.size as usize {
+ return Err(io::Error::new(
+ io::ErrorKind::UnexpectedEof,
+ format!(
+ "BTree Bloom filter expected {} bytes, read {}",
+ lazy.handle.size,
+ bits.len()
+ ),
+ ));
+ }
+ BloomFilter::from_bytes(lazy.handle.expected_entries, bits)
+ })
+ .await?;
+ Ok(filter.test_hash(hash_bytes(key)))
+ }
+
/// Equal query: returns row ids for the given key.
pub async fn query_equal(&self, key: &[u8]) -> io::Result<RoaringTreemap> {
- self.range_query(key, key, true, true).await
+ let cmp = &self.key_comparator;
+ if self
+ .min_key
+ .as_deref()
+ .is_none_or(|min| cmp(key, min) == Ordering::Less)
+ || self
+ .max_key
+ .as_deref()
+ .is_none_or(|max| cmp(key, max) == Ordering::Greater)
+ {
+ return Ok(RoaringTreemap::new());
+ }
+ if !self.bloom_might_contain(key).await? {
+ return Ok(RoaringTreemap::new());
+ }
+
+ let index_block = self.sst_reader.index_block();
+ let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
+ let Some((_last_key, handle_bytes)) = index_iter.next() else {
+ return Ok(RoaringTreemap::new());
+ };
+ let handle = BlockHandle::decode(handle_bytes)?;
+ let block = self.read_data_block(&handle).await?;
+ let (found, mut entry_iter) = block.seek_and_iter(key, cmp);
+ let mut result = RoaringTreemap::new();
+ if let (true, Some((_entry_key, value))) = (found, entry_iter.next()) {
+ insert_row_ids_into(value, &mut result)?;
+ }
+ Ok(result)
}
/// Less than query.
@@ -334,11 +420,21 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexReader<F>
{
return Ok(RoaringTreemap::new());
}
+ let mut bloom_matches = Vec::with_capacity(sorted_keys.len());
+ for key in sorted_keys {
+ if self.bloom_might_contain(key).await? {
+ bloom_matches.push(key);
+ }
+ }
+ if bloom_matches.is_empty() {
+ return Ok(RoaringTreemap::new());
+ }
+
// The index block is already resident in memory. Resolve every target
key to its first
// possible data block and coalesce adjacent targets that share the
same block handle.
let index_block = self.sst_reader.index_block();
let mut target_blocks: Vec<(BlockHandle, Vec<&[u8]>)> = Vec::new();
- for key in sorted_keys {
+ for key in bloom_matches {
let (_, mut index_iter) = index_block.seek_and_iter(key, cmp);
let Some((_last_key, handle_bytes)) = index_iter.next() else {
break;
diff --git a/crates/paimon/src/btree/sst_file.rs
b/crates/paimon/src/btree/sst_file.rs
index 988c32c8..5695a21a 100644
--- a/crates/paimon/src/btree/sst_file.rs
+++ b/crates/paimon/src/btree/sst_file.rs
@@ -21,7 +21,10 @@ use crate::btree::block::{
compress_block, compute_crc32, decompress_block, BlockCompressionType,
BlockHandle,
BlockReader, BlockTrailer, BlockWriter, BLOCK_HANDLE_MAX_ENCODED_LENGTH,
BLOCK_TRAILER_LENGTH,
};
+use crate::btree::bloom_filter::BloomFilter;
+use crate::btree::footer::BloomFilterHandle;
use crate::io::FileWrite;
+use crate::spec::murmur_hash::hash_bytes;
use bytes::Bytes;
use std::io;
@@ -42,6 +45,7 @@ pub struct SstFileWriter {
compression_level: i32,
last_key: Option<Vec<u8>>,
record_count: u64,
+ bloom_hashes: Option<Vec<i32>>,
}
impl SstFileWriter {
@@ -70,9 +74,24 @@ impl SstFileWriter {
compression_level,
last_key: None,
record_count: 0,
+ bloom_hashes: None,
}
}
+ /// Create a writer with optional Bloom filter collection.
+ pub(super) fn with_bloom_filter(
+ writer: Box<dyn FileWrite>,
+ block_size: usize,
+ compression_type: BlockCompressionType,
+ compression_level: i32,
+ bloom_filter_enabled: bool,
+ ) -> Self {
+ let mut writer =
+ Self::with_compression_level(writer, block_size, compression_type,
compression_level);
+ writer.bloom_hashes = bloom_filter_enabled.then(Vec::new);
+ writer
+ }
+
/// Current write position in the output.
pub fn position(&self) -> u64 {
self.bytes_written
@@ -91,6 +110,9 @@ impl SstFileWriter {
/// Put a key-value pair. Keys must be monotonically increasing.
pub async fn put(&mut self, key: &[u8], value: &[u8]) -> io::Result<()> {
self.data_block_writer.add(key, value);
+ if let Some(hashes) = &mut self.bloom_hashes {
+ hashes.push(hash_bytes(key));
+ }
// Only clone key if it changed
match &self.last_key {
@@ -161,6 +183,25 @@ impl SstFileWriter {
Ok(block_handle)
}
+ /// Build and write the optional Bloom filter from the keys collected by
[`Self::put`].
+ pub async fn write_bloom_filter(&mut self, fpp: f64) ->
io::Result<Option<BloomFilterHandle>> {
+ let Some(hashes) = self.bloom_hashes.take() else {
+ return Ok(None);
+ };
+ let Some(filter) = BloomFilter::from_hashes(&hashes, fpp)? else {
+ return Ok(None);
+ };
+ let handle = BloomFilterHandle {
+ offset: self.position(),
+ size: u32::try_from(filter.bytes().len()).map_err(|_| {
+ io::Error::new(io::ErrorKind::InvalidData, "Bloom filter
exceeds u32 size")
+ })?,
+ expected_entries: filter.expected_entries(),
+ };
+ self.write_bytes(filter.bytes()).await?;
+ Ok(Some(handle))
+ }
+
/// Write raw bytes (e.g., footer, null bitmap).
pub async fn write_raw(&mut self, data: &[u8]) -> io::Result<()> {
self.write_bytes(data).await
diff --git a/crates/paimon/src/btree/tests.rs b/crates/paimon/src/btree/tests.rs
index 252ff883..89ac793b 100644
--- a/crates/paimon/src/btree/tests.rs
+++ b/crates/paimon/src/btree/tests.rs
@@ -16,12 +16,15 @@
// under the License.
use crate::btree::block::BlockCompressionType;
+use crate::btree::bloom_filter::BloomFilter;
+use crate::btree::footer::{BTreeFileFooter, BTREE_FOOTER_ENCODED_LENGTH};
use crate::btree::meta::BTreeIndexMeta;
use crate::btree::query::IndexQuery;
use crate::btree::reader::BTreeIndexReader;
use crate::btree::test_util::{BytesFileRead, VecFileWrite};
use crate::btree::writer::BTreeIndexWriter;
use crate::io::FileRead;
+use crate::spec::murmur_hash::hash_bytes;
use crate::spec::{DataType, Datum, PredicateOperator, VarCharType};
use bytes::Bytes;
use std::ops::Range;
@@ -91,6 +94,120 @@ async fn test_write_read_roundtrip() {
assert_eq!(all.len(), 100);
}
+fn read_footer(data: &[u8]) -> BTreeFileFooter {
+ BTreeFileFooter::read_footer(&data[data.len() -
BTREE_FOOTER_ENCODED_LENGTH..]).unwrap()
+}
+
+#[tokio::test]
+async fn test_bloom_filter_is_disabled_by_default() {
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::new(Box::new(buf.clone()), 64,
BlockCompressionType::None);
+ writer.write(Some(&int_key(1)), 0).await.unwrap();
+ writer.finish().await.unwrap();
+
+ assert!(read_footer(&buf.to_vec()).bloom_filter_handle.is_none());
+}
+
+#[tokio::test]
+async fn test_bloom_filter_skips_missing_point_lookup_data_reads() {
+ let buf = VecFileWrite::new();
+ let mut writer = BTreeIndexWriter::with_comparator_and_options(
+ Box::new(buf.clone()),
+ 64,
+ BlockCompressionType::None,
+ 1,
+ true,
+ int_cmp,
+ );
+ for i in 0..1000 {
+ let key = int_key(i * 2);
+ writer.write(Some(&key), i as i64).await.unwrap();
+ if i == 4 {
+ writer.write(Some(&key), 1000).await.unwrap();
+ }
+ }
+ let write_result = writer.finish().await.unwrap();
+ let data = Bytes::from(buf.to_vec());
+ let footer = read_footer(&data);
+ let handle = footer
+ .bloom_filter_handle
+ .expect("enabled non-empty index must write a Bloom filter");
+ assert_eq!(handle.expected_entries, 1000, "one entry per distinct key");
+
+ let bloom = BloomFilter::from_bytes(
+ handle.expected_entries,
+ data.slice(handle.offset as usize..handle.offset as usize +
handle.size as usize),
+ )
+ .unwrap();
+ let missing_key = (1..2000)
+ .step_by(2)
+ .map(int_key)
+ .find(|key| !bloom.test_hash(hash_bytes(key)))
+ .expect("test data should contain a Bloom-negative missing key");
+
+ for use_in in [false, true] {
+ let ranges = Arc::new(Mutex::new(Vec::new()));
+ let reader = BTreeIndexReader::open(
+ Box::new(RecordingFileRead {
+ data: data.clone(),
+ ranges: ranges.clone(),
+ }),
+ data.len() as u64,
+ &write_result.meta,
+ int_cmp,
+ )
+ .await
+ .unwrap();
+ ranges.lock().unwrap().clear();
+
+ let rows = if use_in {
+ reader.query_in(&[missing_key.as_slice()]).await.unwrap()
+ } else {
+ reader.query_equal(&missing_key).await.unwrap()
+ };
+ assert!(rows.is_empty());
+ assert_eq!(
+ *ranges.lock().unwrap(),
+ vec![handle.offset..handle.offset + u64::from(handle.size)]
+ );
+ }
+
+ let ranges = Arc::new(Mutex::new(Vec::new()));
+ let reader = BTreeIndexReader::open(
+ Box::new(RecordingFileRead {
+ data: data.clone(),
+ ranges: ranges.clone(),
+ }),
+ data.len() as u64,
+ &write_result.meta,
+ int_cmp,
+ )
+ .await
+ .unwrap();
+ ranges.lock().unwrap().clear();
+ let in_keys = [missing_key.as_slice()];
+ let (equal_rows, in_rows) =
+ tokio::join!(reader.query_equal(&missing_key),
reader.query_in(&in_keys));
+ assert!(equal_rows.unwrap().is_empty());
+ assert!(in_rows.unwrap().is_empty());
+ assert_eq!(
+ *ranges.lock().unwrap(),
+ vec![handle.offset..handle.offset + u64::from(handle.size)],
+ "concurrent point lookups should share one lazy Bloom read"
+ );
+
+ let reader = write_and_open(&buf, &write_result, int_cmp).await;
+ assert_eq!(
+ reader
+ .query_equal(&int_key(8))
+ .await
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![4, 1000]
+ );
+}
+
#[tokio::test]
async fn test_duplicate_keys() {
let buf = VecFileWrite::new();
diff --git a/crates/paimon/src/btree/writer.rs
b/crates/paimon/src/btree/writer.rs
index aa6ab2bb..6743174f 100644
--- a/crates/paimon/src/btree/writer.rs
+++ b/crates/paimon/src/btree/writer.rs
@@ -31,6 +31,8 @@ use roaring::RoaringTreemap;
use std::cmp::Ordering;
use std::io;
+const BLOOM_FILTER_FPP: f64 = 0.05;
+
/// BTree index writer. Writes sorted key -> row_id_list entries into a BTree
index file.
///
/// Usage:
@@ -103,13 +105,33 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F>
{
compression_type: BlockCompressionType,
compression_level: i32,
cmp: F,
+ ) -> Self {
+ Self::with_comparator_and_options(
+ writer,
+ block_size,
+ compression_type,
+ compression_level,
+ false,
+ cmp,
+ )
+ }
+
+ /// Create a writer with all Java-compatible BTree file options.
+ pub fn with_comparator_and_options(
+ writer: Box<dyn FileWrite>,
+ block_size: usize,
+ compression_type: BlockCompressionType,
+ compression_level: i32,
+ bloom_filter_enabled: bool,
+ cmp: F,
) -> Self {
Self {
- sst_writer: SstFileWriter::with_compression_level(
+ sst_writer: SstFileWriter::with_bloom_filter(
writer,
block_size,
compression_type,
compression_level,
+ bloom_filter_enabled,
),
current_row_ids: Vec::new(),
last_key: None,
@@ -208,8 +230,8 @@ impl<F: Fn(&[u8], &[u8]) -> Ordering> BTreeIndexWriter<F> {
// Write null bitmap
let null_bitmap_handle = self.write_null_bitmap().await?;
- // No bloom filter for now (same as Java: todo)
- let bloom_filter_handle = None;
+ // Write the optional Bloom filter after the null bitmap and before
the index block.
+ let bloom_filter_handle =
self.sst_writer.write_bloom_filter(BLOOM_FILTER_FPP).await?;
// Write index block
let index_block_handle = self.sst_writer.write_index_block().await?;
diff --git a/crates/paimon/src/spec/murmur_hash.rs
b/crates/paimon/src/spec/murmur_hash.rs
index 3029e67b..08f228ce 100644
--- a/crates/paimon/src/spec/murmur_hash.rs
+++ b/crates/paimon/src/spec/murmur_hash.rs
@@ -68,6 +68,24 @@ pub fn hash_by_words(data: &[u8]) -> i32 {
fmix(h1 ^ data.len() as u32) as i32
}
+/// Murmur3 32-bit hash over arbitrary bytes.
+///
+/// This matches Java Paimon's `MurmurHashUtils.hashBytes` with `DEFAULT_SEED
= 42`.
+/// In particular, Java mixes each trailing signed byte independently instead
of combining the
+/// tail bytes into one word as the canonical Murmur3 algorithm does.
+pub(crate) fn hash_bytes(data: &[u8]) -> i32 {
+ let (words, tail) = data.as_chunks::<4>();
+ let mut h1 = DEFAULT_SEED;
+ for word in words {
+ h1 = mix_h1(h1, mix_k1(u32::from_le_bytes(*word)));
+ }
+ for byte in tail {
+ let signed = i32::from(*byte as i8) as u32;
+ h1 = mix_h1(h1, mix_k1(signed));
+ }
+ fmix(h1 ^ data.len() as u32) as i32
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -89,6 +107,14 @@ mod tests {
assert_eq!(h, expected);
}
+ #[test]
+ fn test_hash_bytes_matches_java() {
+ assert_eq!(hash_bytes(b"a"), 1_485_273_170);
+ assert_eq!(hash_bytes(b"hello"), -1_008_564_952);
+ assert_eq!(hash_bytes(b"world"), -623_458_850);
+ assert_eq!(hash_bytes(&[0xff]), 1_398_487_324);
+ }
+
#[test]
#[should_panic(expected = "word-aligned")]
fn test_hash_non_aligned_panics() {
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index db58c4bc..fff014d5 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -16,7 +16,7 @@
// under the License.
use super::*;
-use crate::btree::BTreeIndexMeta;
+use crate::btree::{BTreeFileFooter, BTreeIndexMeta};
use crate::catalog::Identifier;
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
@@ -677,7 +677,12 @@ async fn scan_ids(table: &Table, predicate: Predicate) ->
Vec<i32> {
#[tokio::test]
async fn test_execute_writes_btree_index_manifest_and_file() {
let table_path = "memory:/test_btree_global_index_builder_e2e";
- let table = test_table_with_path(table_path, table_options("10"));
+ let mut options = table_options("10");
+ options.insert(
+ "btree-index.bloom-filter.enabled".to_string(),
+ "true".to_string(),
+ );
+ let table = test_table_with_path(table_path, options);
setup_dirs(&table).await;
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
@@ -721,6 +726,23 @@ async fn
test_execute_writes_btree_index_manifest_and_file() {
assert_eq!(index_file.row_count, 3);
assert!(index_file.file_size > 0);
+ let index_path = format!("{table_path}/index/{}", index_file.file_name);
+ let index_bytes = table
+ .file_io()
+ .new_input(&index_path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let footer = BTreeFileFooter::read_footer(
+ &index_bytes[index_bytes.len() - BTreeFileFooter::ENCODED_LENGTH..],
+ )
+ .unwrap();
+ let bloom_handle = footer
+ .bloom_filter_handle
+ .expect("enabled BTree build must write a Bloom filter");
+ assert_eq!(bloom_handle.expected_entries, 2);
+
let global_meta = index_file
.global_index_meta
.as_ref()
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index 8e412c07..1b7996ae 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -105,11 +105,12 @@ impl SortedGlobalIndexBuildBuilder<'_> {
let GlobalIndexWriteOptions::Sorted(write_options) =
write_options else {
unreachable!("BTree uses sorted write options")
};
- let mut writer =
BTreeIndexWriter::with_comparator_and_compression_level(
+ let mut writer =
BTreeIndexWriter::with_comparator_and_options(
writer,
write_options.block_size,
write_options.compression_type,
write_options.compression_level,
+ write_options.bloom_filter_enabled,
cmp,
);
for (key, local_row_id) in &rows {
diff --git a/crates/paimon/src/table/sorted_global_index_options.rs
b/crates/paimon/src/table/sorted_global_index_options.rs
index 982a95cd..e7a16422 100644
--- a/crates/paimon/src/table/sorted_global_index_options.rs
+++ b/crates/paimon/src/table/sorted_global_index_options.rs
@@ -24,6 +24,7 @@ use crate::{Error, Result};
use std::collections::HashMap;
pub(crate) const BTREE_BLOCK_SIZE_OPTION: &str = "btree-index.block-size";
+pub(crate) const BTREE_BLOOM_FILTER_ENABLED_OPTION: &str =
"btree-index.bloom-filter.enabled";
pub(crate) const BTREE_COMPRESSION_OPTION: &str = "btree-index.compression";
pub(crate) const BTREE_COMPRESSION_LEVEL_OPTION: &str =
"btree-index.compression-level";
pub(crate) const BITMAP_DICTIONARY_BLOCK_SIZE_OPTION: &str =
"bitmap-index.dictionary-block-size";
@@ -43,6 +44,7 @@ pub(crate) struct SortedIndexWriteOptions {
pub(crate) block_size: usize,
pub(crate) compression_type: BlockCompressionType,
pub(crate) compression_level: i32,
+ pub(crate) bloom_filter_enabled: bool,
}
impl SortedIndexWriteOptions {
@@ -81,10 +83,29 @@ impl SortedIndexWriteOptions {
compression_level_option,
DEFAULT_COMPRESSION_LEVEL,
)?,
+ bloom_filter_enabled: if index_type == BTREE_GLOBAL_INDEX_TYPE {
+ parse_boolean(options, BTREE_BLOOM_FILTER_ENABLED_OPTION,
false)?
+ } else {
+ false
+ },
})
}
}
+fn parse_boolean(options: &HashMap<String, String>, option: &str, default:
bool) -> Result<bool> {
+ let Some(raw) = options.get(option) else {
+ return Ok(default);
+ };
+ match raw.trim().to_ascii_lowercase().as_str() {
+ "true" => Ok(true),
+ "false" => Ok(false),
+ _ => Err(Error::DataInvalid {
+ message: format!("Option '{option}' must be true or false, got:
{raw}"),
+ source: None,
+ }),
+ }
+}
+
fn parse_block_size(
options: &HashMap<String, String>,
option: &str,
@@ -163,6 +184,7 @@ mod tests {
block_size: 64 * 1024,
compression_type: BlockCompressionType::None,
compression_level: 1,
+ bloom_filter_enabled: false,
}
);
for index_type in [BITMAP_GLOBAL_INDEX_TYPE,
MULTIVALUE_GLOBAL_INDEX_TYPE] {
@@ -172,6 +194,7 @@ mod tests {
block_size: 16 * 1024,
compression_type: BlockCompressionType::None,
compression_level: 1,
+ bloom_filter_enabled: false,
}
);
}
@@ -210,9 +233,20 @@ mod tests {
block_size: 32 * 1024,
compression_type: BlockCompressionType::Lz4,
compression_level: 7,
+ bloom_filter_enabled: false,
}
);
}
+
+ let options = HashMap::from([(
+ BTREE_BLOOM_FILTER_ENABLED_OPTION.to_string(),
+ " TrUe ".to_string(),
+ )]);
+ assert!(
+ SortedIndexWriteOptions::from_options(BTREE_GLOBAL_INDEX_TYPE,
&options)
+ .unwrap()
+ .bloom_filter_enabled
+ );
}
#[test]
@@ -234,6 +268,11 @@ mod tests {
BTREE_COMPRESSION_LEVEL_OPTION,
"fast",
),
+ (
+ BTREE_GLOBAL_INDEX_TYPE,
+ BTREE_BLOOM_FILTER_ENABLED_OPTION,
+ "yes",
+ ),
] {
let options = HashMap::from([(option.to_string(),
value.to_string())]);
assert!(SortedIndexWriteOptions::from_options(index_type,
&options).is_err());
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 85ff5b3c..76b56034 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1077,7 +1077,7 @@ CALL sys.create_global_index(
table => 'paimon.my_db.my_table',
index_column => 'id',
index_type => 'btree',
- options =>
'btree-index.block-size=64kb,btree-index.compression=zstd,btree-index.compression-level=1'
+ options =>
'btree-index.block-size=64kb,btree-index.bloom-filter.enabled=true,btree-index.compression=zstd,btree-index.compression-level=1'
);
CALL sys.create_global_index(
@@ -1114,8 +1114,9 @@ row are indexed once. All three sorted index types accept
are `btree-index.block-size`, `bitmap-index.dictionary-block-size`, or
`multivalue-index.dictionary-block-size`, together with the corresponding
`*.compression` (`none`, `zstd`, `lz4`, or `lzo`) and `*.compression-level`
-options. Per-call options override table options. Bitmap and multivalue global
-indexes use Java-compatible bitmap files.
+options. BTree additionally accepts `btree-index.bloom-filter.enabled` (default
+`false`) to accelerate equality and `IN` lookups. Per-call options override
table
+options. Bitmap and multivalue global indexes use Java-compatible bitmap files.
FM global indexes support character-string columns and exact byte-substring
`contains`, `IS NULL`, and `IS NOT NULL` predicates. They use the
@@ -2404,6 +2405,7 @@ deletion vectors enabled.
| `global-index.row-count-per-shard` | `100000` | Maximum row count per vector
global-index shard. |
| `sorted-index.records-per-range` | `100000` | Maximum row count per BTree,
bitmap, multivalue, or FM global-index file range; falls back to legacy
`btree-index.records-per-range`. |
| `btree-index.block-size` | `64kb` | Target BTree data-block size. |
+| `btree-index.bloom-filter.enabled` | `false` | Writes a Bloom filter used to
avoid BTree data-block reads for missing equality and `IN` keys. |
| `btree-index.compression` | `none` | BTree block compression: `none`,
`zstd`, `lz4`, or `lzo`. |
| `btree-index.compression-level` | `1` | BTree compression level (used by
codecs that support levels). |
| `bitmap-index.dictionary-block-size` | `16kb` | Target bitmap
dictionary-block size. |
