Merge pull request #225 from mkhairi/feat/search-vector-subquery-composition

feat(sql): accept SEARCH ... USING VECTOR in subquery position and with quoted collection names

Author: farhan-syah

nodedb-cluster-tests/tests/transport_security.rs

@@ -38,12 +38,12 @@ fn insecure_counter_guard() -> &'static Mutex<()> {
     LOCK.get_or_init(|| Mutex::new(()))
 }
 
+use nodedb_cluster::topology::{ClusterTopology, NodeInfo, NodeState};
 use nodedb_cluster::transport::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
 use nodedb_cluster::{
     MacKey, NexarTransport, RaftRpcHandler, TlsCredentials, TransportCredentials,
     generate_node_credentials, insecure_transport_count, spki_pin_from_cert_der,
 };
-use nodedb_cluster::topology::{ClusterTopology, NodeInfo, NodeState};
 use nodedb_raft::message::{AppendEntriesRequest, AppendEntriesResponse, RequestVoteResponse};
 use nodedb_raft::transport::RaftTransport;
 

nodedb-physical/src/physical_plan/mod.rs

@@ -324,6 +324,8 @@ impl PhysicalPlan {
             | PhysicalPlan::Graph(GraphOp::WccSuperstep(_)) => None,
             // Exchange: recurse into the child plan to extract the collection.
             PhysicalPlan::Query(QueryOp::Exchange(op)) => op.child.collection(),
+            // PostProcess: recurse into the materialized input plan.
+            PhysicalPlan::Query(QueryOp::PostProcess { input, .. }) => input.collection(),
             // ProviderScan is a catalog/constant source — no user collection.
             PhysicalPlan::Query(QueryOp::ProviderScan { .. }) => None,
             // KV ops carry their own collection (sorted-index-only ops → None).

nodedb-physical/src/physical_plan/query.rs

@@ -92,6 +92,46 @@ pub enum QueryOp {
         distinct: bool,
     },
 
+    /// Relational post-processing over a materialized child plan.
+    ///
+    /// Coordinator-resolved, exactly like [`QueryOp::Exchange`]: at resolve time
+    /// the coordinator gathers `input`'s rows, flattens them to the relational
+    /// row shape, and rewrites this node into a [`QueryOp::ProviderScan`]
+    /// carrying the same relational parameters — which applies
+    /// filter→offset→sort→distinct→project→limit on a single core. A Data-Plane
+    /// core must NEVER see this node (defensive error if it does).
+    ///
+    /// Lowered from `SqlPlan::Subquery`: an outer `ORDER BY` / `OFFSET` /
+    /// `DISTINCT` / post-reorder `LIMIT` over a subquery/derived-table body
+    /// whose leaf plan could not absorb those constraints. Constraints the leaf
+    /// DID absorb (a `WHERE` pushed into a search engine, an unordered `LIMIT`
+    /// folded into `top_k`) are applied by `input` and not repeated here.
+    PostProcess {
+        /// The subquery body to materialize. Wrapped in `Exchange{Gather}` by
+        /// the converter when the body is a sharded source, so the gather runs
+        /// exactly once over the full union before post-processing.
+        input: Box<crate::physical_plan::PhysicalPlan>,
+        /// Serialized `Vec<ScanFilter>` (MessagePack). Empty = no predicate.
+        #[serde(default)]
+        filters: Vec<u8>,
+        /// Output column names to keep. Empty = emit all columns.
+        #[serde(default)]
+        projection: Vec<String>,
+        /// Sort keys: `(column_name, ascending)`. Empty = unordered.
+        #[serde(default)]
+        sort_keys: Vec<(String, bool)>,
+        /// Maximum rows to emit after offset/sort/distinct/projection.
+        /// `None` = unlimited.
+        #[serde(default)]
+        limit: Option<usize>,
+        /// Number of rows to skip before applying limit.
+        #[serde(default)]
+        offset: usize,
+        /// SQL DISTINCT: deduplicate on the projected row.
+        #[serde(default)]
+        distinct: bool,
+    },
+
     /// Aggregate: GROUP BY + aggregate functions.
     Aggregate {
         collection: String,

nodedb-physical/src/physical_plan/routing.rs

@@ -75,6 +75,13 @@ pub fn plan_contains_cluster_partitioned_leaf(plan: &PhysicalPlan) -> bool {
             plan_contains_cluster_partitioned_leaf(child)
         }
 
+        // Recurse through PostProcess (its materialized input is the real
+        // plan). Normally resolved to a `ProviderScan` before routing, but a
+        // conservative recursion keeps routing correct if one survives.
+        PhysicalPlan::Query(QueryOp::PostProcess { input, .. }) => {
+            plan_contains_cluster_partitioned_leaf(input)
+        }
+
         // Recurse through lateral outer plans.
         PhysicalPlan::Query(QueryOp::LateralTopK { outer_plan, .. })
         | PhysicalPlan::Query(QueryOp::LateralLoop { outer_plan, .. }) => {

nodedb-sql/src/parser/preprocess/pipeline.rs

@@ -13,7 +13,9 @@
 use super::function_args::rewrite_object_literal_args;
 use super::lex::{first_sql_word, has_brace_outside_literals, has_operator_outside_literals};
 use super::object_literal_stmt::try_rewrite_object_literal;
-use super::search_vector::try_rewrite_search_using_vector;
+use super::search_vector::{
+    try_rewrite_nested_search_using_vector, try_rewrite_search_using_vector,
+};
 use super::temporal::extract as extract_temporal;
 use super::vector_ops::{
     rewrite_arrow_distance, rewrite_cosine_distance, rewrite_neg_inner_product,
@@ -56,10 +58,16 @@ pub fn preprocess(sql: &str) -> Result<Option<PreprocessedSql>, SqlError> {
     // `SELECT * FROM <coll> ORDER BY vector_distance(...) LIMIT k` before any
     // first-word dispatch — the rewritten form re-enters the rest of the
     // pipeline as a plain SELECT.
+    //
+    // The same clause in subquery position — `FROM (SEARCH ...) s` — is
+    // rewritten in place so a k-NN result composes with joins and filters.
     let (temporal_sql, search_vector_rewritten) =
         match try_rewrite_search_using_vector(&temporal_sql) {
             Some(rewritten) => (rewritten, true),
-            None => (temporal_sql, false),
+            None => match try_rewrite_nested_search_using_vector(&temporal_sql) {
+                Some(rewritten) => (rewritten, true),
+                None => (temporal_sql, false),
+            },
         };
 
     let first_word = first_sql_word(&temporal_sql)