nodedb_cluster/distributed_array/
local_executor.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Local Data-Plane execution trait for array shard operations.
4//!
5//! `ArrayLocalExecutor` is defined here in `nodedb-cluster` and
6//! implemented in the main `nodedb` binary, which has access to the
7//! SPSC bridge and the Data Plane array engine. The shard-side handler
8//! (`handler.rs`) holds an `Arc<dyn ArrayLocalExecutor>` and calls
9//! through it to execute slice and surrogate-bitmap-scan operations
10//! on the local node.
11//!
12//! This keeps `nodedb-cluster` free of a compile-time dependency on
13//! `nodedb` while still producing real results from the Data Plane.
14
15use crate::distributed_array::merge::ArrayAggPartial;
16use crate::distributed_array::wire::{
17    ArrayShardAggReq, ArrayShardDeleteReq, ArrayShardPutReq, ArrayShardSliceReq,
18};
19use crate::error::Result;
20
21/// Result of a local shard slice: the per-row bytes plus the bitemporal
22/// below-horizon signal.
23///
24/// `truncated_before_horizon` is computed by the Data Plane (it is `true`
25/// when the `system_time` cutoff predates every stored tile version, so the
26/// shard produced zero rows for that reason). It MUST be threaded back to the
27/// coordinator so the OR-reduce across shards can surface an incomplete-result
28/// signal to the client — dropping it would silently report complete results.
29pub struct ArraySliceExec {
30    pub rows: Vec<Vec<u8>>,
31    pub truncated_before_horizon: bool,
32}
33
34/// Result of a local shard partial aggregate: the per-group partial states
35/// plus the bitemporal below-horizon signal (see [`ArraySliceExec`]).
36pub struct ArrayAggExec {
37    pub partials: Vec<ArrayAggPartial>,
38    pub truncated_before_horizon: bool,
39}
40
41/// Execute array operations against the local Data Plane.
42///
43/// The implementor (in `nodedb`) routes the call through the SPSC bridge
44/// to the appropriate TPC core and returns the serialised zerompk result
45/// rows or bitmap bytes.
46#[async_trait::async_trait]
47pub trait ArrayLocalExecutor: Send + Sync + 'static {
48    /// Execute a coord-range slice and return zerompk-encoded row bytes.
49    ///
50    /// `array_id_msgpack` — zerompk encoding of `nodedb_array::types::ArrayId`.
51    /// `slice_msgpack` — zerompk encoding of `nodedb_array::query::Slice`.
52    /// `attr_projection` — attribute index list; empty means all attributes.
53    /// `limit` — maximum rows to return per shard (0 = unlimited).
54    /// `cell_filter_msgpack` — zerompk encoding of `SurrogateBitmap`; empty
55    ///   means no filter.
56    /// `shard_hilbert_range` — optional `[lo, hi]` Hilbert-prefix range; when
57    ///   set only tiles whose prefix falls in this range are returned, preventing
58    ///   duplicate rows in single-node harnesses where all vShards share one
59    ///   Data Plane. `None` = no Hilbert filter.
60    ///
61    /// Returns the per-row bytes (one element per matching row, each the
62    /// native-msgpack encoding of that row) plus the `truncated_before_horizon`
63    /// signal from the Data Plane.
64    async fn exec_slice(&self, req: &ArrayShardSliceReq) -> Result<ArraySliceExec>;
65
66    /// Execute a surrogate-bitmap scan and return the zerompk-encoded
67    /// `SurrogateBitmap` bytes for matching cells.
68    ///
69    /// `array_id_msgpack` — zerompk encoding of `nodedb_array::types::ArrayId`.
70    /// `slice_msgpack` — zerompk encoding of `nodedb_array::query::Slice`.
71    async fn exec_surrogate_bitmap_scan(
72        &self,
73        array_id_msgpack: &[u8],
74        slice_msgpack: &[u8],
75    ) -> Result<Vec<u8>>;
76
77    /// Execute a partial aggregate on this shard and return the partial states.
78    ///
79    /// The Data Plane computes the aggregate with `return_partial = true`, so it
80    /// returns partial states (plus the `truncated_before_horizon` signal)
81    /// rather than finalized scalars. The coordinator merges partials from all
82    /// shards before finalizing.
83    async fn exec_agg(&self, req: &ArrayShardAggReq) -> Result<ArrayAggExec>;
84
85    /// Apply a cell-batch write to the local array engine.
86    ///
87    /// `req.cells_msgpack` is a zerompk encoding of
88    /// `Vec<nodedb::engine::array::wal::ArrayPutCell>`. All cells belong to
89    /// the same Hilbert-prefix tile. The shard handler has already validated
90    /// that this shard owns the tile; the executor dispatches directly to the
91    /// Data Plane without further routing checks.
92    async fn exec_put(&self, req: &ArrayShardPutReq) -> Result<u64>;
93
94    /// Delete cells by exact coordinates from the local array engine.
95    ///
96    /// Takes the full `ArrayShardDeleteReq` (not just the coord bytes) because
97    /// the implementor needs the request's Hilbert routing metadata
98    /// (`representative_hilbert_prefix` / `prefix_bits`) to derive the owning
99    /// vShard when it replicates the delete to that shard's data Raft group,
100    /// mirroring [`Self::exec_put`].
101    ///
102    /// Returns the `applied_lsn` (equal to `req.wal_lsn` on success).
103    async fn exec_delete(&self, req: &ArrayShardDeleteReq) -> Result<u64>;
104}
nodedb_cluster/distributed_array/local_executor.rs

nodedb_cluster/distributed_array/
local_executor.rs
