Skip to content

Commit a37f92c

Browse files
authored
Logdb: support for prefix bloom filters (opendata-oss#486)
Use a bloom filter based on the log key prefix.
1 parent 27a97de commit a37f92c

10 files changed

Lines changed: 1069 additions & 10 deletions

File tree

common/src/serde/terminated_bytes.rs

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,41 @@ pub fn deserialize(buf: &mut &[u8]) -> Result<Bytes, DeserializeError> {
121121
})
122122
}
123123

124+
/// Returns the position just past the terminator that closes a
125+
/// terminated-bytes-encoded segment within `bytes`, starting the scan at
126+
/// `start`.
127+
///
128+
/// Walks the encoded stream byte-by-byte, treating `ESCAPE_BYTE` (`0x01`)
129+
/// as an escape that consumes the following byte regardless of value, and
130+
/// stopping at the first `TERMINATOR_BYTE` (`0x00`) outside an escape
131+
/// sequence. Returns `None` if the input ends before the terminator is
132+
/// reached, including the case where the last byte is the start of an
133+
/// escape sequence (truncation-unsafe — the next byte could shift the
134+
/// terminator's position).
135+
///
136+
/// Unlike [`deserialize`], this function does not decode the payload — it
137+
/// only locates the boundary. Callers that need just the end offset (e.g.
138+
/// a prefix extractor sizing a hashable prefix) can use this without
139+
/// allocating.
140+
pub fn find_terminator_end(bytes: &[u8], start: usize) -> Option<usize> {
141+
let mut i = start;
142+
while i < bytes.len() {
143+
match bytes[i] {
144+
TERMINATOR_BYTE => return Some(i + 1),
145+
ESCAPE_BYTE => {
146+
// Escape byte consumes the next byte. If the stream ends
147+
// mid-escape we cannot decide where the terminator lands.
148+
i += 2;
149+
if i > bytes.len() {
150+
return None;
151+
}
152+
}
153+
_ => i += 1,
154+
}
155+
}
156+
None
157+
}
158+
124159
/// Creates a [`BytesRange`] for scanning all keys with the given logical prefix.
125160
///
126161
/// The prefix is first incremented using `lex_increment`, then both bounds
@@ -260,6 +295,83 @@ mod tests {
260295
assert_eq!(slice, &[0xDE, 0xAD]);
261296
}
262297

