nodedb_cluster/distributed_array/wire.rs
1// SPDX-License-Identifier: BUSL-1.1
2
3//! Wire types for array shard RPC messages.
4//!
5//! All request/response pairs are zerompk-serialisable. Complex sub-payloads
6//! (slice predicates, cell batches) are carried as opaque msgpack bytes —
7//! the same convention used by `ArrayOp` in the bridge physical plan — so
8//! this crate does not need a compile-time dependency on `nodedb-array`.
9
10/// Scatter request: coordinator asks a shard to execute a coord-range slice.
11///
12/// `slice_msgpack` is a zerompk encoding of `nodedb_array::query::Slice`.
13/// `attr_projection` is the attribute index list; empty means all attributes.
14#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
15pub struct ArrayShardSliceReq {
16 /// Array scoped by tenant + name (zerompk encoding of `nodedb_array::types::ArrayId`).
17 pub array_id_msgpack: Vec<u8>,
18 /// Zerompk encoding of `nodedb_array::query::Slice`.
19 pub slice_msgpack: Vec<u8>,
20 pub attr_projection: Vec<u32>,
21 pub limit: u32,
22 /// Optional surrogate pre-filter bitmap bytes (zerompk encoding of `SurrogateBitmap`).
23 /// Empty slice means no filter.
24 pub cell_filter_msgpack: Vec<u8>,
25 /// Prefix bits used for Hilbert routing (1–16). 0 means no routing validation.
26 ///
27 /// When non-zero the shard verifies that its `local_vshard_id` is covered
28 /// by the Hilbert ranges in `slice_hilbert_ranges` before executing the
29 /// scan. A mismatch returns `ClusterError::WrongOwner` so the coordinator
30 /// can retry against a refreshed routing table.
31 pub prefix_bits: u8,
32 /// Inclusive Hilbert-prefix ranges `(lo, hi)` that this slice covers,
33 /// pre-computed by the coordinator from the slice predicate. Each entry
34 /// is encoded as two consecutive `u64` values (little-endian). Empty
35 /// means unbounded (no routing validation is performed).
36 pub slice_hilbert_ranges: Vec<(u64, u64)>,
37 /// Hilbert-prefix range `[lo, hi]` that this shard owns. The shard
38 /// applies this as a pre-filter so it only returns cells whose Hilbert
39 /// prefix falls within the range. `None` means unbounded (return all
40 /// matching cells). Used by the distributed shard handler to prevent
41 /// duplicate rows in single-node harnesses where all vShards share one
42 /// Data Plane.
43 pub shard_hilbert_range: Option<(u64, u64)>,
44 /// Bitemporal system-time scope forwarded from `ArrayOp::Slice::system_time`.
45 ///
46 /// - `Current` = live read.
47 /// - `AsOf(t)` = point-in-time at `t`.
48 /// - `AllVersions` = audit-log fan-out — shard emits all live cell-versions,
49 /// each tagged with `ArrayCell::system_time`.
50 pub system_time: nodedb_types::SystemTimeScope,
51 /// Bitemporal valid-time point forwarded from `ArrayOp::Slice::valid_at_ms`.
52 /// `None` = no valid-time filter.
53 pub valid_at_ms: Option<i64>,
54}
55
56/// Gather response: shard returns matching rows as opaque msgpack row bytes.
57#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
58pub struct ArrayShardSliceResp {
59 pub shard_id: u32,
60 /// Zerompk-encoded rows. Each element is a zerompk encoding of one result row.
61 pub rows_msgpack: Vec<Vec<u8>>,
62 /// True when the shard hit the `limit` and may have more rows.
63 pub truncated: bool,
64 /// True when `system_as_of` is below the oldest tile version on this shard
65 /// and the shard produced zero rows as a result of that horizon.
66 pub truncated_before_horizon: bool,
67}
68
69/// Scatter request: coordinator asks a shard to compute a partial aggregate.
70///
71/// Mirrors `ArrayOp::Aggregate` — attribute index + reducer + optional filter.
72#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
73pub struct ArrayShardAggReq {
74 pub array_id_msgpack: Vec<u8>,
75 pub attr_idx: u32,
76 /// Zerompk encoding of `ArrayReducer` (c_enum, 1 byte on wire).
77 pub reducer_msgpack: Vec<u8>,
78 /// Negative means no group-by; non-negative is the dimension index.
79 pub group_by_dim: i32,
80 /// Optional surrogate pre-filter; empty = all cells.
81 pub cell_filter_msgpack: Vec<u8>,
82 /// Hilbert-prefix range `[lo, hi]` that this shard owns. The shard
83 /// applies this as a pre-filter so it only counts cells whose Hilbert
84 /// prefix falls within the range. `None` means unbounded (scan all).
85 pub shard_hilbert_range: Option<(u64, u64)>,
86 /// Bitemporal system-time cutoff forwarded from `ArrayOp::Aggregate::system_as_of`.
87 /// `None` = live read.
88 pub system_as_of: Option<i64>,
89 /// Bitemporal valid-time point forwarded from `ArrayOp::Aggregate::valid_at_ms`.
90 /// `None` = no valid-time filter.
91 pub valid_at_ms: Option<i64>,
92}
93
94/// Gather response: shard returns partial aggregate(s) for merge.
95#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
96pub struct ArrayShardAggResp {
97 pub shard_id: u32,
98 /// One partial per group-by bucket (or one entry when group_by_dim < 0).
99 pub partials: Vec<super::merge::ArrayAggPartial>,
100 /// True when `system_as_of` is below the oldest tile version on this shard
101 /// and the shard produced zero rows as a result of that horizon.
102 pub truncated_before_horizon: bool,
103}
104
105/// Scatter request: coordinator forwards a cell write to the owning shard.
106///
107/// `cells_msgpack` is a zerompk encoding of
108/// `Vec<nodedb::engine::array::wal::ArrayPutCell>`.
109/// All cells in this batch belong to the same Hilbert-prefix bucket.
110#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
111pub struct ArrayShardPutReq {
112 pub array_id_msgpack: Vec<u8>,
113 pub cells_msgpack: Vec<u8>,
114 pub wal_lsn: u64,
115 /// Hilbert prefix of any representative cell in this batch (used by
116 /// the shard handler to validate routing). Set to 0 when routing
117 /// metadata is unavailable (pre-routing clients).
118 pub representative_hilbert_prefix: u64,
119 /// Prefix bits used for routing (1–16). 0 means no validation.
120 pub prefix_bits: u8,
121}
122
123/// Acknowledgement: shard confirms the put was applied.
124#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
125pub struct ArrayShardPutResp {
126 pub shard_id: u32,
127 pub applied_lsn: u64,
128}
129
130/// Scatter request: coordinator asks a shard to delete cells by exact coords.
131///
132/// `coords_msgpack` is a zerompk encoding of `Vec<Vec<CoordValue>>`.
133#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
134pub struct ArrayShardDeleteReq {
135 pub array_id_msgpack: Vec<u8>,
136 pub coords_msgpack: Vec<u8>,
137 pub wal_lsn: u64,
138 /// Hilbert prefix of any representative coord in this batch. Used by the
139 /// shard handler to validate that it still owns the target Hilbert bucket.
140 /// Set to 0 when routing metadata is unavailable.
141 pub representative_hilbert_prefix: u64,
142 /// Prefix bits used for Hilbert routing (1–16). 0 means no validation.
143 pub prefix_bits: u8,
144}
145
146/// Acknowledgement: shard confirms the delete was applied.
147#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
148pub struct ArrayShardDeleteResp {
149 pub shard_id: u32,
150 pub applied_lsn: u64,
151}
152
153/// Scatter request: coordinator asks a shard to run a surrogate bitmap scan.
154///
155/// Used by the cross-engine fusion path to collect surrogates for cells
156/// matching a coord-range slice.
157#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
158pub struct ArrayShardSurrogateBitmapReq {
159 pub array_id_msgpack: Vec<u8>,
160 /// Zerompk encoding of `nodedb_array::query::Slice`.
161 pub slice_msgpack: Vec<u8>,
162}
163
164/// Response: shard returns a surrogate bitmap for matching cells.
165#[derive(Debug, Clone, zerompk::ToMessagePack, zerompk::FromMessagePack)]
166pub struct ArrayShardSurrogateBitmapResp {
167 pub shard_id: u32,
168 /// Zerompk encoding of `SurrogateBitmap` for the matching cells.
169 pub bitmap_msgpack: Vec<u8>,
170}