security(sql): enforce reconstructed SQL provenance

Author: farhan-syah

.github/workflows/test.yml

@@ -64,6 +64,10 @@ jobs:
         run: bash scripts/ci/check_calvin_determinism.sh
       - name: Plane-separation gate
         run: bash scripts/ci/check_plane_separation.sh
+      - name: Reconstructed-SQL gate
+        run: |
+          python3 scripts/ci/check_reconstructed_sql.py --self-test
+          python3 scripts/ci/check_reconstructed_sql.py
       - name: Install cargo-deny
         uses: taiki-e/install-action@v2
         with:

nodedb-physical/src/physical_plan/cluster_event.rs

@@ -0,0 +1,49 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Event-Plane operations executed by the receiving Control Plane.
+//!
+//! These plans are cluster-RPC envelopes only. They must never cross the
+//! Control Plane → Data Plane bridge.
+
+/// Cluster-routed Event-Plane operation.
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    serde::Serialize,
+    serde::Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
+pub enum ClusterEventOp {
+    /// Consume one CDC partition from the leader node's local event buffer.
+    ConsumeStream {
+        stream_name: String,
+        group_name: String,
+        partition: u32,
+        limit: u64,
+    },
+    /// Publish one durable-topic message on the topic's home node.
+    PublishTopic { topic_name: String, payload: String },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ClusterEventOp;
+    use crate::physical_plan::{PhysicalPlan, wire};
+
+    #[test]
+    fn cluster_event_plan_roundtrips_over_cluster_wire() {
+        let plan = PhysicalPlan::ClusterEvent(ClusterEventOp::ConsumeStream {
+            stream_name: "orders; no SQL".into(),
+            group_name: "Analytics".into(),
+            partition: 7,
+            limit: 128,
+        });
+        let encoded = wire::encode(&plan).expect("encode typed cluster event");
+        assert_eq!(
+            wire::decode(&encoded).expect("decode typed cluster event"),
+            plan
+        );
+    }
+}

nodedb-physical/src/physical_plan/mod.rs

@@ -8,6 +8,7 @@
 
 pub mod array;
 pub mod cluster_array;
+pub mod cluster_event;
 pub mod columnar;
 pub mod crdt;
 pub mod document;
@@ -26,6 +27,7 @@ pub mod wire;
 
 pub use array::{ArrayBinaryOp, ArrayOp, ArrayReducer};
 pub use cluster_array::ClusterArrayOp;
+pub use cluster_event::ClusterEventOp;
 pub use columnar::{ColumnarInsertIntent, ColumnarOp};
 pub use crdt::CrdtOp;
 pub use document::{
@@ -89,6 +91,9 @@ pub enum PhysicalPlan {
     /// Cluster-mode array operations executed by the coordinator on the
     /// Control Plane. Never sent to the Data Plane.
     ClusterArray(ClusterArrayOp),
+    /// Event-Plane operations executed by a receiving Control Plane.
+    /// Never sent to the Data Plane.
+    ClusterEvent(ClusterEventOp),
 }
 
 impl PhysicalPlan {
@@ -200,6 +205,7 @@ impl PhysicalPlan {
             | PhysicalPlan::Meta(_)
             | PhysicalPlan::Array(_)
             | PhysicalPlan::ClusterArray(_)
+            | PhysicalPlan::ClusterEvent(_)
             | PhysicalPlan::Query(_) => false,
         }
     }
@@ -336,7 +342,8 @@ impl PhysicalPlan {
             | PhysicalPlan::Query(_)
             | PhysicalPlan::Meta(_)
             | PhysicalPlan::Array(_)
-            | PhysicalPlan::ClusterArray(_) => None,
+            | PhysicalPlan::ClusterArray(_)
+            | PhysicalPlan::ClusterEvent(_) => None,
         }
     }
 

nodedb-physical/src/physical_plan/routing.rs

@@ -117,6 +117,7 @@ pub fn plan_contains_cluster_partitioned_leaf(plan: &PhysicalPlan) -> bool {
         | PhysicalPlan::Spatial(_)
         | PhysicalPlan::Crdt(_)
         | PhysicalPlan::Meta(_)
+        | PhysicalPlan::ClusterEvent(_)
         | PhysicalPlan::Query(QueryOp::ProviderScan { .. })
         | PhysicalPlan::Query(QueryOp::PartialAggregate { .. })
         | PhysicalPlan::Query(QueryOp::PartialAggregateState { .. })

nodedb-sql/src/ddl_ast/parse/collection/column_list.rs

@@ -64,9 +64,12 @@ pub(super) fn extract_column_pairs(body: &str) -> Result<Vec<(String, String)>,
 /// Heuristic: does the first token in the paren body look like a WITH-clause
 /// key rather than a column name+type?
 fn is_with_clause_inner(upper_inner: &str) -> bool {
-    let first_tok = upper_inner.split_whitespace().next().unwrap_or("");
+    let first_tok = upper_inner
+        .split(|character: char| character.is_whitespace() || character == '=')
+        .next()
+        .unwrap_or("");
     matches!(
-        first_tok.trim_end_matches(['=', '\'']),
+        first_tok,
         "ENGINE"
             | "PROFILE"
             | "VECTOR_FIELD"
@@ -319,6 +322,15 @@ fn parse_col_token(token: &str) -> Result<Option<(String, String)>, SqlError> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn with_clause_without_spaces_is_not_a_column_list() {
+        assert!(
+            extract_column_pairs("WITH (engine='document_schemaless')")
+                .expect("WITH options")
+                .is_empty()
+        );
+    }
+
     #[test]
     fn generated_clause_after_unicode_type_preserves_original_offsets() {
         let parsed = parse_col_token("slug CUSTOMffff GENERATED ALWAYS AS (lower(name))")

nodedb/src/control/clone/resolver/rewrite.rs

@@ -289,7 +289,8 @@ pub fn rewrite_plan_for_source(
         | PhysicalPlan::Query(_)
         | PhysicalPlan::Meta(_)
         | PhysicalPlan::Array(_)
-        | PhysicalPlan::ClusterArray(_) => Ok(None),
+        | PhysicalPlan::ClusterArray(_)
+        | PhysicalPlan::ClusterEvent(_) => Ok(None),
     }
 }
 
@@ -322,7 +323,8 @@ pub(super) fn extract_collection_from_plan(plan: &PhysicalPlan) -> Option<&str>
         | PhysicalPlan::Query(_)
         | PhysicalPlan::Meta(_)
         | PhysicalPlan::Array(_)
-        | PhysicalPlan::ClusterArray(_) => None,
+        | PhysicalPlan::ClusterArray(_)
+        | PhysicalPlan::ClusterEvent(_) => None,
     }
 }
 

nodedb/src/control/cluster/calvin/scheduler/driver/core/routing.rs

@@ -77,7 +77,9 @@ pub(crate) fn plan_vshard_in_database(plan: &PhysicalPlan, database_id: Database
         // Cluster-fanned-out array ops are handled entirely by the
         // Control-Plane `ArrayCoordinator` and never dispatched to the Data
         // Plane (see `data/executor/dispatch/visitor.rs`'s `unreachable!`).
-        PhysicalPlan::ClusterArray(_) => PlanRouting::ControlPlaneOnly,
+        PhysicalPlan::ClusterArray(_) | PhysicalPlan::ClusterEvent(_) => {
+            PlanRouting::ControlPlaneOnly
+        }
         // Reads / query operators / metadata ops: `is_write_plan` already
         // excludes every variant of these four families upstream.
         PhysicalPlan::Text(_) => PlanRouting::NotAWrite,