298+
#[test]
299+
fn should_find_terminator_end_on_simple_input() {
300+
// given — serialize "abc" so we know exactly where the terminator lands
301+
let mut buf = BytesMut::new();
302+
serialize(b"abc", &mut buf);
303+
// expected layout: 'a' 'b' 'c' 0x00 (4 bytes)
304+
305+
// when / then — the terminator end sits one byte past the terminator
306+
assert_eq!(find_terminator_end(&buf, 0), Some(4));
307+
}
308+
309+
#[test]
310+
fn should_find_terminator_end_skipping_escaped_zero_byte() {
311+
// given — payload contains 0x00 which encodes to 0x01 0x01, then "x",
312+
// then the real terminator. The inner 0x01 must NOT be mistaken for
313+
// the terminator.
314+
let mut buf = BytesMut::new();
315+
serialize(&[0x00, b'x'], &mut buf);
316+
// layout: 0x01 0x01 'x' 0x00 (4 bytes)
317+
318+
// when / then
319+
assert_eq!(find_terminator_end(&buf, 0), Some(4));
320+
}
321+
322+
#[test]
323+
fn should_find_terminator_end_skipping_escaped_escape_byte() {
324+
// given — payload contains 0x01 which encodes to 0x01 0x02. The 0x02
325+
// looks innocuous but the escape pair must be consumed atomically so
326+
// the position is correctly tracked.
327+
let mut buf = BytesMut::new();
328+
serialize(&[0x01, b'x'], &mut buf);
329+
// layout: 0x01 0x02 'x' 0x00 (4 bytes)
330+
331+
// when / then
332+
assert_eq!(find_terminator_end(&buf, 0), Some(4));
333+
}
334+
335+
#[test]
336+
fn should_find_terminator_end_from_offset_start() {
337+
// given — a synthetic header followed by an encoded payload. Searching
338+
// from offset 3 skips past the header bytes (which may legally contain
339+
// 0x00).
340+
let mut buf = BytesMut::from(&[0x00, 0xFF, 0x42][..]);
341+
serialize(b"key", &mut buf);
342+
// layout: [00 FF 42] [k e y 00]; payload terminator at index 6, end at 7
343+
344+
// when / then
345+
assert_eq!(find_terminator_end(&buf, 3), Some(7));
346+
}
347+
348+
#[test]
349+
fn should_return_none_when_terminator_absent() {
350+
// given — bytes without a terminator
351+
let bytes: &[u8] = b"abc";
352+
353+
// when / then
354+
assert_eq!(find_terminator_end(bytes, 0), None);
355+
}
356+
357+
#[test]
358+
fn should_return_none_when_input_ends_mid_escape() {
359+
// given — last byte is the escape byte itself
360+
let bytes: &[u8] = &[b'a', ESCAPE_BYTE];
361+
362+
// when / then
363+
assert_eq!(find_terminator_end(bytes, 0), None);
364+
}
365+
366+
#[test]
367+
fn should_return_none_when_start_is_past_end() {
368+
// given
369+
let bytes: &[u8] = &[0x00];
370+
371+
// when / then
372+
assert_eq!(find_terminator_end(bytes, 5), None);
373+
}
374+
263375
#[test]
264376
fn should_not_have_encoded_prefix_collision() {
265377
// Encoded "a" should not be a prefix of encoded "ab"

common/src/storage/factory.rs

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ pub use slatedb::db_cache::foyer_hybrid::FoyerHybridCache;
1717
pub use slatedb::db_cache::{CachedEntry, CachedKey, SplitCache};
1818
use slatedb::object_store::{self, ObjectStore};
1919
pub use slatedb::{CompactorBuilder, DbBuilder};
20-
use slatedb::{DbReader, SstReader};
20+
use slatedb::{DbReader, FilterPolicy, SstReader};
2121
use tracing::info;
2222
use uuid::Uuid;
2323

@@ -158,6 +158,9 @@ impl StorageBuilder {
158158
let adapter = SlateDbStorage::merge_operator_adapter(op);
159159
db_builder = db_builder.with_merge_operator(Arc::new(adapter));
160160
}
161+
if let Some(policies) = self.semantics.filter_policies {
162+
db_builder = db_builder.with_filter_policies(policies);
163+
}
161164
let db = db_builder.build().await.map_err(|e| {
162165
StorageError::Storage(format!("Failed to create SlateDB: {}", e))
163166
})?;
@@ -246,6 +249,7 @@ impl StorageReaderRuntime {
246249
#[derive(Default)]
247250
pub struct StorageSemantics {
248251
pub(crate) merge_operator: Option<Arc<dyn MergeOperator>>,
252+
pub(crate) filter_policies: Option<Vec<Arc<dyn FilterPolicy>>>,
249253
}
250254

251255
impl StorageSemantics {
@@ -262,6 +266,15 @@ impl StorageSemantics {
262266
self.merge_operator = Some(op);
263267
self
264268
}
269+
270+
/// Sets the filter policies for SlateDB writers and readers.
271+
///
272+
/// System crates use this to keep the writer, compactor, and standalone
273+
/// reader paths aligned on SST filter encoding/decoding behavior.
274+
pub fn with_filter_policies(mut self, policies: Vec<Arc<dyn FilterPolicy>>) -> Self {
275+
self.filter_policies = Some(policies);
276+
self
277+
}
265278
}
266279

267280
pub fn new_slatedb_compactor_builder(
@@ -387,6 +400,9 @@ pub async fn create_storage_read(
387400
let adapter = SlateDbStorage::merge_operator_adapter(op);
388401
builder = builder.with_merge_operator(Arc::new(adapter));
389402
}
403+
if let Some(policies) = semantics.filter_policies {
404+
builder = builder.with_filter_policies(policies);
405+
}
390406
if let Some(cache) = cache.clone() {
391407
builder = builder.with_db_cache(cache);
392408
}

common/src/storage/mod.rs

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -242,6 +242,24 @@ pub trait StorageRead: Any + Send + Sync {
242242
range: BytesRange,
243243
) -> StorageResult<Box<dyn StorageIterator + Send + 'static>>;
244244

245+
/// Returns an iterator over records whose key starts with `prefix`.
246+
///
247+
/// Backends that support prefix-aware bloom filters (e.g. SlateDB with a
248+
/// configured `PrefixExtractor`) consult those filters here, allowing
249+
/// SSTs that contain no matching keys to be skipped without a block
250+
/// read. Backends without such filters fall back to a range scan over
251+
/// the prefix.
252+
///
253+
/// The default implementation delegates to [`Self::scan_iter`] with
254+
/// the range derived from the prefix, which preserves correctness for
255+
/// any backend; backends that can do better should override this.
256+
async fn scan_prefix_iter(
257+
&self,
258+
prefix: Bytes,
259+
) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
260+
self.scan_iter(BytesRange::prefix(prefix)).await
261+
}
262+
245263
/// Collects all records in the range into a Vec.
246264
#[tracing::instrument(level = "trace", skip_all)]
247265
async fn scan(&self, range: BytesRange) -> StorageResult<Vec<Record>> {

common/src/storage/slate.rs

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -234,6 +234,22 @@ impl StorageRead for SlateDbStorage {
234234
Ok(Box::new(SlateDbIterator { iter }))
235235
}
236236

237+
/// Slatedb consults its SST-level filters on `scan_prefix` but not on
238+
/// `scan`, so routing prefix scans through this path is what lets a
239+
/// configured `PrefixExtractor` actually skip SSTs.
240+
#[tracing::instrument(level = "trace", skip_all)]
241+
async fn scan_prefix_iter(
242+
&self,
243+
prefix: Bytes,
244+
) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
245+
let iter = self
246+
.db
247+
.scan_prefix_with_options(prefix, &default_scan_options())
248+
.await
249+
.map_err(StorageError::from_storage)?;
250+
Ok(Box::new(SlateDbIterator { iter }))
251+
}
252+
237253
fn slate_read(&self) -> Option<SlateReadHandle> {
238254
self.sst_reader.as_ref().map(|sst_reader| SlateReadHandle {
239255
// Live writer manifest: reflects everything flushed so far.
@@ -302,6 +318,19 @@ impl StorageRead for SlateDbStorageSnapshot {
302318
.map_err(StorageError::from_storage)?;
303319
Ok(Box::new(SlateDbIterator { iter }))
304320
}
321+
322+
#[tracing::instrument(level = "trace", skip_all)]
323+
async fn scan_prefix_iter(
324+
&self,
325+
prefix: Bytes,
326+
) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
327+
let iter = self
328+
.snapshot
329+
.scan_prefix_with_options(prefix, &default_scan_options())
330+
.await
331+
.map_err(StorageError::from_storage)?;
332+
Ok(Box::new(SlateDbIterator { iter }))
333+
}
305334
}
306335

307336
#[async_trait]
@@ -511,6 +540,19 @@ impl StorageRead for SlateDbStorageReader {
511540
Ok(Box::new(SlateDbIterator { iter }))
512541
}
513542

543+
#[tracing::instrument(level = "trace", skip_all)]
544+
async fn scan_prefix_iter(
545+
&self,
546+
prefix: Bytes,
547+
) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
548+
let iter = self
549+
.reader
550+
.scan_prefix_with_options(prefix, &default_scan_options())
551+
.await
552+
.map_err(StorageError::from_storage)?;
553+
Ok(Box::new(SlateDbIterator { iter }))
554+
}
555+
514556
fn slate_read(&self) -> Option<SlateReadHandle> {
515557
self.sst_reader.as_ref().map(|sst_reader| SlateReadHandle {
516558
// The DbReader's last-polled manifest. Scans on this same handle

0 commit comments

Comments
 (0)