nodedb_physical/physical_plan/
wire.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Wire-format encode/decode helpers for PhysicalPlan.
4//!
5//! MessagePack encoding via zerompk. Used by the cluster layer to ship
6//! physical plans over the wire as part of `ExecuteRequest` RPC.
7
8use super::{PhysicalPlan, QueryOp};
9
10/// Errors produced by the wire encode/decode helpers. Self-contained so this
11/// module can move into the shared `nodedb-physical` crate without dragging
12/// Origin's `Error` type along.
13#[derive(Debug, thiserror::Error)]
14pub enum WireError {
15    #[error("{0}")]
16    InvalidPlan(&'static str),
17    #[error("plan codec: {0}")]
18    Codec(String),
19}
20
21/// Encode a `PhysicalPlan` to MessagePack bytes.
22///
23/// Returns an error for plans that are node-local by construction and must
24/// never be shipped over the QUIC wire:
25/// - `ClusterArray` variants (handled on the Control Plane); and
26/// - `QueryOp::ShuffleJoinConsume` and `QueryOp::ShuffleAggregateConsume`
27///   (carry node-local staged-file paths; built locally by the part-owner's
28///   consume hook and dispatched only to that same node's Data Plane).
29///
30/// `QueryOp::PartialAggregateState` is wire-SHIPPABLE — it carries a collection
31/// name plus an optional `input` sub-plan (a wire-shippable `ProviderScan`) and
32/// is dispatched to a remote producer's Data Plane — so it is intentionally NOT
33/// rejected here.
34pub fn encode(plan: &PhysicalPlan) -> Result<Vec<u8>, WireError> {
35    if matches!(plan, PhysicalPlan::ClusterArray(_)) {
36        return Err(WireError::InvalidPlan(
37            "ClusterArray plans must not be sent over the wire",
38        ));
39    }
40    if matches!(
41        plan,
42        PhysicalPlan::Query(QueryOp::ShuffleJoinConsume { .. })
43    ) {
44        return Err(WireError::InvalidPlan(
45            "ShuffleJoinConsume plans carry node-local paths and must not be sent over the wire",
46        ));
47    }
48    if matches!(
49        plan,
50        PhysicalPlan::Query(QueryOp::ShuffleAggregateConsume { .. })
51    ) {
52        return Err(WireError::InvalidPlan(
53            "ShuffleAggregateConsume plans carry node-local paths and must not be sent over the wire",
54        ));
55    }
56    zerompk::to_msgpack_vec(plan).map_err(|e| WireError::Codec(format!("encode: {e}")))
57}
58
59/// Decode a `PhysicalPlan` from MessagePack bytes.
60pub fn decode(bytes: &[u8]) -> Result<PhysicalPlan, WireError> {
61    zerompk::from_msgpack(bytes).map_err(|e| WireError::Codec(format!("decode: {e}")))
62}
63
64/// Encode a `Vec<PhysicalPlan>` to MessagePack bytes.
65///
66/// Used by the Calvin scheduler when building `TxClass::plans` bytes for a
67/// cross-shard transaction that will be shipped through the sequencer.
68pub fn encode_batch(plans: &Vec<PhysicalPlan>) -> Result<Vec<u8>, WireError> {
69    for plan in plans {
70        if matches!(plan, PhysicalPlan::ClusterArray(_)) {
71            return Err(WireError::InvalidPlan(
72                "ClusterArray plans must not be shipped via the sequencer",
73            ));
74        }
75        if matches!(
76            plan,
77            PhysicalPlan::Query(QueryOp::ShuffleJoinConsume { .. })
78        ) {
79            return Err(WireError::InvalidPlan(
80                "ShuffleJoinConsume plans carry node-local paths and must not be shipped via the sequencer",
81            ));
82        }
83        if matches!(
84            plan,
85            PhysicalPlan::Query(QueryOp::ShuffleAggregateConsume { .. })
86        ) {
87            return Err(WireError::InvalidPlan(
88                "ShuffleAggregateConsume plans carry node-local paths and must not be shipped via the sequencer",
89            ));
90        }
91    }
92    zerompk::to_msgpack_vec(plans).map_err(|e| WireError::Codec(format!("batch encode: {e}")))
93}
94
95/// Decode a `Vec<PhysicalPlan>` from MessagePack bytes.
96///
97/// Used by the Calvin scheduler to decode the opaque `TxClass::plans` blob
98/// into executable plans for dispatch via `MetaOp::CalvinExecute`.
99pub fn decode_batch(bytes: &[u8]) -> Result<Vec<PhysicalPlan>, WireError> {
100    zerompk::from_msgpack(bytes).map_err(|e| WireError::Codec(format!("batch decode: {e}")))
101}
102
103#[cfg(test)]
104mod tests {
105    use nodedb_types::{DatabaseId, QualifiedCollection};
106
107    use super::*;
108    use crate::physical_plan::JoinProjection;
109
110    #[test]
111    fn hash_join_tail_semantics_roundtrip() {
112        let plan = PhysicalPlan::Query(QueryOp::HashJoin {
113            left_collection: QualifiedCollection::new(DatabaseId::DEFAULT, "left"),
114            right_collection: QualifiedCollection::new(DatabaseId::DEFAULT, "right"),
115            left_alias: Some("l".into()),
116            right_alias: Some("r".into()),
117            on: vec![("id".into(), "id".into())],
118            join_type: "left".into(),
119            limit: 10,
120            projection: vec![JoinProjection {
121                source: "l.id".into(),
122                output: "id".into(),
123            }],
124            computed_projection: vec![1, 2],
125            join_filters: vec![3, 4],
126            post_filters: vec![5, 6],
127            left_input: None,
128            right_input: None,
129            left_rls_filters: Vec::new(),
130            right_rls_filters: Vec::new(),
131            left_scan_filters: Vec::new(),
132            right_scan_filters: Vec::new(),
133            left_bitmap: None,
134            right_bitmap: None,
135        });
136
137        let encoded = encode(&plan).expect("hash join encodes");
138        let decoded = decode(&encoded).expect("hash join decodes");
139        assert_eq!(decoded, plan);
140    }
141}
nodedb_physical/physical_plan/wire.rs

nodedb_physical/physical_plan/
wire.rs
