fix(data-plane): vector string parsing, cross join, and graph tenant scoping

strict_format: handle VECTOR columns arriving as strings on the UPDATE
path (sqlparser debug repr, ARRAY[...] SQL literal, and JSON-style).
Extract the float conversion and dimension validation into helpers to
remove duplication.

hash join: add cross-join (cartesian product) support. Previously the
executor had no code path for join_type="cross", causing silent empty
results on any CROSS JOIN query.

graph algo: pass tenant_id into execute_graph_algo() and scope
source_node with the tenant prefix so path/traversal algorithms
resolve correctly against CSR node keys stored under tenant scope.

document read: remove a spurious filter_map-to-Some wrapping that was
masking a unit type mismatch, and simplify to a plain map.

Author: farhan-syah

nodedb/src/data/executor/dispatch/graph.rs

@@ -80,7 +80,9 @@ impl CoreLoop {
                 options.max_visited,
             ),
 
-            GraphOp::Algo { algorithm, params } => self.execute_graph_algo(task, algorithm, params),
+            GraphOp::Algo { algorithm, params } => {
+                self.execute_graph_algo(task, tid, algorithm, params)
+            }
 
             GraphOp::Match { query } => self.execute_graph_match(task, query),
         }

nodedb/src/data/executor/handlers/document/read.rs

