From dc88d79cc14db1e05f5dd4785d5abea8ee8aed37 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Tue, 25 Aug 2026 11:24:56 +0800 Subject: [PATCH] feat(file_index): add Java-compatible Bloom filter index Signed-off-by: QuakeWang --- Cargo.lock | 1 + crates/paimon/Cargo.toml | 1 + .../bloom_filter/bloom_filter_64.rs | 241 +++++++ .../src/file_index/bloom_filter/fast_hash.rs | 328 +++++++++ .../paimon/src/file_index/bloom_filter/mod.rs | 675 ++++++++++++++++++ crates/paimon/src/file_index/mod.rs | 5 +- 6 files changed, 1250 insertions(+), 1 deletion(-) create mode 100644 crates/paimon/src/file_index/bloom_filter/bloom_filter_64.rs create mode 100644 crates/paimon/src/file_index/bloom_filter/fast_hash.rs create mode 100644 crates/paimon/src/file_index/bloom_filter/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 3eebe366..dbf1a2ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4622,6 +4622,7 @@ dependencies = [ "tempfile", "tokio", "tokio-util", + "twox-hash", "typed-builder", "unicode-segmentation", "url", diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index ae96e242..dc09b964 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -125,6 +125,7 @@ hex = "0.4" hmac = "0.12" sha1 = "0.10" sha2 = "0.10" +twox-hash = { version = "2.1.3", default-features = false, features = ["xxhash64"] } md-5 = "0.10" regex = "1" uuid = { version = "1", features = ["v4"] } diff --git a/crates/paimon/src/file_index/bloom_filter/bloom_filter_64.rs b/crates/paimon/src/file_index/bloom_filter/bloom_filter_64.rs new file mode 100644 index 00000000..3bd51c18 --- /dev/null +++ b/crates/paimon/src/file_index/bloom_filter/bloom_filter_64.rs @@ -0,0 +1,241 @@ +// 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. + +use bytes::Bytes; + +use crate::{Error, Result}; + +const BITS_PER_BYTE: usize = u8::BITS as usize; + +pub(super) struct BloomFilter64 { + bit_set: BitSet, + num_hash_functions: i32, +} + +impl BloomFilter64 { + pub(super) fn try_new(items: i32, fpp: f64) -> Result { + if items <= 0 { + return Err(Error::ConfigInvalid { + message: format!("Bloom filter items must be positive, but was {items}"), + }); + } + if !fpp.is_finite() || fpp <= 0.0 || fpp >= 1.0 { + return Err(Error::ConfigInvalid { + message: format!("Bloom filter fpp must be finite and in (0, 1), but was {fpp}"), + }); + } + + let log_two = 2.0_f64.ln(); + let estimated_bits = -(f64::from(items) * fpp.ln()) / (log_two * log_two); + if !estimated_bits.is_finite() || estimated_bits > f64::from(i32::MAX) { + return Err(Error::ConfigInvalid { + message: format!( + "Bloom filter size is not representable for items {items} and fpp {fpp}" + ), + }); + } + + // Java truncates the estimate and always advances to the next byte boundary. + let estimated_bits = estimated_bits as i32; + let num_bits = estimated_bits + .checked_add(8 - estimated_bits.rem_euclid(8)) + .ok_or_else(|| Error::ConfigInvalid { + message: format!("Bloom filter size overflows for items {items} and fpp {fpp}"), + })?; + let num_hash_functions = ((f64::from(num_bits) / f64::from(items)) * log_two) + .round() + .max(1.0) as i32; + + Ok(Self { + bit_set: BitSet::try_zeroed(num_bits as usize)?, + num_hash_functions, + }) + } + + pub(super) fn from_serialized(num_hash_functions: i32, bytes: Bytes) -> Result { + if bytes.is_empty() { + return Err(Error::FileIndexFormatInvalid { + message: "Bloom filter bitset must not be empty".to_string(), + }); + } + let bit_size = bytes.len().checked_mul(BITS_PER_BYTE).ok_or_else(|| { + Error::FileIndexFormatInvalid { + message: "Bloom filter bit count overflows usize".to_string(), + } + })?; + if num_hash_functions <= 0 { + return Err(Error::FileIndexFormatInvalid { + message: format!( + "Bloom filter hash function count must be positive, but was {num_hash_functions}" + ), + }); + } + if num_hash_functions as usize > bit_size { + return Err(Error::FileIndexFormatInvalid { + message: format!( + "Bloom filter hash function count {num_hash_functions} exceeds bit count {}", + bit_size + ), + }); + } + let bit_set = BitSet::Shared(bytes); + + Ok(Self { + bit_set, + num_hash_functions, + }) + } + + pub(super) fn add_hash(&mut self, hash64: u64) { + for iteration in 1..=self.num_hash_functions { + let position = self.position(hash64, iteration); + self.bit_set.set(position); + } + } + + pub(super) fn test_hash(&self, hash64: u64) -> bool { + (1..=self.num_hash_functions).all(|iteration| { + let position = self.position(hash64, iteration); + self.bit_set.get(position) + }) + } + + pub(super) fn num_hash_functions(&self) -> i32 { + self.num_hash_functions + } + + pub(super) fn bytes(&self) -> &[u8] { + self.bit_set.bytes() + } + + fn position(&self, hash64: u64, iteration: i32) -> usize { + let hash1 = hash64 as i32; + let hash2 = (hash64 >> 32) as i32; + let mut combined_hash = hash1.wrapping_add(iteration.wrapping_mul(hash2)); + if combined_hash < 0 { + combined_hash = !combined_hash; + } + combined_hash as usize % self.bit_set.bit_size() + } +} + +enum BitSet { + Mutable(Vec), + Shared(Bytes), +} + +impl BitSet { + fn try_zeroed(num_bits: usize) -> Result { + debug_assert!(num_bits > 0 && num_bits.is_multiple_of(BITS_PER_BYTE)); + let len = num_bits / BITS_PER_BYTE; + let mut bytes = Vec::new(); + bytes + .try_reserve_exact(len) + .map_err(|error| Error::ConfigInvalid { + message: format!("Failed to allocate {len} bytes for Bloom filter: {error}"), + })?; + bytes.resize(len, 0); + Ok(Self::Mutable(bytes)) + } + + fn set(&mut self, index: usize) { + match self { + Self::Mutable(bytes) => bytes[index >> 3] |= 1 << (index & 0x07), + Self::Shared(_) => unreachable!("serialized Bloom filter bitset is immutable"), + } + } + + fn get(&self, index: usize) -> bool { + self.bytes()[index >> 3] & (1 << (index & 0x07)) != 0 + } + + fn bit_size(&self) -> usize { + self.bytes().len() * BITS_PER_BYTE + } + + fn bytes(&self) -> &[u8] { + match self { + Self::Mutable(bytes) => bytes, + Self::Shared(bytes) => bytes, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_java_byte_alignment() { + let filter = BloomFilter64::try_new(82, 0.1).unwrap(); + + assert_eq!(filter.bytes().len(), 50); + assert_eq!(filter.num_hash_functions(), 3); + } + + #[test] + fn test_add_and_test_hash() { + let mut filter = BloomFilter64::try_new(10, 0.1).unwrap(); + let hash = 0x1234_5678_90ab_cdef; + + assert!(!filter.test_hash(hash)); + filter.add_hash(hash); + assert!(filter.test_hash(hash)); + } + + #[test] + fn test_reject_invalid_config() { + for items in [i32::MIN, -1, 0] { + assert!(matches!( + BloomFilter64::try_new(items, 0.1), + Err(Error::ConfigInvalid { .. }) + )); + } + for fpp in [ + f64::NEG_INFINITY, + -0.1, + 0.0, + 1.0, + 2.0, + f64::INFINITY, + f64::NAN, + ] { + assert!(matches!( + BloomFilter64::try_new(10, fpp), + Err(Error::ConfigInvalid { .. }) + )); + } + assert!(matches!( + BloomFilter64::try_new(i32::MAX, f64::MIN_POSITIVE), + Err(Error::ConfigInvalid { .. }) + )); + } + + #[test] + fn test_reject_malformed_serialized_parts() { + for (hash_functions, bytes) in [(0, &b"\0"[..]), (-1, &b"\0"[..]), (9, &b"\0"[..])] { + assert!(matches!( + BloomFilter64::from_serialized(hash_functions, Bytes::copy_from_slice(bytes)), + Err(Error::FileIndexFormatInvalid { .. }) + )); + } + assert!(matches!( + BloomFilter64::from_serialized(1, Bytes::new()), + Err(Error::FileIndexFormatInvalid { .. }) + )); + } +} diff --git a/crates/paimon/src/file_index/bloom_filter/fast_hash.rs b/crates/paimon/src/file_index/bloom_filter/fast_hash.rs new file mode 100644 index 00000000..9eb98588 --- /dev/null +++ b/crates/paimon/src/file_index/bloom_filter/fast_hash.rs @@ -0,0 +1,328 @@ +// 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. + +use twox_hash::XxHash64; + +use crate::spec::{DataType, Datum}; +use crate::{Error, Result}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum FastHash { + String, + Bytes, + TinyInt, + SmallInt, + Int, + BigInt, + Float, + Double, + Date, + Time, + TimestampMillis, + TimestampMicros, + LocalZonedTimestampMillis, + LocalZonedTimestampMicros, +} + +impl FastHash { + pub(super) fn try_new(data_type: &DataType) -> Result { + match data_type { + DataType::Char(_) | DataType::VarChar(_) => Ok(Self::String), + DataType::Binary(_) | DataType::VarBinary(_) => Ok(Self::Bytes), + DataType::TinyInt(_) => Ok(Self::TinyInt), + DataType::SmallInt(_) => Ok(Self::SmallInt), + DataType::Int(_) => Ok(Self::Int), + DataType::BigInt(_) => Ok(Self::BigInt), + DataType::Float(_) => Ok(Self::Float), + DataType::Double(_) => Ok(Self::Double), + DataType::Date(_) => Ok(Self::Date), + DataType::Time(_) => Ok(Self::Time), + DataType::Timestamp(timestamp_type) if timestamp_type.precision() <= 3 => { + Ok(Self::TimestampMillis) + } + DataType::Timestamp(_) => Ok(Self::TimestampMicros), + DataType::LocalZonedTimestamp(timestamp_type) if timestamp_type.precision() <= 3 => { + Ok(Self::LocalZonedTimestampMillis) + } + DataType::LocalZonedTimestamp(_) => Ok(Self::LocalZonedTimestampMicros), + _ => Err(Error::Unsupported { + message: format!("Bloom filter does not support data type {data_type:?}"), + }), + } + } + + pub(super) fn hash(self, datum: &Datum) -> Result { + match (self, datum) { + (Self::String, Datum::String(value)) => Ok(XxHash64::oneshot(0, value.as_bytes())), + (Self::Bytes, Datum::Bytes(value)) => Ok(XxHash64::oneshot(0, value)), + (Self::TinyInt, Datum::TinyInt(value)) => Ok(wang_hash(i64::from(*value))), + (Self::SmallInt, Datum::SmallInt(value)) => Ok(wang_hash(i64::from(*value))), + (Self::Int, Datum::Int(value)) => Ok(wang_hash(i64::from(*value))), + (Self::BigInt, Datum::Long(value)) => Ok(wang_hash(*value)), + (Self::Float, Datum::Float(value)) => { + Ok(wang_hash(i64::from(java_float_to_int_bits(*value)))) + } + (Self::Double, Datum::Double(value)) => Ok(wang_hash(java_double_to_long_bits(*value))), + (Self::Date, Datum::Date(value)) => Ok(wang_hash(i64::from(*value))), + (Self::Time, Datum::Time(value)) => Ok(wang_hash(i64::from(*value))), + (Self::TimestampMillis, Datum::Timestamp { millis, nanos }) => { + hash_timestamp(false, *millis, *nanos) + } + (Self::TimestampMicros, Datum::Timestamp { millis, nanos }) => { + hash_timestamp(true, *millis, *nanos) + } + (Self::LocalZonedTimestampMillis, Datum::LocalZonedTimestamp { millis, nanos }) => { + hash_timestamp(false, *millis, *nanos) + } + (Self::LocalZonedTimestampMicros, Datum::LocalZonedTimestamp { millis, nanos }) => { + hash_timestamp(true, *millis, *nanos) + } + _ => Err(Error::DataInvalid { + message: format!("Datum {datum:?} does not match Bloom hash strategy {self:?}"), + source: None, + }), + } + } +} + +fn hash_timestamp(micros: bool, millis: i64, nanos: i32) -> Result { + if !(0..=999_999).contains(&nanos) { + return Err(Error::DataInvalid { + message: format!("Timestamp nanos-of-millisecond is out of range: {nanos}"), + source: None, + }); + } + if !micros { + return Ok(wang_hash(millis)); + } + + let micros = millis + .checked_mul(1_000) + .and_then(|value| value.checked_add(i64::from(nanos / 1_000))) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "Timestamp cannot be represented in microseconds: millis={millis}, nanos={nanos}" + ), + source: None, + })?; + Ok(wang_hash(micros)) +} + +fn java_float_to_int_bits(value: f32) -> i32 { + if value.is_nan() { + 0x7fc0_0000_u32 as i32 + } else { + value.to_bits() as i32 + } +} + +fn java_double_to_long_bits(value: f64) -> i64 { + if value.is_nan() { + 0x7ff8_0000_0000_0000_u64 as i64 + } else { + value.to_bits() as i64 + } +} + +fn wang_hash(mut key: i64) -> u64 { + key = (!key).wrapping_add(key.wrapping_shl(21)); + key ^= key >> 24; + key = key + .wrapping_add(key.wrapping_shl(3)) + .wrapping_add(key.wrapping_shl(8)); + key ^= key >> 14; + key = key + .wrapping_add(key.wrapping_shl(2)) + .wrapping_add(key.wrapping_shl(4)); + key ^= key >> 28; + key = key.wrapping_add(key.wrapping_shl(31)); + key as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{ + BigIntType, BinaryType, CharType, DateType, DoubleType, FloatType, IntType, + LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType, + VarCharType, + }; + + #[test] + fn test_java_hash_fixtures() { + let fixtures = [ + ( + DataType::Char(CharType::new(20).unwrap()), + Datum::String("Paimon-派蒙".to_string()), + 0xb09c_177e_1aaf_64c1, + ), + ( + DataType::VarChar(VarCharType::new(20).unwrap()), + Datum::String("Paimon-派蒙".to_string()), + 0xb09c_177e_1aaf_64c1, + ), + ( + DataType::Binary(BinaryType::new(4).unwrap()), + Datum::Bytes(vec![0x00, 0x01, 0xfe, 0xff]), + 0x662c_71e0_4101_34be, + ), + ( + DataType::VarBinary(VarBinaryType::new(10).unwrap()), + Datum::Bytes(vec![]), + 0xef46_db37_51d8_e999, + ), + ( + DataType::TinyInt(TinyIntType::new()), + Datum::TinyInt(-128), + 0xe547_e844_4a8f_cdd1, + ), + ( + DataType::SmallInt(SmallIntType::new()), + Datum::SmallInt(-12_345), + 0x6a48_82d9_d48f_ffa6, + ), + ( + DataType::Int(IntType::new()), + Datum::Int(-123_456_789), + 0xe60f_1a14_2420_2ebd, + ), + ( + DataType::BigInt(BigIntType::new()), + Datum::Long(i64::MIN + 123), + 0x52d7_f67f_a5ee_3244, + ), + ( + DataType::Float(FloatType::new()), + Datum::Float(f32::from_bits(0x7fa1_2345)), + 0x67c2_7c6d_9936_ae63, + ), + ( + DataType::Float(FloatType::new()), + Datum::Float(-0.0), + 0x111e_c0fd_6aa8_626c, + ), + ( + DataType::Double(DoubleType::new()), + Datum::Double(f64::from_bits(0x7ff1_2345_6789_abcd)), + 0x13d2_d3f2_cc0e_846e, + ), + ( + DataType::Double(DoubleType::new()), + Datum::Double(-0.0), + 0x3be7_d0f7_780d_e548, + ), + ( + DataType::Date(DateType::new()), + Datum::Date(-1), + 0x5bca_8684_3795_0d03, + ), + ( + DataType::Time(TimeType::new(3).unwrap()), + Datum::Time(86_399_999), + 0x0697_db46_7133_6cb5, + ), + ( + DataType::Timestamp(TimestampType::new(3).unwrap()), + Datum::Timestamp { + millis: -123_456_789, + nanos: 0, + }, + 0xe60f_1a14_2420_2ebd, + ), + ( + DataType::Timestamp(TimestampType::new(6).unwrap()), + Datum::Timestamp { + millis: -1, + nanos: 999_000, + }, + 0x5bca_8684_3795_0d03, + ), + ( + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()), + Datum::LocalZonedTimestamp { + millis: 1_700_000_000_123, + nanos: 0, + }, + 0xf52b_5278_fb88_f260, + ), + ( + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(6).unwrap()), + Datum::LocalZonedTimestamp { + millis: 1_700_000_000_123, + nanos: 456_000, + }, + 0x5e6e_89d6_5e49_5754, + ), + ]; + + for (data_type, datum, expected) in fixtures { + assert_eq!( + FastHash::try_new(&data_type).unwrap().hash(&datum).unwrap(), + expected + ); + } + } + + #[test] + fn test_reject_mismatched_datum_and_invalid_timestamp() { + assert!(matches!( + FastHash::try_new(&DataType::BigInt(BigIntType::new())) + .unwrap() + .hash(&Datum::Int(1)), + Err(Error::DataInvalid { .. }) + )); + assert!(matches!( + FastHash::try_new(&DataType::Timestamp(TimestampType::new(6).unwrap())) + .unwrap() + .hash(&Datum::Timestamp { + millis: 0, + nanos: 1_000_000, + }), + Err(Error::DataInvalid { .. }) + )); + } + + #[test] + fn test_hash_strategy_compatibility() { + assert_eq!( + FastHash::try_new(&DataType::Char(CharType::with_nullable(false, 3).unwrap())).unwrap(), + FastHash::try_new(&DataType::VarChar(VarCharType::new(200).unwrap())).unwrap() + ); + assert_eq!( + FastHash::try_new(&DataType::Binary(BinaryType::new(4).unwrap())).unwrap(), + FastHash::try_new(&DataType::VarBinary( + VarBinaryType::try_new(false, 1_024).unwrap() + )) + .unwrap() + ); + assert_eq!( + FastHash::try_new(&DataType::BigInt(BigIntType::new())).unwrap(), + FastHash::try_new(&DataType::BigInt(BigIntType::with_nullable(false))).unwrap() + ); + assert_eq!( + FastHash::try_new(&DataType::Timestamp(TimestampType::new(4).unwrap())).unwrap(), + FastHash::try_new(&DataType::Timestamp( + TimestampType::with_nullable(false, 9).unwrap() + )) + .unwrap() + ); + assert_ne!( + FastHash::try_new(&DataType::Timestamp(TimestampType::new(3).unwrap())).unwrap(), + FastHash::try_new(&DataType::Timestamp(TimestampType::new(4).unwrap())).unwrap() + ); + } +} diff --git a/crates/paimon/src/file_index/bloom_filter/mod.rs b/crates/paimon/src/file_index/bloom_filter/mod.rs new file mode 100644 index 00000000..9d5708ed --- /dev/null +++ b/crates/paimon/src/file_index/bloom_filter/mod.rs @@ -0,0 +1,675 @@ +// 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. + +mod bloom_filter_64; +mod fast_hash; + +use bytes::{Bytes, BytesMut}; + +use crate::common::Options; +use crate::file_index::file_index_reader::FileIndexReader; +use crate::file_index::file_index_result::FileIndexResult; +use crate::spec::{DataType, Datum, PredicateOperator}; +use crate::{Error, Result}; + +use bloom_filter_64::BloomFilter64; +use fast_hash::FastHash; + +const DEFAULT_ITEMS: i32 = 1_000_000; +const DEFAULT_FPP: f64 = 0.1; +const ITEMS: &str = "items"; +const FPP: &str = "fpp"; + +pub(crate) struct BloomFilterWriter { + hash_function: FastHash, + filter: BloomFilter64, +} + +impl BloomFilterWriter { + pub(crate) fn try_new(data_type: DataType, options: &Options) -> Result { + let hash_function = FastHash::try_new(&data_type)?; + let items = parse_option(options, ITEMS, DEFAULT_ITEMS)?; + let fpp = parse_option(options, FPP, DEFAULT_FPP)?; + let filter = BloomFilter64::try_new(items, fpp)?; + Ok(Self { + hash_function, + filter, + }) + } + + pub(crate) fn write(&mut self, datum: Option<&Datum>) -> Result<()> { + if let Some(datum) = datum { + self.filter.add_hash(self.hash_function.hash(datum)?); + } + Ok(()) + } + + pub(crate) fn serialized_bytes(&self) -> Bytes { + let mut serialized = BytesMut::with_capacity(4 + self.filter.bytes().len()); + serialized.extend_from_slice(&self.filter.num_hash_functions().to_be_bytes()); + serialized.extend_from_slice(self.filter.bytes()); + serialized.freeze() + } +} + +pub(crate) struct BloomFilterReader { + hash_function: FastHash, + filter: BloomFilter64, +} + +impl BloomFilterReader { + pub(crate) fn try_new(data_type: DataType, serialized: Bytes) -> Result { + let hash_function = FastHash::try_new(&data_type)?; + let header = serialized + .get(..4) + .ok_or_else(|| Error::FileIndexFormatInvalid { + message: format!( + "Bloom filter payload must contain a 4-byte header, but had {} bytes", + serialized.len() + ), + })?; + let num_hash_functions = i32::from_be_bytes(header.try_into().unwrap()); + let filter = BloomFilter64::from_serialized(num_hash_functions, serialized.slice(4..))?; + Ok(Self { + hash_function, + filter, + }) + } + + fn may_contain_literal(&self, datum: &Datum) -> Result { + let hash = self.hash_function.hash(datum)?; + if self.filter.test_hash(hash) { + return Ok(true); + } + + // Rust predicates currently treat signed zero as equal, while the Java + // hash contract preserves the sign bit. Test the equivalent zero hash + // before pruning without changing the serialized Bloom format. + match (self.hash_function, datum) { + (FastHash::Float, Datum::Float(value)) if *value == 0.0 => { + let opposite = Datum::Float(-*value); + Ok(self.filter.test_hash(self.hash_function.hash(&opposite)?)) + } + (FastHash::Double, Datum::Double(value)) if *value == 0.0 => { + let opposite = Datum::Double(-*value); + Ok(self.filter.test_hash(self.hash_function.hash(&opposite)?)) + } + _ => Ok(false), + } + } + + fn evaluate_literal(&self, datum: &Datum) -> FileIndexResult { + match self.may_contain_literal(datum) { + Ok(false) => FileIndexResult::Skip, + Ok(_) | Err(_) => FileIndexResult::Remain, + } + } +} + +impl FileIndexReader for BloomFilterReader { + fn evaluate( + &self, + _column: &str, + _index: usize, + data_type: &DataType, + operator: PredicateOperator, + literals: &[Datum], + ) -> FileIndexResult { + if FastHash::try_new(data_type).ok() != Some(self.hash_function) { + return FileIndexResult::Remain; + } + + match operator { + PredicateOperator::Eq if literals.len() == 1 => self.evaluate_literal(&literals[0]), + PredicateOperator::In => literals + .iter() + .fold(FileIndexResult::Skip, |result, datum| { + result.or(self.evaluate_literal(datum)) + }), + _ => FileIndexResult::Remain, + } + } +} + +fn parse_option(options: &Options, key: &str, default: T) -> Result +where + T: std::str::FromStr, + T::Err: std::fmt::Display, +{ + match options.get(key) { + Some(value) => value.parse().map_err(|error| Error::ConfigInvalid { + message: format!("Invalid Bloom filter option {key}={value}: {error}"), + }), + None => Ok(default), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::{ + BigIntType, BinaryType, CharType, DateType, DoubleType, FloatType, IntType, + LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType, + VarCharType, + }; + + fn options(items: &str, fpp: &str) -> Options { + let mut options = Options::new(); + options.set(ITEMS, items); + options.set(FPP, fpp); + options + } + + fn evaluate( + reader: &BloomFilterReader, + data_type: &DataType, + operator: PredicateOperator, + literals: &[Datum], + ) -> FileIndexResult { + reader.evaluate("field", 0, data_type, operator, literals) + } + + #[test] + fn test_java_golden_payload_and_predicates() { + let data_type = DataType::BigInt(BigIntType::new()); + let mut writer = + BloomFilterWriter::try_new(data_type.clone(), &options("10", "0.1")).unwrap(); + for datum in [ + Datum::Long(-1), + Datum::Long(0), + Datum::Long(1), + Datum::Long(42), + ] { + writer.write(Some(&datum)).unwrap(); + } + writer.write(None).unwrap(); + + let serialized = writer.serialized_bytes(); + assert_eq!( + serialized.as_ref(), + &hex::decode("00000003818281005001").unwrap() + ); + + let bitset_ptr = serialized[4..].as_ptr(); + let reader = BloomFilterReader::try_new(data_type.clone(), serialized).unwrap(); + assert_eq!(reader.filter.bytes().as_ptr(), bitset_ptr); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(42)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(43)] + ), + FileIndexResult::Skip + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::In, + &[Datum::Long(43), Datum::Long(42)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::In, + &[Datum::Long(43), Datum::Long(44)] + ), + FileIndexResult::Skip + ); + } + + #[test] + fn test_high_hash_count_uses_standard_big_endian_header() { + let data_type = DataType::BigInt(BigIntType::new()); + let mut writer = + BloomFilterWriter::try_new(data_type.clone(), &options("1", "2.938735877055719e-39")) + .unwrap(); + writer.write(Some(&Datum::Long(42))).unwrap(); + + let serialized = writer.serialized_bytes(); + assert_eq!(&serialized[..4], &133_i32.to_be_bytes()); + assert_eq!(serialized.len(), 28); + + let reader = BloomFilterReader::try_new(data_type.clone(), serialized).unwrap(); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(42)] + ), + FileIndexResult::Remain + ); + } + + #[test] + fn test_float_predicates_remain_conservative_for_signed_zero_and_nan() { + let float_type = DataType::Float(FloatType::new()); + let mut writer = + BloomFilterWriter::try_new(float_type.clone(), &options("10", "0.1")).unwrap(); + writer.write(Some(&Datum::Float(0.0))).unwrap(); + writer + .write(Some(&Datum::Float(f32::from_bits(0x7fc0_0001)))) + .unwrap(); + let reader = + BloomFilterReader::try_new(float_type.clone(), writer.serialized_bytes()).unwrap(); + + assert_eq!( + evaluate( + &reader, + &float_type, + PredicateOperator::Eq, + &[Datum::Float(0.0)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &float_type, + PredicateOperator::In, + &[Datum::Float(123.0), Datum::Float(-0.0)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &float_type, + PredicateOperator::Eq, + &[Datum::Float(-0.0)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &float_type, + PredicateOperator::Eq, + &[Datum::Float(f32::from_bits(0xffc0_1234))] + ), + FileIndexResult::Remain + ); + + let double_type = DataType::Double(DoubleType::new()); + let mut writer = + BloomFilterWriter::try_new(double_type.clone(), &options("10", "0.1")).unwrap(); + writer.write(Some(&Datum::Double(-0.0))).unwrap(); + let reader = + BloomFilterReader::try_new(double_type.clone(), writer.serialized_bytes()).unwrap(); + assert_eq!( + evaluate( + &reader, + &double_type, + PredicateOperator::Eq, + &[Datum::Double(0.0)] + ), + FileIndexResult::Remain + ); + } + + #[test] + fn test_hash_compatible_schema_changes_keep_pruning() { + let options = options("10", "0.1"); + + let bigint = DataType::BigInt(BigIntType::new()); + let mut writer = BloomFilterWriter::try_new(bigint.clone(), &options).unwrap(); + writer.write(Some(&Datum::Long(42))).unwrap(); + let reader = BloomFilterReader::try_new(bigint, writer.serialized_bytes()).unwrap(); + assert_eq!( + evaluate( + &reader, + &DataType::BigInt(BigIntType::with_nullable(false)), + PredicateOperator::Eq, + &[Datum::Long(43)] + ), + FileIndexResult::Skip + ); + + let char_type = DataType::Char(CharType::with_nullable(false, 3).unwrap()); + let mut writer = BloomFilterWriter::try_new(char_type.clone(), &options).unwrap(); + writer + .write(Some(&Datum::String("abc".to_string()))) + .unwrap(); + let reader = BloomFilterReader::try_new(char_type, writer.serialized_bytes()).unwrap(); + assert_eq!( + evaluate( + &reader, + &DataType::VarChar(VarCharType::new(200).unwrap()), + PredicateOperator::Eq, + &[Datum::String("zzz".to_string())] + ), + FileIndexResult::Skip + ); + + let timestamp4 = DataType::Timestamp(TimestampType::new(4).unwrap()); + let mut writer = BloomFilterWriter::try_new(timestamp4.clone(), &options).unwrap(); + writer + .write(Some(&Datum::Timestamp { + millis: 1_700_000_000_123, + nanos: 456_000, + })) + .unwrap(); + let reader = BloomFilterReader::try_new(timestamp4, writer.serialized_bytes()).unwrap(); + let missing = [Datum::Timestamp { + millis: 1_700_000_000_124, + nanos: 456_000, + }]; + assert_eq!( + evaluate( + &reader, + &DataType::Timestamp(TimestampType::with_nullable(false, 9).unwrap()), + PredicateOperator::Eq, + &missing + ), + FileIndexResult::Skip + ); + assert_eq!( + evaluate( + &reader, + &DataType::Timestamp(TimestampType::new(3).unwrap()), + PredicateOperator::Eq, + &missing + ), + FileIndexResult::Remain + ); + } + + #[test] + fn test_no_false_negative_for_supported_types() { + let fixtures = [ + ( + DataType::Char(CharType::new(20).unwrap()), + Datum::String("Paimon-派蒙".to_string()), + ), + ( + DataType::VarChar(VarCharType::new(20).unwrap()), + Datum::String("Paimon-派蒙".to_string()), + ), + ( + DataType::Binary(BinaryType::new(4).unwrap()), + Datum::Bytes(vec![0x00, 0x01, 0xfe, 0xff]), + ), + ( + DataType::VarBinary(VarBinaryType::new(10).unwrap()), + Datum::Bytes(vec![]), + ), + (DataType::TinyInt(TinyIntType::new()), Datum::TinyInt(-128)), + ( + DataType::SmallInt(SmallIntType::new()), + Datum::SmallInt(-12_345), + ), + (DataType::Int(IntType::new()), Datum::Int(-123_456_789)), + ( + DataType::BigInt(BigIntType::new()), + Datum::Long(i64::MIN + 123), + ), + ( + DataType::Float(FloatType::new()), + Datum::Float(f32::from_bits(0x7fa1_2345)), + ), + ( + DataType::Double(DoubleType::new()), + Datum::Double(f64::from_bits(0x7ff1_2345_6789_abcd)), + ), + (DataType::Date(DateType::new()), Datum::Date(-1)), + ( + DataType::Time(TimeType::new(3).unwrap()), + Datum::Time(86_399_999), + ), + ( + DataType::Timestamp(TimestampType::new(3).unwrap()), + Datum::Timestamp { + millis: -123_456_789, + nanos: 0, + }, + ), + ( + DataType::Timestamp(TimestampType::new(6).unwrap()), + Datum::Timestamp { + millis: -1, + nanos: 999_000, + }, + ), + ( + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()), + Datum::LocalZonedTimestamp { + millis: 1_700_000_000_123, + nanos: 0, + }, + ), + ( + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(6).unwrap()), + Datum::LocalZonedTimestamp { + millis: 1_700_000_000_123, + nanos: 456_000, + }, + ), + ]; + + for (data_type, datum) in fixtures { + let mut writer = + BloomFilterWriter::try_new(data_type.clone(), &options("10", "0.1")).unwrap(); + writer.write(Some(&datum)).unwrap(); + let reader = + BloomFilterReader::try_new(data_type.clone(), writer.serialized_bytes()).unwrap(); + + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[datum]), + FileIndexResult::Remain + ); + } + } + + #[test] + fn test_fixed_sequence_has_no_false_negatives_and_bounded_fpp() { + fn next_value(state: &mut u64) -> i64 { + *state = state + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + *state as i64 + } + + const ITEMS: usize = 5_000; + let data_type = DataType::BigInt(BigIntType::new()); + let mut writer = + BloomFilterWriter::try_new(data_type.clone(), &options("5000", "0.01")).unwrap(); + let mut state = 0x4d59_5df4_d0f3_3173; + let inserted = (0..ITEMS) + .map(|_| next_value(&mut state)) + .collect::>(); + for value in &inserted { + writer.write(Some(&Datum::Long(*value))).unwrap(); + } + let reader = + BloomFilterReader::try_new(data_type.clone(), writer.serialized_bytes()).unwrap(); + + for value in inserted { + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(value)] + ), + FileIndexResult::Remain + ); + } + + let false_positives = (0..ITEMS) + .filter(|_| { + let value = next_value(&mut state); + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(value)], + ) == FileIndexResult::Remain + }) + .count(); + assert!( + false_positives <= ITEMS * 3 / 100, + "false-positive rate exceeded 3%: {false_positives}/{ITEMS}" + ); + } + + #[test] + fn test_unsupported_predicates_and_invalid_literals_remain() { + let data_type = DataType::BigInt(BigIntType::new()); + let mut writer = + BloomFilterWriter::try_new(data_type.clone(), &options("10", "0.1")).unwrap(); + writer.write(Some(&Datum::Long(42))).unwrap(); + let reader = + BloomFilterReader::try_new(data_type.clone(), writer.serialized_bytes()).unwrap(); + + for operator in [ + PredicateOperator::IsNull, + PredicateOperator::IsNotNull, + PredicateOperator::NotEq, + PredicateOperator::Lt, + PredicateOperator::LtEq, + PredicateOperator::Gt, + PredicateOperator::GtEq, + PredicateOperator::NotIn, + PredicateOperator::StartsWith, + PredicateOperator::EndsWith, + PredicateOperator::Contains, + PredicateOperator::ArrayContains, + PredicateOperator::ArraysOverlap, + PredicateOperator::ArrayContainsAll, + PredicateOperator::Like, + PredicateOperator::Between, + PredicateOperator::NotBetween, + ] { + assert_eq!( + evaluate(&reader, &data_type, operator, &[Datum::Long(43)]), + FileIndexResult::Remain + ); + } + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::Eq, &[]), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(42), Datum::Long(43)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::In, + &[Datum::Long(43), Datum::Int(44)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate(&reader, &data_type, PredicateOperator::In, &[]), + FileIndexResult::Skip + ); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Int(43)] + ), + FileIndexResult::Remain + ); + assert_eq!( + evaluate( + &reader, + &DataType::Int(IntType::new()), + PredicateOperator::Eq, + &[Datum::Long(43)] + ), + FileIndexResult::Remain + ); + } + + #[test] + fn test_strict_config_validation() { + let data_type = DataType::BigInt(BigIntType::new()); + for (items, fpp) in [ + ("0", "0.1"), + ("-1", "0.1"), + ("abc", "0.1"), + ("2147483648", "0.1"), + ("10", "0"), + ("10", "1"), + ("10", "-0.1"), + ("10", "NaN"), + ("10", "inf"), + ("10", "abc"), + ] { + assert!(matches!( + BloomFilterWriter::try_new(data_type.clone(), &options(items, fpp)), + Err(Error::ConfigInvalid { .. }) + )); + } + } + + #[test] + fn test_strict_payload_validation() { + let data_type = DataType::BigInt(BigIntType::new()); + for payload in [ + &b""[..], + &b"\0\0\0"[..], + &b"\0\0\0\x01"[..], + &b"\0\0\0\0\0"[..], + &b"\xff\xff\xff\xff\0"[..], + &b"\0\0\0\x09\0"[..], + ] { + assert!(matches!( + BloomFilterReader::try_new(data_type.clone(), Bytes::copy_from_slice(payload)), + Err(Error::FileIndexFormatInvalid { .. }) + )); + } + + let mut high_hash_count = vec![0, 0, 0, 133]; + high_hash_count.extend_from_slice(&[0; 24]); + let reader = + BloomFilterReader::try_new(data_type.clone(), Bytes::from(high_hash_count)).unwrap(); + assert_eq!( + evaluate( + &reader, + &data_type, + PredicateOperator::Eq, + &[Datum::Long(42)] + ), + FileIndexResult::Skip + ); + } +} diff --git a/crates/paimon/src/file_index/mod.rs b/crates/paimon/src/file_index/mod.rs index c50dfd6b..79eca6ea 100644 --- a/crates/paimon/src/file_index/mod.rs +++ b/crates/paimon/src/file_index/mod.rs @@ -15,8 +15,11 @@ // specific language governing permissions and limitations // under the License. +// Bloom reader/writer and predicate plumbing stay crate-private until the +// factory, data-writer, and scan integration land in later changes. +#[allow(dead_code)] +pub(crate) mod bloom_filter; mod file_index_format; -// Keep the predicate foundation crate-private until a concrete reader validates its contract. #[allow(dead_code)] pub(crate) mod file_index_predicate; #[allow(dead_code)]