nodedb_physical/physical_plan/
cluster_array.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Control-Plane cluster array operations.
4//!
5//! These variants are emitted by the SQL converter when cluster mode is
6//! active and an array operation spans multiple vShards. They are handled
7//! exclusively by the Control-Plane dispatch loop (pgwire routing handler)
8//! which calls the `ArrayCoordinator` before anything touches the SPSC
9//! bridge. They are **never sent to the Data Plane**.
10//!
11//! The converter emits local `ArrayOp` variants in single-node mode and
12//! `ClusterArrayOp` variants in cluster mode. Downstream code that sees
13//! a `PhysicalPlan::ClusterArray(_)` knows it must call the coordinator,
14//! not the SPSC bridge.
15
16use nodedb_array::types::ArrayId;
17use nodedb_types::SystemTimeScope;
18
19/// Cluster-mode array operations executed by the coordinator on the
20/// Control Plane.
21#[derive(
22    Debug,
23    Clone,
24    PartialEq,
25    serde::Serialize,
26    serde::Deserialize,
27    zerompk::ToMessagePack,
28    zerompk::FromMessagePack,
29)]
30pub enum ClusterArrayOp {
31    /// Fan-out a slice query across all target vShards and merge rows.
32    ///
33    /// `slice_tile_keys` — pre-computed tile keys, at least one per routing
34    /// bucket the slice touches. They pick the shards to contact. An empty
35    /// vec scans every shard.
36    ///
37    /// `prefix_bits` — routing granularity from the array catalog entry.
38    Slice {
39        array_id: ArrayId,
40        /// zerompk-encoded `Slice` predicate.
41        slice_msgpack: Vec<u8>,
42        /// Projected attribute indices (empty = all).
43        attr_projection: Vec<u32>,
44        /// Maximum rows returned. 0 = unlimited.
45        limit: u32,
46        /// Pre-computed tile keys for shard selection.
47        slice_tile_keys: Vec<u64>,
48        /// Routing granularity from the array catalog entry.
49        prefix_bits: u8,
50        /// Bitemporal system-time scope for this slice fan-out.
51        ///
52        /// - `Current`: live read.
53        /// - `AsOf(t)`: point-in-time at `t`.
54        /// - `AllVersions`: audit-log fan-out — each shard emits all live
55        ///   cell-versions; the coordinator merge-sorts by
56        ///   `ArrayCell::system_time` ascending before applying the limit.
57        system_time: SystemTimeScope,
58        /// Bitemporal valid-time point. `None` = no valid-time filter.
59        valid_at_ms: Option<i64>,
60    },
61
62    /// Fan-out an aggregate across all target vShards and reduce partials.
63    Agg {
64        array_id: ArrayId,
65        /// Index of the attribute to aggregate.
66        attr_idx: u32,
67        /// zerompk-encoded `ArrayReducer`.
68        reducer_msgpack: Vec<u8>,
69        /// Dimension index to group by (-1 = no grouping).
70        group_by_dim: i32,
71        /// Pre-computed tile keys for shard selection (empty = all).
72        slice_tile_keys: Vec<u64>,
73        /// Routing granularity.
74        prefix_bits: u8,
75        /// Bitemporal system-time cutoff. `None` = live read.
76        system_as_of: Option<i64>,
77        /// Bitemporal valid-time point. `None` = no valid-time filter.
78        valid_at_ms: Option<i64>,
79    },
80
81    /// Fan-out a cell write batch, partitioned by routing bucket.
82    ///
83    /// `cells` — list of `(tile_key, zerompk-encoded single-cell bytes)`.
84    /// The coordinator partitions these by the tile key's vShard and sends
85    /// each shard's subset via `coord_put`.
86    Put {
87        array_id: ArrayId,
88        /// zerompk-encoded `ArrayId` bytes for the wire request.
89        array_id_msgpack: Vec<u8>,
90        /// Cells as `(tile_key, cell_msgpack)` pairs.
91        cells: Vec<(u64, Vec<u8>)>,
92        /// WAL LSN allocated by the Control Plane for this batch.
93        wal_lsn: u64,
94        /// Routing granularity.
95        prefix_bits: u8,
96    },
97
98    /// Fan-out a coord delete batch, partitioned by routing bucket.
99    ///
100    /// `coords` — `(tile_key, zerompk-encoded single-coord bytes)`.
101    Delete {
102        array_id: ArrayId,
103        /// zerompk-encoded `ArrayId` bytes for the wire request.
104        array_id_msgpack: Vec<u8>,
105        /// Coords as `(tile_key, coord_msgpack)` pairs.
106        coords: Vec<(u64, Vec<u8>)>,
107        /// WAL LSN.
108        wal_lsn: u64,
109        /// Routing granularity.
110        prefix_bits: u8,
111    },
112}
113
114impl ClusterArrayOp {
115    /// The array this op targets. Total over every variant, so read-set
116    /// tracking and the commit validator key on the same name a shard-local
117    /// `ArrayOp` reports.
118    pub fn array_id(&self) -> &ArrayId {
119        match self {
120            ClusterArrayOp::Slice { array_id, .. }
121            | ClusterArrayOp::Agg { array_id, .. }
122            | ClusterArrayOp::Put { array_id, .. }
123            | ClusterArrayOp::Delete { array_id, .. } => array_id,
124        }
125    }
126}
nodedb_physical/physical_plan/cluster_array.rs

nodedb_physical/physical_plan/
cluster_array.rs
