velesdb_core/collection/core/
scroll.rs

1//! Scroll cursor for paginated iteration over collection points.
2//!
3//! Provides `ScrollBatch` and `Collection::scroll_batch` for deterministic,
4//! ascending-ID iteration with optional payload filtering.
5
6use crate::collection::expiry::{is_payload_expired, now_unix_secs};
7use crate::collection::types::Collection;
8use crate::error::{Error, Result};
9use crate::filter::Filter;
10use crate::point::Point;
11use crate::storage::{PayloadStorage, VectorStorage};
12
13/// Result of a single scroll batch operation.
14///
15/// Contains the points in this batch (ascending ID order) and the cursor
16/// position for resuming iteration.
17#[derive(Debug, Clone)]
18pub struct ScrollBatch {
19    /// Points in this batch, ordered by ascending ID.
20    pub points: Vec<Point>,
21    /// Cursor for the next batch (`None` if no more points).
22    /// This is the ID of the last point in this batch.
23    pub next_cursor: Option<u64>,
24}
25
26impl Collection {
27    /// Returns the next batch of points starting after `cursor`.
28    ///
29    /// - `cursor`: `None` to start from the beginning, `Some(id)` to resume
30    ///   after the given point ID (exclusive).
31    /// - `batch_size`: Maximum number of points to return. Must be > 0.
32    /// - `filter`: Optional payload filter. Points not matching are skipped.
33    ///
34    /// Points are returned in ascending ID order for deterministic iteration.
35    ///
36    /// # Errors
37    ///
38    /// Returns `Error::Config` if `batch_size` is 0.
39    pub fn scroll_batch(
40        &self,
41        cursor: Option<u64>,
42        batch_size: usize,
43        filter: Option<&Filter>,
44    ) -> Result<ScrollBatch> {
45        if batch_size == 0 {
46            return Err(Error::Config(
47                "batch_size must be greater than 0".to_string(),
48            ));
49        }
50
51        // all_point_ids() returns IDs pre-sorted via BTreeSet (see crud_read_delete.rs).
52        // Binary search via partition_point is O(log N) per batch.
53        let ids = self.all_point_ids();
54
55        let start = match cursor {
56            Some(c) => ids.partition_point(|&id| id <= c),
57            None => 0,
58        };
59
60        let candidates = &ids[start..];
61        let points = self.collect_filtered_batch(candidates, batch_size, filter);
62
63        let next_cursor = points.last().map(|p| p.id);
64        Ok(ScrollBatch {
65            points,
66            next_cursor,
67        })
68    }
69
70    /// Collects up to `batch_size` points from `candidate_ids`, applying an optional filter.
71    fn collect_filtered_batch(
72        &self,
73        candidate_ids: &[u64],
74        batch_size: usize,
75        filter: Option<&Filter>,
76    ) -> Vec<Point> {
77        let config = self.config.read();
78        let is_metadata_only = config.metadata_only;
79        drop(config);
80
81        let payload_storage = self.payload_storage.read();
82        let vector_storage = self.vector_storage.read();
83        let now_secs = now_unix_secs();
84
85        let mut points = Vec::with_capacity(batch_size);
86        for &id in candidate_ids {
87            if points.len() >= batch_size {
88                break;
89            }
90            if let Some(point) =
91                Self::build_point(id, is_metadata_only, &*payload_storage, &*vector_storage)
92            {
93                // TTL-expired points are invisible; the scan keeps going until
94                // `batch_size` live points are collected (or ids run out).
95                if is_payload_expired(point.payload.as_ref(), now_secs) {
96                    continue;
97                }
98                if Self::passes_filter(&point, filter) {
99                    points.push(point);
100                }
101            }
102        }
103        points
104    }
105
106    /// Builds a `Point` from storage. Always returns `Some`; points without a
107    /// stored vector get an empty vector slice.
108    #[allow(clippy::unnecessary_wraps)] // Reason: Option return used by caller's if-let pattern
109    fn build_point(
110        id: u64,
111        is_metadata_only: bool,
112        payload_storage: &dyn PayloadStorage,
113        vector_storage: &dyn VectorStorage,
114    ) -> Option<Point> {
115        let payload = payload_storage.retrieve(id).ok().flatten();
116        // Graph nodes inserted via upsert_node_payload() have no vector in storage.
117        // Use unwrap_or_default() so payload-only nodes are included, not silently skipped.
118        let vector = if is_metadata_only {
119            Vec::new()
120        } else {
121            vector_storage
122                .retrieve(id)
123                .ok()
124                .flatten()
125                .unwrap_or_default()
126        };
127        Some(Point {
128            id,
129            vector,
130            payload,
131            sparse_vectors: None,
132        })
133    }
134
135    /// Returns `true` if the point passes the optional filter.
136    fn passes_filter(point: &Point, filter: Option<&Filter>) -> bool {
137        match (filter, &point.payload) {
138            (Some(f), Some(payload)) => f.matches(payload),
139            (Some(_), None) => false,
140            (None, _) => true,
141        }
142    }
143}
velesdb_core/collection/core/scroll.rs

velesdb_core/collection/core/
scroll.rs
