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
// SPDX-License-Identifier: BUSL-1.1
//! Executor handler for `QueryOp::ProviderScan`.
//!
//! Decodes the pre-materialized msgpack row array, applies predicate filtering,
//! offset, sort, distinct deduplication, column projection, and limit — in that
//! order — then emits the resulting rows via `response_with_payload`.
use nodedb_query::msgpack_scan;
use crate::bridge::envelope::{ErrorCode, Response};
use crate::bridge::scan_filter::ScanFilter;
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::handlers::sort_utils::sort_msgpack_rows;
use crate::data::executor::msgpack_utils::write_str;
use crate::data::executor::response_codec::encode_binary_rows;
use crate::data::executor::task::ExecutionTask;
/// Parameters for [`CoreLoop::execute_provider_scan`].
pub(in crate::data::executor) struct ProviderScanParams<'a> {
pub rows_bytes: &'a [u8],
pub filters_bytes: &'a [u8],
pub projection: &'a [String],
pub sort_keys: &'a [(String, bool)],
pub limit: Option<usize>,
pub offset: usize,
pub distinct: bool,
}
impl CoreLoop {
/// Execute a `ProviderScan` plan node.
///
/// Processing order: decode rows → filter → offset → sort → distinct →
/// project → limit → emit.
pub(in crate::data::executor) fn execute_provider_scan(
&mut self,
task: &ExecutionTask,
params: ProviderScanParams<'_>,
) -> Response {
let ProviderScanParams {
rows_bytes,
filters_bytes,
projection,
sort_keys,
limit,
offset,
distinct,
} = params;
// ── 1. Decode the flat msgpack row array. ────────────────────────────
// ProviderScan rows are flat column maps *by contract*: catalog and
// constant results are produced flat, and a gathered storage side is
// flattened at the Exchange-resolution boundary (`exchange::resolve`)
// before it is embedded here. The relational-operator layer therefore
// sees exactly one row shape — this handler never sniffs or unwraps a
// `{id, data}` storage wrapper.
let mut rows = decode_flat_row_array(rows_bytes);
// ── 2. Filter. ────────────────────────────────────────────────────────
if !filters_bytes.is_empty() {
let predicates: Vec<ScanFilter> = match zerompk::from_msgpack(filters_bytes) {
Ok(f) => f,
Err(e) => {
return self.response_error(
task,
ErrorCode::Internal {
detail: format!("ProviderScan: malformed filter bytes: {e}"),
},
);
}
};
if !predicates.is_empty() {
rows.retain(|row| predicates.iter().all(|f| f.matches_binary(row)));
}
}
// ── 3. Offset. ────────────────────────────────────────────────────────
if offset > 0 {
if offset >= rows.len() {
rows.clear();
} else {
rows.drain(..offset);
}
}
// ── 4. Sort. ──────────────────────────────────────────────────────────
if !sort_keys.is_empty() {
sort_msgpack_rows(&mut rows, sort_keys);
}
// ── 5. Distinct (on the would-be projected row). ──────────────────────
// Deduplicate on the projected shape so SQL DISTINCT semantics are
// honoured: two rows with the same projected columns but different
// non-projected columns are considered equal.
if distinct {
let mut seen: std::collections::HashSet<Vec<u8>> = std::collections::HashSet::new();
rows.retain(|row| {
let key = if projection.is_empty() {
row.clone()
} else {
project_row_by_names(row, projection)
};
seen.insert(key)
});
}
// ── 6. Project. ───────────────────────────────────────────────────────
let rows: Vec<Vec<u8>> = if projection.is_empty() {
rows
} else {
rows.into_iter()
.map(|row| project_row_by_names(&row, projection))
.collect()
};
// ── 7. Limit. ─────────────────────────────────────────────────────────
let rows = if let Some(n) = limit {
rows.into_iter().take(n).collect()
} else {
rows
};
// ── 8. Emit. ──────────────────────────────────────────────────────────
let payload = encode_binary_rows(&rows);
self.response_with_payload(task, payload)
}
}
/// Decode a flat msgpack row array (the `encode_binary_rows` format) into
/// individual row byte vectors. Each element is a flat column map; its bytes are
/// returned verbatim so subsequent steps operate per-row without re-encoding.
/// Storage `{id, data}` wrappers are NOT handled here — they are flattened at
/// the Exchange-resolution boundary before reaching a relational operator.
fn decode_flat_row_array(bytes: &[u8]) -> Vec<Vec<u8>> {
if bytes.is_empty() {
return Vec::new();
}
let Some((count, mut pos)) = msgpack_scan::array_header(bytes, 0) else {
return Vec::new();
};
let mut rows = Vec::with_capacity(count);
for _ in 0..count {
let start = pos;
let Some(end) = msgpack_scan::skip_value(bytes, pos) else {
break;
};
rows.push(bytes[start..end].to_vec());
pos = end;
}
rows
}
/// Project a single msgpack map row, keeping only the named columns.
///
/// Returns a new msgpack map containing only the fields whose name appears in
/// `projection`, preserving the original msgpack value bytes verbatim (no
/// decode). If a projection column is not present in the row it is silently
/// omitted. If `projection` is empty this function must not be called (the
/// caller is expected to skip projection for the empty case).
fn project_row_by_names(row: &[u8], projection: &[String]) -> Vec<u8> {
let Some((count, mut pos)) = msgpack_scan::map_header(row, 0) else {
return row.to_vec();
};
let mut entries: Vec<(&str, usize, usize)> = Vec::with_capacity(projection.len());
// We need string slices into the `row` bytes. Use `msgpack_scan::read_str`
// which returns an `Option<&str>` backed by the input slice.
for _ in 0..count {
let key: Option<&str> = msgpack_scan::read_str(row, pos);
let key_end = match msgpack_scan::skip_value(row, pos) {
Some(p) => p,
None => break,
};
let val_start = key_end;
let val_end = match msgpack_scan::skip_value(row, val_start) {
Some(p) => p,
None => break,
};
if let Some(k) = key
&& projection.iter().any(|p| p == k)
{
entries.push((k, val_start, val_end));
}
pos = val_end;
}
let mut buf = Vec::with_capacity(row.len());
// Write map header.
let n = entries.len();
if n < 16 {
buf.push(0x80 | n as u8);
} else if n <= u16::MAX as usize {
buf.push(0xDE);
buf.extend_from_slice(&(n as u16).to_be_bytes());
} else {
buf.push(0xDF);
buf.extend_from_slice(&(n as u32).to_be_bytes());
}
for (key, vs, ve) in &entries {
write_str(&mut buf, key);
buf.extend_from_slice(&row[*vs..*ve]);
}
buf
}