@@ -122,15 +122,15 @@ impl CoreLoop {
                 Ok(docs) if docs.is_empty() => {
                     // Sparse empty — try KV / columnar engines.
                     let fallback = self.scan_collection(tid, collection, fetch_limit);
-                    if let Ok(ref docs) = fallback {
-                        if !docs.is_empty() {
-                            warn!(
-                                core = self.core_id,
-                                %collection,
-                                count = docs.len(),
-                                "document scan fallback to scan_collection"
-                            );
-                        }
+                    if let Ok(ref docs) = fallback
+                        && !docs.is_empty()
+                    {
+                        warn!(
+                            core = self.core_id,
+                            %collection,
+                            count = docs.len(),
+                            "document scan fallback to scan_collection"
+                        );
                     }
                     fallback
                 }
@@ -217,16 +217,16 @@ impl CoreLoop {
                         .into_iter()
                         .skip(offset)
                         .take(limit)
-                        .filter_map(|(doc_id, val)| {
+                        .map(|(doc_id, val)| {
                             let data = super::super::super::strict_format::binary_tuple_to_json(
                                 &val, schema,
                             )
                             .unwrap_or(serde_json::Value::Null);
                             let projected = apply_projection(data, &computed_cols, projection);
-                            Some(super::super::super::response_codec::DocumentRow {
+                            super::super::super::response_codec::DocumentRow {
                                 id: doc_id,
                                 data: projected,
-                            })
+                            }
                         })
                         .collect();
                     return self.send_document_rows_transformed(task, &result, stream_chunk_size);

nodedb/src/data/executor/handlers/graph_algo.rs

@@ -8,13 +8,15 @@ use tracing::debug;
 
 use crate::bridge::envelope::{ErrorCode, Response};
 use crate::data::executor::core_loop::CoreLoop;
+use crate::data::executor::scoping::scoped_node;
 use crate::data::executor::task::ExecutionTask;
 use crate::engine::graph::algo::params::{AlgoParams, GraphAlgorithm};
 
 impl CoreLoop {
     pub(in crate::data::executor) fn execute_graph_algo(
         &self,
         task: &ExecutionTask,
+        tid: u32,
         algorithm: &GraphAlgorithm,
         params: &AlgoParams,
     ) -> Response {
@@ -35,6 +37,18 @@ impl CoreLoop {
             );
         }
 
+        // Scope source_node with tenant_id so it matches CSR node keys.
+        // Only clone params when source_node needs to be rewritten.
+        let scoped_params;
+        let params = if params.source_node.is_some() {
+            let mut p = params.clone();
+            p.source_node = p.source_node.map(|n| scoped_node(tid, &n));
+            scoped_params = p;
+            &scoped_params
+        } else {
+            params
+        };
+
         use crate::engine::graph::algo;
 
         let result: Result<Vec<u8>, crate::Error> = match algorithm {

nodedb/src/data/executor/handlers/join/hash.rs

@@ -110,6 +110,26 @@ pub(super) fn probe_hash_index(p: &ProbeParams<'_>) -> Vec<serde_json::Value> {
     let is_right = p.join_type == "right" || p.join_type == "full";
     let is_semi = p.join_type == "semi";
     let is_anti = p.join_type == "anti";
+    let is_cross = p.join_type == "cross";
+
+    // Cross join: cartesian product (no hash lookup needed).
+    if is_cross {
+        let mut results = Vec::new();
+        for (_, left_val) in p.probe_docs {
+            for (_, right_val) in p.index_docs {
+                if results.len() >= p.limit {
+                    return results;
+                }
+                results.push(merge_join_docs_binary(
+                    left_val,
+                    Some(right_val),
+                    p.probe_collection,
+                    p.index_collection,
+                ));
+            }
+        }
+        return results;
+    }
 
     let mut index_matched: std::collections::HashSet<usize> = std::collections::HashSet::new();
     let mut results = Vec::new();

nodedb/src/data/executor/strict_format.rs

@@ -150,22 +150,17 @@ fn coerce_value(val: &Value, col_type: &ColumnType, col_name: &str) -> Result<Va
         ColumnType::Vector(dim) => match val {
             Value::Bytes(b) if b.len() == *dim as usize * 4 => Ok(val.clone()),
             Value::Array(arr) => {
-                let floats: Vec<f32> = arr
-                    .iter()
-                    .filter_map(|v| match v {
-                        Value::Float(f) => Some(*f as f32),
-                        Value::Integer(n) => Some(*n as f32),
-                        _ => None,
-                    })
-                    .collect();
-                if floats.len() != *dim as usize {
-                    return Err(format!(
-                        "column '{col_name}': expected VECTOR({dim}), got {} elements",
-                        floats.len()
-                    ));
+                let floats = extract_vector_floats(arr);
+                validate_and_encode_vector(col_name, *dim, &floats)
+            }
+            Value::String(s) => {
+                // UPDATE path may serialize ARRAY literal as string — parse it.
+                match parse_vector_string(s) {
+                    Some(floats) => validate_and_encode_vector(col_name, *dim, &floats),
+                    None => Err(format!(
+                        "column '{col_name}': expected VECTOR array, got String({s:?})"
+                    )),
                 }
-                let bytes: Vec<u8> = floats.iter().flat_map(|f| f.to_le_bytes()).collect();
-                Ok(Value::Bytes(bytes))
             }
             _ => Err(format!(
                 "column '{col_name}': expected VECTOR array, got {val:?}"
@@ -175,6 +170,84 @@ fn coerce_value(val: &Value, col_type: &ColumnType, col_name: &str) -> Result<Va
     }
 }
 
+/// Extract f32 floats from a `Value::Array`.
+fn extract_vector_floats(arr: &[Value]) -> Vec<f32> {
+    arr.iter()
+        .filter_map(|v| match v {
+            Value::Float(f) => Some(*f as f32),
+            Value::Integer(n) => Some(*n as f32),
+            _ => None,
+        })
+        .collect()
+}
+
+/// Validate dimension count and encode as little-endian bytes.
+fn validate_and_encode_vector(col_name: &str, dim: u32, floats: &[f32]) -> Result<Value, String> {
+    if floats.len() != dim as usize {
+        return Err(format!(
+            "column '{col_name}': expected VECTOR({dim}), got {} elements",
+            floats.len()
+        ));
+    }
+    let bytes: Vec<u8> = floats.iter().flat_map(|f| f.to_le_bytes()).collect();
+    Ok(Value::Bytes(bytes))
+}
+
+/// Parse a vector from string representations that may arrive via UPDATE path.
+///
+/// Handles formats like:
+/// - `"ArrayLiteral([Literal(Float(0.9)), Literal(Float(0.1)), ...])"` (sqlparser debug repr)
+/// - `"ARRAY[0.1, 0.2, 0.3]"` (SQL literal)
+/// - `"[0.1, 0.2, 0.3]"` (JSON-style)
+fn parse_vector_string(s: &str) -> Option<Vec<f32>> {
+    // Try ARRAY[...] SQL literal format.
+    let upper = s.to_uppercase();
+    if upper.starts_with("ARRAY[") {
+        let start = s.find('[')? + 1;
+        let end = s.rfind(']')?;
+        if end <= start {
+            return None;
+        }
+        let inner = &s[start..end];
+        let floats: Vec<f32> = inner
+            .split(',')
+            .filter_map(|tok| tok.trim().parse::<f32>().ok())
+            .collect();
+        if !floats.is_empty() {
+            return Some(floats);
+        }
+    }
+
+    // Try JSON-style [0.1, 0.2, ...] format.
+    if s.starts_with('[') && s.ends_with(']') {
+        let inner = &s[1..s.len() - 1];
+        let floats: Vec<f32> = inner
+            .split(',')
+            .filter_map(|tok| tok.trim().parse::<f32>().ok())
+            .collect();
+        if !floats.is_empty() {
+            return Some(floats);
+        }
+    }
+
+    // Try sqlparser debug repr: "ArrayLiteral([Literal(Float(0.9)), ...])"
+    if s.starts_with("ArrayLiteral(") {
+        let floats: Vec<f32> = s
+            .split("Float(")
+            .skip(1)
+            .filter_map(|chunk| {
+                let end = chunk.find(')')?;
+                chunk[..end].parse::<f32>().ok()
+            })
+            .collect();
+        if !floats.is_empty() {
+            return Some(floats);
+        }
+    }
+
+    None
+}
+
 /// Convert a typed `Value` to JSON (for pgwire output only).
 fn value_to_json(val: &Value) -> serde_json::Value {
     match val {