Method scan

nodedb/src/engine/kv/engine.rs:292–381

SCAN: cursor-based iteration with optional key pattern matching and index-accelerated predicate pushdown. If `filter_field` and `filter_value` are provided AND a secondary index exists for that field, the scan uses the index to narrow candidates (O(log n) + O(k) where k = matching keys) instead of full table scan. Returns `(entries, next_cursor_bytes)`. `next_cursor_bytes` is empty when the scan

(&self, params: KvScanParams<'_>)

    /// when the scan is complete. Each entry is `(key, value)`.
    /// `params.surrogate_ceiling` enforces clone snapshot isolation when set.
    pub fn scan(&self, params: KvScanParams<'_>) -> ScanResult {
        let KvScanParams {
            tenant_id,
            collection,
            cursor,
            count,
            now_ms,
            match_pattern,
            filter_field,
            filter_value,
            surrogate_ceiling,
        } = params;
        let tkey = table_key(tenant_id, collection);
        let table = match self.tables.get(&tkey) {
            Some(t) => t,
            None => return (Vec::new(), Vec::new()),
        };

        let surrogate_visible = |s: u32| -> bool {
            match surrogate_ceiling {
                Some(c) => s == 0 || s <= c,
                None => true,
            }
        };

        // Index-accelerated path: if we have an equality filter and an index, use it.
        // Also checks composite indexes for prefix matches.
        if let Some(field) = filter_field
            && let Some(value) = filter_value
            && let Some(idx_set) = self.indexes.get(&tkey)
        {
            // Try single-field index first.
            let candidate_keys = if idx_set.get_index(field).is_some() {
                idx_set.lookup_eq(field, value)
            } else if let Some(ci) = idx_set.find_composite_with_prefix(field) {
                // Composite index prefix match: use leading field.
                ci.lookup_prefix(&[value])
            } else {
                Vec::new() // No index available — will fall through to full scan.
            };

            if !candidate_keys.is_empty() {
                let mut results = Vec::with_capacity(count.min(candidate_keys.len()));

                for pk in candidate_keys {
                    if results.len() >= count {
                        break;
                    }
                    if let Some((val, surrogate)) = table.get_with_surrogate(pk, now_ms)
                        && (match_pattern.is_none()
                            || super::scan::matches_pattern_pub(pk, match_pattern))
                        && surrogate_visible(surrogate.as_u32())
                    {
                        results.push((pk.to_vec(), val.to_vec()));
                    }
                }

                return (results, Vec::new());