From c2365ccf3d949733a63dec89de50fe6ddef12266 Mon Sep 17 00:00:00 2001 From: Roy Alcala Ortiz Date: Mon, 15 Jun 2026 18:00:34 +0000 Subject: [PATCH 1/2] feat(store): add key_prefix_from cursor to Query for prefix scans Adds an optional field to the Query struct that serves as a cursor position for prefix scans. When set together with a KeyFilter::Prefix, the B-tree seek starts at the cursor position instead of the prefix start, enabling efficient iteration over large datasets from a known position. - Add to Query and QueryBuilder - Add builder method - Pass cursor through IndexKind::KeyAuthor to ByKeyBounds - Use cursor as lower bound for B-tree range scan (O(log n)) --- src/store.rs | 20 ++++++++++++++++++++ src/store/fs/bounds.rs | 15 ++++++++------- src/store/fs/query.rs | 3 ++- src/store/util.rs | 4 ++++ 4 files changed, 34 insertions(+), 8 deletions(-) diff --git a/src/store.rs b/src/store.rs index e5946a08..3d08e8c8 100644 --- a/src/store.rs +++ b/src/store.rs @@ -171,6 +171,7 @@ pub struct QueryBuilder { offset: u64, include_empty: bool, sort_direction: SortDirection, + from: Option, } impl QueryBuilder { @@ -189,6 +190,19 @@ impl QueryBuilder { self.filter_key = KeyFilter::Prefix(key.as_ref().to_vec().into()); self } + /// Filter by key prefix, starting from a cursor position. + /// + /// Only keys at or after `from` will be returned. The seek is performed + /// at the B-tree level (O(log n)). + pub fn key_prefix_from( + mut self, + prefix: impl AsRef<[u8]>, + from: impl AsRef<[u8]>, + ) -> Self { + self.filter_key = KeyFilter::Prefix(prefix.as_ref().to_vec().into()); + self.from = Some(from.as_ref().to_vec().into()); + self + } /// Filter by author. pub fn author(mut self, author: AuthorId) -> Self { self.filter_author = AuthorFilter::Exact(author); @@ -258,6 +272,7 @@ impl From> for Query { offset: builder.offset, include_empty: builder.include_empty, sort_direction: builder.sort_direction, + from: builder.from, } } } @@ -272,6 +287,7 @@ impl From> for Query { offset: builder.offset, include_empty: builder.include_empty, sort_direction: builder.sort_direction, + from: builder.from, } } } @@ -287,6 +303,10 @@ pub struct Query { offset: u64, include_empty: bool, sort_direction: SortDirection, + /// When set together with a `KeyFilter::Prefix`, the cursor position + /// from which the prefix scan starts. Keys strictly before this + /// position are skipped at the B-tree level (O(log n) seek). + pub(crate) from: Option, } impl Query { diff --git a/src/store/fs/bounds.rs b/src/store/fs/bounds.rs index f8234512..536965b4 100644 --- a/src/store/fs/bounds.rs +++ b/src/store/fs/bounds.rs @@ -103,7 +103,7 @@ impl From<(Bound, Bound)> for RecordsBounds { /// Supports bounds by key. pub struct ByKeyBounds(Bound, Bound); impl ByKeyBounds { - pub fn new(ns: NamespaceId, matcher: &KeyFilter) -> Self { + pub fn new(ns: NamespaceId, matcher: &KeyFilter, from: Option<&[u8]>) -> Self { match matcher { KeyFilter::Any => Self::namespace(ns), KeyFilter::Exact(key) => { @@ -112,7 +112,8 @@ impl ByKeyBounds { Self(Bound::Included(start), Bound::Included(end)) } KeyFilter::Prefix(ref prefix) => { - let start = Bound::Included((ns.to_bytes(), prefix.clone(), [0u8; 32])); + let start_key = from.unwrap_or(prefix); + let start = Bound::Included((ns.to_bytes(), Bytes::copy_from_slice(start_key), [0u8; 32])); let mut ns_end = ns.to_bytes(); let mut key_end = prefix.to_vec(); @@ -245,14 +246,14 @@ mod tests { ); assert_eq!(bounds.end_bound(), Bound::Unbounded); - let bounds = ByKeyBounds::new(ns, &KeyFilter::Any); + let bounds = ByKeyBounds::new(ns, &KeyFilter::Any, None); assert_eq!( bounds.start_bound(), Bound::Included(&(ns.to_bytes(), Bytes::new(), [0u8; 32])) ); assert_eq!(bounds.end_bound(), Bound::Unbounded); - let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![1u8].into())); + let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![1u8].into()), None); assert_eq!( bounds.start_bound(), Bound::Included(&(ns.to_bytes(), vec![1u8].into(), [0u8; 32])) @@ -262,7 +263,7 @@ mod tests { Bound::Excluded(&(ns.to_bytes(), vec![2u8].into(), [0u8; 32])) ); - let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![255u8].into())); + let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![255u8].into()), None); assert_eq!( bounds.start_bound(), Bound::Included(&(ns.to_bytes(), vec![255u8].into(), [0u8; 32])) @@ -272,7 +273,7 @@ mod tests { let ns = NamespaceId::from(&[2u8; 32]); let mut ns_end = ns.to_bytes(); ns_end[31] = 3u8; - let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![255u8].into())); + let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![255u8].into()), None); assert_eq!( bounds.start_bound(), Bound::Included(&(ns.to_bytes(), vec![255u8].into(), [0u8; 32])) @@ -282,7 +283,7 @@ mod tests { Bound::Excluded(&(ns_end, Bytes::new(), [0u8; 32])) ); - let bounds = ByKeyBounds::new(ns, &KeyFilter::Exact(vec![1u8].into())); + let bounds = ByKeyBounds::new(ns, &KeyFilter::Exact(vec![1u8].into()), None); assert_eq!( bounds.start_bound(), Bound::Included(&(ns.to_bytes(), vec![1u8].into(), [0u8; 32])) diff --git a/src/store/fs/query.rs b/src/store/fs/query.rs index 9393afd9..0d78d2d1 100644 --- a/src/store/fs/query.rs +++ b/src/store/fs/query.rs @@ -63,8 +63,9 @@ impl QueryIterator { range, author_filter, latest_per_key, + from, } => { - let bounds = ByKeyBounds::new(namespace, &range); + let bounds = ByKeyBounds::new(namespace, &range, from.as_deref()); let range = RecordsByKeyRange::with_bounds(tables.records_by_key, tables.records, bounds)?; let selector = latest_per_key.then(LatestPerKeySelector::default); diff --git a/src/store/util.rs b/src/store/util.rs index 07e7bdcd..42cd3612 100644 --- a/src/store/util.rs +++ b/src/store/util.rs @@ -14,6 +14,8 @@ pub enum IndexKind { range: KeyFilter, author_filter: AuthorFilter, latest_per_key: bool, + /// When set, the lower bound for the prefix scan. + from: Option, }, } @@ -25,6 +27,7 @@ impl From<&Query> for IndexKind { range: query.filter_key.clone(), author_filter: AuthorFilter::Any, latest_per_key: false, + from: query.from.clone(), }, _ => IndexKind::AuthorKey { range: query.filter_author.clone(), @@ -35,6 +38,7 @@ impl From<&Query> for IndexKind { range: query.filter_key.clone(), author_filter: query.filter_author.clone(), latest_per_key: true, + from: query.from.clone(), }, } } From e1790f90fc1a96219585fe98d1acd5b5b415534e Mon Sep 17 00:00:00 2001 From: Roy Alcala Ortiz Date: Mon, 15 Jun 2026 18:10:10 +0000 Subject: [PATCH 2/2] chore: add test and debug_assert for key_prefix_from cursor - Add unit test for ByKeyBounds with cursor lower bound - Add debug_assert that cursor starts with prefix - Document SortBy::KeyAuthor requirement in doc comment - Document that from is ignored on AuthorKey index path --- src/store.rs | 5 ++++- src/store/fs/bounds.rs | 24 ++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/store.rs b/src/store.rs index 3d08e8c8..0c5f0f4b 100644 --- a/src/store.rs +++ b/src/store.rs @@ -193,7 +193,10 @@ impl QueryBuilder { /// Filter by key prefix, starting from a cursor position. /// /// Only keys at or after `from` will be returned. The seek is performed - /// at the B-tree level (O(log n)). + /// at the B-tree level (O(log n)). Requires `SortBy::KeyAuthor` ordering + /// to use the by-key index, otherwise `from` is ignored. + /// + /// The `from` value must start with the given prefix for correct results. pub fn key_prefix_from( mut self, prefix: impl AsRef<[u8]>, diff --git a/src/store/fs/bounds.rs b/src/store/fs/bounds.rs index 536965b4..d703732d 100644 --- a/src/store/fs/bounds.rs +++ b/src/store/fs/bounds.rs @@ -113,6 +113,10 @@ impl ByKeyBounds { } KeyFilter::Prefix(ref prefix) => { let start_key = from.unwrap_or(prefix); + debug_assert!( + start_key.starts_with(prefix.as_ref()), + "cursor `from` must start with the given prefix" + ); let start = Bound::Included((ns.to_bytes(), Bytes::copy_from_slice(start_key), [0u8; 32])); let mut ns_end = ns.to_bytes(); @@ -293,4 +297,24 @@ mod tests { Bound::Included(&(ns.to_bytes(), vec![1u8].into(), [255u8; 32])) ); } + + #[test] + fn by_key_bounds_prefix_from_cursor() { + let ns = NamespaceId::from(&[1u8; 32]); + + let bounds = ByKeyBounds::new(ns, &KeyFilter::Prefix(vec![b'e', b'v', b't', b':'].into()), Some(b"evt:01JABC...")); + assert_eq!( + bounds.start_bound(), + Bound::Included(&(ns.to_bytes(), Bytes::from_static(b"evt:01JABC..."), [0u8; 32])), + "lower bound should be the cursor, not the prefix" + ); + + let mut key_end = b"evt:".to_vec(); + increment_by_one(&mut key_end); + assert_eq!( + bounds.end_bound(), + Bound::Excluded(&(ns.to_bytes(), Bytes::from(key_end), [0u8; 32])), + "upper bound should still be prefix+1" + ); + } }