1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
// SPDX-License-Identifier: BUSL-1.1
//! KV Scan handler and filter extraction.
use tracing::debug;
use crate::bridge::envelope::{ErrorCode, Response};
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::handlers::scan_budget;
use crate::data::executor::task::ExecutionTask;
use crate::engine::kv::KvScanParams;
use crate::engine::kv::current_ms;
/// Parameters for the KV SCAN handler.
pub(in crate::data::executor) struct KvScanHandlerParams<'a> {
pub did: u64,
pub tid: u64,
pub collection: &'a str,
pub cursor: &'a [u8],
pub count: usize,
pub match_pattern: Option<&'a str>,
pub filters: &'a [u8],
pub sort_keys: &'a [(String, bool)],
pub surrogate_ceiling: Option<u32>,
}
impl CoreLoop {
pub(in crate::data::executor) fn execute_kv_scan(
&self,
task: &ExecutionTask,
params: KvScanHandlerParams<'_>,
) -> Response {
let KvScanHandlerParams {
did,
tid,
collection,
cursor,
count,
match_pattern,
filters,
sort_keys,
surrogate_ceiling,
} = params;
debug!(core = self.core_id, %collection, count, "kv scan");
// Scan-quiesce gate: refuse new scans against a draining
// collection so the purge handler can unlink on-disk files
// without racing an in-flight reader.
let _scan_guard = match self.acquire_scan_guard(task, tid, collection) {
Ok(g) => g,
Err(resp) => return resp,
};
let now_ms = current_ms();
// A no-LIMIT SQL `SELECT * FROM <kv>` arrives as `count == usize::MAX`.
// Bound the engine fetch to a row ceiling derived from the per-query
// memory budget (+1 row to detect "more exist") so the materialized
// `Vec` cannot grow to the whole collection. The RESP cursor
// pagination path always carries a finite `count`, so it is unaffected.
let scan_budget_bytes = self.query_tuning.max_scan_result_bytes;
let unbounded = count == usize::MAX;
let fetch_count = if unbounded {
// KV scans have no row offset (pagination is cursor-based).
scan_budget::fetch_limit_for(count, 0, scan_budget_bytes)
} else {
count
};
// Try to extract a single equality filter for index pushdown.
let (filter_field, filter_value) = extract_eq_filter(filters);
let (mut entries, _next_cursor) = self.kv_engine.scan(KvScanParams {
database_id: did,
tenant_id: tid,
collection,
cursor,
count: fetch_count,
now_ms,
match_pattern,
filter_field: filter_field.as_deref(),
filter_value: filter_value.as_deref(),
surrogate_ceiling,
});
// Read-your-own-writes: fold this transaction's staged KV writes
// into the base scan result before the shared filter/sort/encode
// pipeline below, which then treats merged and base rows alike.
// `matches` always accepts here -- the per-entry filter loop further
// down re-applies `filter_predicates` uniformly to every row
// (base or merged), so there is no need to duplicate that check
// in the merge itself.
if let Some(txn_id) = task.request.txn_id {
let coll_key = (
crate::types::DatabaseId::new(did),
crate::types::TenantId::new(tid),
collection.to_string(),
);
self.merge_kv_overlay_into_scan(txn_id, &coll_key, &mut entries, &|_value: &[u8]| true);
}
// Bound an unbounded (no-LIMIT) scan by the memory budget. Sum the raw
// key+value bytes and surface a deterministic error if the result would
// exceed the budget rather than silently truncating it.
if unbounded {
let total = entries.iter().fold(0usize, |acc, (k, v)| {
acc.saturating_add(k.len()).saturating_add(v.len())
});
if scan_budget::budget_exceeded(total, scan_budget_bytes) {
return self.response_error(task, ErrorCode::ResourcesExhausted);
}
}
// Parse filter predicates for post-scan evaluation.
// Index pushdown handles eq filters on indexed fields, but general
// predicates (gt, lt, in, etc.) need post-scan evaluation.
let filter_predicates: Vec<crate::bridge::scan_filter::ScanFilter> = if !filters.is_empty()
{
zerompk::from_msgpack(filters).unwrap_or_default()
} else {
Vec::new()
};
// Build results as raw msgpack — no serde_json::Value intermediary.
let mut result_entries: Vec<Vec<u8>> = Vec::with_capacity(entries.len());
for (k, v) in &entries {
let key_str = String::from_utf8_lossy(k);
// Two storage shapes coexist by design (see dml.rs::convert_kv_insert):
// - msgpack map (typed columns) — inject `key` in place.
// - raw bytes (single-`value` form / RESP SET) — wrap as
// `{value: <bytes>}` first so the downstream injection
// produces the same `{key, value}` shape every scan path
// expects.
let entry_mp = if nodedb_query::msgpack_scan::map_header(v, 0).is_some() {
nodedb_query::msgpack_scan::inject_str_field(v, "key", &key_str)
} else {
let mut wrapped = Vec::with_capacity(v.len() + 8);
nodedb_query::msgpack_scan::write_map_header(&mut wrapped, 1);
nodedb_query::msgpack_scan::write_str(&mut wrapped, "value");
nodedb_query::msgpack_scan::write_str(&mut wrapped, &String::from_utf8_lossy(v));
nodedb_query::msgpack_scan::inject_str_field(&wrapped, "key", &key_str)
};
// Apply filter predicates post-scan (already works on raw msgpack).
if !filter_predicates.is_empty()
&& !filter_predicates
.iter()
.all(|f| f.matches_binary(&entry_mp))
{
continue;
}
result_entries.push(entry_mp);
}
if !sort_keys.is_empty() {
super::super::sort_utils::sort_msgpack_rows(&mut result_entries, sort_keys);
}
// Build response as flat msgpack array — same format as document/columnar scan.
// RESP SCAN handles cursor pagination at its own handler layer.
let mut payload =
Vec::with_capacity(result_entries.iter().map(|e| e.len()).sum::<usize>() + 64);
nodedb_query::msgpack_scan::write_array_header(&mut payload, result_entries.len());
for entry in &result_entries {
payload.extend_from_slice(entry);
}
if let Some(ref m) = self.metrics {
m.record_kv_scan();
}
self.response_with_payload(task, payload)
}
}
/// Extract a single equality filter from serialized ScanFilter bytes.
///
/// Looks for the first `{"field": "x", "op": "eq", "value": "y"}` filter.
/// Returns `(Some(field), Some(value_bytes))` if found, `(None, None)` otherwise.
pub(in crate::data::executor) fn extract_eq_filter(
filters: &[u8],
) -> (Option<String>, Option<Vec<u8>>) {
if filters.is_empty() {
return (None, None);
}
// Filters are MessagePack-encoded Vec<ScanFilter>.
let Ok(parsed) = zerompk::from_msgpack::<Vec<nodedb_types::json_msgpack::JsonValue>>(filters)
.map(|v| {
v.into_iter()
.map(|jv| jv.0)
.collect::<Vec<serde_json::Value>>()
})
else {
tracing::trace!(
len = filters.len(),
"filter deserialization failed, falling back to full scan"
);
return (None, None);
};
for filter in &parsed {
let Some(field) = filter.get("field").and_then(|v| v.as_str()) else {
continue;
};
let Some(op) = filter.get("op").and_then(|v| v.as_str()) else {
continue;
};
if op != "eq" {
continue;
}
let Some(value) = filter.get("value") else {
continue;
};
let value_bytes = match value {
serde_json::Value::String(s) => s.as_bytes().to_vec(),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
let sortable = (i as u64) ^ (1u64 << 63);
sortable.to_be_bytes().to_vec()
} else {
n.to_string().into_bytes()
}
}
other => other.to_string().into_bytes(),
};
return (Some(field.to_string()), Some(value_bytes));
}
(None, None)
}