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
// SPDX-License-Identifier: Apache-2.0
//! Control-Plane cluster array operations.
//!
//! These variants are emitted by the SQL converter when cluster mode is
//! active and an array operation spans multiple vShards. They are handled
//! exclusively by the Control-Plane dispatch loop (pgwire routing handler)
//! which calls the `ArrayCoordinator` before anything touches the SPSC
//! bridge. They are **never sent to the Data Plane**.
//!
//! The converter emits local `ArrayOp` variants in single-node mode and
//! `ClusterArrayOp` variants in cluster mode. Downstream code that sees
//! a `PhysicalPlan::ClusterArray(_)` knows it must call the coordinator,
//! not the SPSC bridge.
use nodedb_array::types::ArrayId;
use nodedb_types::SystemTimeScope;
/// Cluster-mode array operations executed by the coordinator on the
/// Control Plane.
#[derive(
Debug,
Clone,
PartialEq,
serde::Serialize,
serde::Deserialize,
zerompk::ToMessagePack,
zerompk::FromMessagePack,
)]
pub enum ClusterArrayOp {
/// Fan-out a slice query across all target vShards and merge rows.
///
/// `slice_tile_keys` — pre-computed tile keys, at least one per routing
/// bucket the slice touches. They pick the shards to contact. An empty
/// vec scans every shard.
///
/// `prefix_bits` — routing granularity from the array catalog entry.
Slice {
array_id: ArrayId,
/// zerompk-encoded `Slice` predicate.
slice_msgpack: Vec<u8>,
/// Projected attribute indices (empty = all).
attr_projection: Vec<u32>,
/// Maximum rows returned. 0 = unlimited.
limit: u32,
/// Pre-computed tile keys for shard selection.
slice_tile_keys: Vec<u64>,
/// Routing granularity from the array catalog entry.
prefix_bits: u8,
/// Bitemporal system-time scope for this slice fan-out.
///
/// - `Current`: live read.
/// - `AsOf(t)`: point-in-time at `t`.
/// - `AllVersions`: audit-log fan-out — each shard emits all live
/// cell-versions; the coordinator merge-sorts by
/// `ArrayCell::system_time` ascending before applying the limit.
system_time: SystemTimeScope,
/// Bitemporal valid-time point. `None` = no valid-time filter.
valid_at_ms: Option<i64>,
},
/// Fan-out an aggregate across all target vShards and reduce partials.
Agg {
array_id: ArrayId,
/// Index of the attribute to aggregate.
attr_idx: u32,
/// zerompk-encoded `ArrayReducer`.
reducer_msgpack: Vec<u8>,
/// Dimension index to group by (-1 = no grouping).
group_by_dim: i32,
/// Pre-computed tile keys for shard selection (empty = all).
slice_tile_keys: Vec<u64>,
/// Routing granularity.
prefix_bits: u8,
/// Bitemporal system-time cutoff. `None` = live read.
system_as_of: Option<i64>,
/// Bitemporal valid-time point. `None` = no valid-time filter.
valid_at_ms: Option<i64>,
},
/// Fan-out a cell write batch, partitioned by routing bucket.
///
/// `cells` — list of `(tile_key, zerompk-encoded single-cell bytes)`.
/// The coordinator partitions these by the tile key's vShard and sends
/// each shard's subset via `coord_put`.
Put {
array_id: ArrayId,
/// zerompk-encoded `ArrayId` bytes for the wire request.
array_id_msgpack: Vec<u8>,
/// Cells as `(tile_key, cell_msgpack)` pairs.
cells: Vec<(u64, Vec<u8>)>,
/// WAL LSN allocated by the Control Plane for this batch.
wal_lsn: u64,
/// Routing granularity.
prefix_bits: u8,
},
/// Fan-out a coord delete batch, partitioned by routing bucket.
///
/// `coords` — `(tile_key, zerompk-encoded single-coord bytes)`.
Delete {
array_id: ArrayId,
/// zerompk-encoded `ArrayId` bytes for the wire request.
array_id_msgpack: Vec<u8>,
/// Coords as `(tile_key, coord_msgpack)` pairs.
coords: Vec<(u64, Vec<u8>)>,
/// WAL LSN.
wal_lsn: u64,
/// Routing granularity.
prefix_bits: u8,
},
}
impl ClusterArrayOp {
/// The array this op targets. Total over every variant, so read-set
/// tracking and the commit validator key on the same name a shard-local
/// `ArrayOp` reports.
pub fn array_id(&self) -> &ArrayId {
match self {
ClusterArrayOp::Slice { array_id, .. }
| ClusterArrayOp::Agg { array_id, .. }
| ClusterArrayOp::Put { array_id, .. }
| ClusterArrayOp::Delete { array_id, .. } => array_id,
}
}
}