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.storage.config.read();
78        let is_metadata_only = config.metadata_only;
79        drop(config);
80
81        // LOCK ORDER: vector_storage(2) before payload_storage(3) — was
82        // reversed here, an ABBA risk with Collection::search's canonical
83        // order. See .investigation/http-deadlock-2026-07-22/.
84        let vector_storage = self.storage.vector_storage.read();
85        let payload_storage = self.storage.payload_storage.read();
86        let now_secs = now_unix_secs();
87
88        let mut points = Vec::with_capacity(batch_size);
89        for &id in candidate_ids {
90            if points.len() >= batch_size {
91                break;
92            }
93            if let Some(point) =
94                Self::build_point(id, is_metadata_only, &*payload_storage, &*vector_storage)
95            {
96                // TTL-expired points are invisible; the scan keeps going until
97                // `batch_size` live points are collected (or ids run out).
98                if is_payload_expired(point.payload.as_ref(), now_secs) {
99                    continue;
100                }
101                if Self::passes_filter(&point, filter) {
102                    points.push(point);
103                }
104            }
105        }
106        points
107    }
108
109    /// Builds a `Point` from storage. Always returns `Some`; points without a
110    /// stored vector get an empty vector slice.
111    #[allow(clippy::unnecessary_wraps)] // Reason: Option return used by caller's if-let pattern
112    fn build_point(
113        id: u64,
114        is_metadata_only: bool,
115        payload_storage: &dyn PayloadStorage,
116        vector_storage: &dyn VectorStorage,
117    ) -> Option<Point> {
118        let payload = payload_storage.retrieve(id).ok().flatten();
119        // Graph nodes inserted via upsert_node_payload() have no vector in storage.
120        // Use unwrap_or_default() so payload-only nodes are included, not silently skipped.
121        let vector = if is_metadata_only {
122            Vec::new()
123        } else {
124            vector_storage
125                .retrieve(id)
126                .ok()
127                .flatten()
128                .unwrap_or_default()
129        };
130        Some(Point {
131            id,
132            vector,
133            payload,
134            sparse_vectors: None,
135        })
136    }
137
138    /// Returns `true` if the point passes the optional filter.
139    fn passes_filter(point: &Point, filter: Option<&Filter>) -> bool {
140        match (filter, &point.payload) {
141            (Some(f), Some(payload)) => f.matches(payload),
142            (Some(_), None) => false,
143            (None, _) => true,
144        }
145    }
146}
velesdb_core/collection/core/scroll.rs

velesdb_core/collection/core/
scroll.rs
