refactor(spatial): replace raw geometry bytes with typed Geometry throughout

The query geometry was previously carried as Vec<u8> / &[u8] across
the entire spatial pipeline and parsed in the Data Plane handler,
meaning malformed input was only caught at execution time with no
validation feedback at the SQL or native protocol layer.

Parse and validate the geometry once on the Control Plane — at the
SQL planner (where_search) and native plan builder — then propagate
the typed nodedb_types::geometry::Geometry value through SqlPlan,
PhysicalPlan, and the distributed scatter coordinator. The Data Plane
handler receives a pre-validated value and no longer needs to parse.

Geometry validation (via nodedb_spatial::validate::validate_geometry)
now runs at query entry points and returns a structured SQL error
rather than an Internal error from the executor.

Add integration test spatial_cp_dp_query covering the full CP→DP
round trip for a spatial scan with attribute filters.

Author: farhan-syah

Cargo.lock

@@ -12,8 +12,8 @@ pub struct SpatialScatterPayload {
     pub collection: String,
     pub field: String,
     pub predicate: String,
-    /// Raw query geometry bytes (GeoJSON, passed through as-is).
-    pub query_geometry: Vec<u8>,
+    /// Typed query geometry, parsed and validated on the originating CP.
+    pub query_geometry: nodedb_types::geometry::Geometry,
     pub distance_meters: f64,
     pub limit: u32,
 }
