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_hilbert_ranges` — pre-computed `(lo, hi)` Hilbert-prefix
34    /// ranges that determine which shards to contact. An empty vec means
35    /// "scan all shards".
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 Hilbert-prefix ranges for shard selection.
47        slice_hilbert_ranges: Vec<(u64, 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 Hilbert ranges for shard selection (empty = all).
72        slice_hilbert_ranges: Vec<(u64, 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 Hilbert tile.
82    ///
83    /// `cells` — list of `(hilbert_prefix, zerompk-encoded single-cell
84    /// bytes)`. The coordinator partitions these by prefix and sends each
85    /// 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 `(hilbert_prefix, 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 Hilbert tile.
99    ///
100    /// `coords` — `(hilbert_prefix, 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 `(hilbert_prefix, 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}
nodedb_physical/physical_plan/cluster_array.rs

nodedb_physical/physical_plan/
cluster_array.rs
