feat(store): add key_prefix_from cursor to Query for efficient prefix scans

src/store.rs

@@ -171,6 +171,7 @@ pub struct QueryBuilder<K> {
     offset: u64,
     include_empty: bool,
     sort_direction: SortDirection,
+    from: Option<Bytes>,
 }
 
 impl<K> QueryBuilder<K> {
@@ -189,6 +190,22 @@ impl<K> QueryBuilder<K> {
         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)). 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]>,
+        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 +275,7 @@ impl From<QueryBuilder<SingleLatestPerKeyQuery>> for Query {
             offset: builder.offset,
             include_empty: builder.include_empty,
             sort_direction: builder.sort_direction,
+            from: builder.from,
         }
     }
 }
@@ -272,6 +290,7 @@ impl From<QueryBuilder<FlatQuery>> for Query {
             offset: builder.offset,
             include_empty: builder.include_empty,
             sort_direction: builder.sort_direction,
+            from: builder.from,
         }
     }
 }
@@ -287,6 +306,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<Bytes>,
 }
 
 impl Query {

src/store/fs/bounds.rs

@@ -103,7 +103,7 @@ impl From<(Bound<RecordsIdOwned>, Bound<RecordsIdOwned>)> for RecordsBounds {
 /// Supports bounds by key.
 pub struct ByKeyBounds(Bound<RecordsByKeyIdOwned>, Bound<RecordsByKeyIdOwned>);
 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,12 @@ 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);
+                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();
                 let mut key_end = prefix.to_vec();
@@ -245,14 +250,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 +267,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 +277,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 +287,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]))
@@ -292,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"
+        );
+    }
 }

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);

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<bytes::Bytes>,
     },
 }
 
@@ -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(),
             },
         }
     }