@@ -44,15 +44,15 @@ impl SpatialScatterGather {
         collection: &str,
         field: &str,
         predicate: &str,
-        query_geometry_json: &[u8],
+        query_geometry: nodedb_types::geometry::Geometry,
         distance_meters: f64,
         limit: usize,
     ) -> Vec<(u32, VShardEnvelope)> {
         let msg = SpatialScatterPayload {
             collection: collection.to_string(),
             field: field.to_string(),
             predicate: predicate.to_string(),
-            query_geometry: query_geometry_json.to_vec(),
+            query_geometry,
             distance_meters,
             limit: limit as u32,
         };
@@ -106,11 +106,14 @@ mod tests {
     #[test]
     fn scatter_envelopes_built() {
         let coord = SpatialScatterGather::new(1, vec![0, 1, 2]);
-        let query =
-            serde_json::to_vec(&serde_json::json!({"type": "Point", "coordinates": [0.0, 0.0]}))
-                .unwrap();
-        let envs =
-            coord.build_scatter_envelopes("buildings", "geom", "st_dwithin", &query, 1000.0, 100);
+        let envs = coord.build_scatter_envelopes(
+            "buildings",
+            "geom",
+            "st_dwithin",
+            nodedb_types::geometry::Geometry::point(0.0, 0.0),
+            1000.0,
+            100,
+        );
         assert_eq!(envs.len(), 3);
         for (shard_id, env) in &envs {
             assert_eq!(env.msg_type, VShardMessageType::SpatialScatterRequest);

nodedb-cluster/src/distributed_spatial/coordinator.rs

@@ -9,6 +9,7 @@ description = "SQL parser, planner, and optimizer for NodeDB"
 [dependencies]
 nodedb-types = { workspace = true }
 nodedb-query = { workspace = true }
+nodedb-spatial = { workspace = true }
 sqlparser = { version = "0.61", features = ["visitor"] }
 thiserror = { workspace = true }
 chrono = { workspace = true }

nodedb-sql/Cargo.toml

@@ -97,6 +97,16 @@ fn plan_spatial_from_where(
         context: name.into(),
     })?;
     let geom_str = extract_geometry_arg(geom_arg)?;
+    let geometry: nodedb_types::geometry::Geometry =
+        sonic_rs::from_str(&geom_str).map_err(|e| SqlError::InvalidFunction {
+            detail: format!("invalid geometry in {name}: {e}"),
+        })?;
+    let issues = nodedb_spatial::validate::validate_geometry(&geometry);
+    if !issues.is_empty() {
+        return Err(SqlError::InvalidFunction {
+            detail: format!("invalid geometry in {name}: {}", issues.join("; ")),
+        });
+    }
     let distance = if args.len() >= 3 {
         extract_float(&args[2]).unwrap_or(0.0)
     } else {
@@ -106,7 +116,7 @@ fn plan_spatial_from_where(
         collection: table.name.clone(),
         field,
         predicate,
-        query_geometry: geom_str.into_bytes(),
+        query_geometry: geometry,
         distance_meters: distance,
         attribute_filters: Vec::new(),
         limit: 1000,

nodedb-sql/src/planner/select/where_search.rs

@@ -361,7 +361,7 @@ pub enum SqlPlan {
         collection: String,
         field: String,
         predicate: SpatialPredicate,
-        query_geometry: Vec<u8>,
+        query_geometry: nodedb_types::geometry::Geometry,
         distance_meters: f64,
         attribute_filters: Vec<Filter>,
         limit: usize,

nodedb-sql/src/types/plan.rs

@@ -1,6 +1,6 @@
 //! Spatial engine operations dispatched to the Data Plane.
 
-use nodedb_types::SurrogateBitmap;
+use nodedb_types::{SurrogateBitmap, geometry::Geometry};
 
 /// Spatial predicate type for R-tree index scan.
 #[derive(
@@ -42,8 +42,8 @@ pub enum SpatialOp {
         collection: String,
         field: String,
         predicate: SpatialPredicate,
-        /// Query geometry serialized as GeoJSON bytes.
-        query_geometry: Vec<u8>,
+        /// Typed query geometry, parsed and validated on the Control Plane.
+        query_geometry: Geometry,
         /// Distance threshold in meters (for ST_DWithin). 0 for non-distance predicates.
         distance_meters: f64,
         /// Additional attribute filters applied after spatial candidates.

nodedb/src/bridge/physical_plan/spatial.rs

@@ -208,7 +208,7 @@ mod tests {
             collection: "places".into(),
             field: "location".into(),
             predicate: SpatialPredicate::DWithin,
-            query_geometry: b"{}".to_vec(),
+            query_geometry: nodedb_types::geometry::Geometry::point(0.0, 0.0),
             distance_meters: 500.0,
             attribute_filters: vec![],
             limit: 20,

nodedb/src/bridge/physical_plan/wire.rs

@@ -733,7 +733,7 @@ pub(super) fn convert_spatial_scan(p: SpatialScanParams<'_>) -> crate::Result<Ve
             collection: collection.into(),
             field: field.to_string(),
             predicate: sp,
-            query_geometry: query_geometry.to_vec(),
+            query_geometry: query_geometry.clone(),
             distance_meters: *distance_meters,
             attribute_filters: attr_bytes,
             limit: *limit,

nodedb/src/control/planner/sql_plan_convert/scan.rs

@@ -105,7 +105,7 @@ pub(super) struct SpatialScanParams<'a> {
     pub collection: &'a str,
     pub field: &'a str,
     pub predicate: &'a nodedb_sql::types::SpatialPredicate,
-    pub query_geometry: &'a [u8],
+    pub query_geometry: &'a nodedb_types::geometry::Geometry,
     pub distance_meters: &'a f64,
     pub attribute_filters: &'a [Filter],
     pub limit: &'a usize,

nodedb/src/control/planner/sql_plan_convert/scan_params.rs

@@ -6,13 +6,24 @@ use crate::bridge::envelope::PhysicalPlan;
 use crate::bridge::physical_plan::{SpatialOp, SpatialPredicate};
 
 pub(crate) fn build_scan(fields: &TextFields, collection: &str) -> crate::Result<PhysicalPlan> {
-    let query_geometry = fields
+    let raw_bytes = fields
         .query_geometry
         .as_ref()
         .ok_or_else(|| crate::Error::BadRequest {
             detail: "missing 'query_geometry'".to_string(),
-        })?
-        .clone();
+        })?;
+
+    let geometry: nodedb_types::geometry::Geometry =
+        sonic_rs::from_slice(raw_bytes).map_err(|e| crate::Error::BadRequest {
+            detail: format!("invalid query geometry: {e}"),
+        })?;
+
+    let issues = nodedb_spatial::validate::validate_geometry(&geometry);
+    if !issues.is_empty() {
+        return Err(crate::Error::BadRequest {
+            detail: format!("invalid query geometry: {}", issues.join("; ")),
+        });
+    }
 
     let predicate_str = fields.spatial_predicate.as_deref().unwrap_or("dwithin");
     let predicate = match predicate_str.to_lowercase().as_str() {
@@ -35,7 +46,7 @@ pub(crate) fn build_scan(fields: &TextFields, collection: &str) -> crate::Result
         collection: collection.to_string(),
         field,
         predicate,
-        query_geometry,
+        query_geometry: geometry,
         distance_meters,
         attribute_filters: Vec::new(),
         limit,