fix(sql): apply outer ORDER BY / OFFSET / DISTINCT over subquery bodies

A subquery/derived-table body that cannot absorb the outer query's
constraints (e.g. a k-NN search leaf under an outer ORDER BY / OFFSET /
DISTINCT, or a reorder-dependent LIMIT) previously dropped those
constraints silently, returning the body's raw rows.

Add SqlPlan::Subquery, lowered to a coordinator-resolved
QueryOp::PostProcess that gathers the body's rows and applies
filter -> offset -> sort -> distinct -> project -> limit exactly once
over the full result, then wire it through every plan visitor,
routing, RLS injection, authorization, schema-derivation, and
materialization site that matches on SqlPlan/QueryOp/PhysicalPlan.

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/types/plan/cacheability.rs

@@ -76,6 +76,7 @@ impl SqlPlan {
                 .fold(outer.cache_eligibility(), |eligibility, (_, plan)| {
                     eligibility.combine(plan.cache_eligibility())
                 }),
+            Self::Subquery { input, .. } => input.cache_eligibility(),
             Self::LateralTopK { outer, .. } => outer.cache_eligibility(),
             Self::LateralLoop { outer, inner, .. } => {
                 outer.cache_eligibility().combine(inner.cache_eligibility())

nodedb-sql/src/types/plan/variants.rs

@@ -478,6 +478,40 @@ pub enum SqlPlan {
         outer: Box<SqlPlan>,
     },
 
+    /// Relational post-processing over a subquery/derived-table body whose leaf
+    /// plan cannot absorb the outer query's constraints.
+    ///
+    /// Produced by CTE / derived-table inlining when the referenced body is not
+    /// a plain `Scan` (which carries its own filters/sort/limit/offset/distinct)
+    /// and the outer reference adds constraints the body has no slot for — an
+    /// `ORDER BY`, `OFFSET`, `DISTINCT`, or a `LIMIT` that must apply *after* a
+    /// reorder. Without this node those constraints were silently dropped,
+    /// returning the body's unordered/unoffset/undeduplicated rows.
+    ///
+    /// The Data Plane materializes `input`'s rows, then applies, in this order:
+    /// filter → offset → sort → distinct → project → limit — matching
+    /// `QueryOp::ProviderScan` semantics (which this lowers onto). Constraints
+    /// the body CAN absorb (e.g. an unordered `LIMIT` folded into a vector
+    /// search's `top_k`, or a `WHERE` pushed into the engine as a post-filter)
+    /// are applied at the leaf during inlining and are NOT repeated here.
+    Subquery {
+        /// The subquery body whose rows are post-processed.
+        input: Box<SqlPlan>,
+        /// Predicates applied to the materialized rows (outer `WHERE` that the
+        /// body leaf did not absorb).
+        filters: Vec<Filter>,
+        /// Outer projection (target list). Empty = inherit the body's columns.
+        projection: Vec<Projection>,
+        /// Outer `ORDER BY` keys applied over the materialized rows.
+        sort_keys: Vec<SortKey>,
+        /// Outer `OFFSET` (0 = none).
+        offset: usize,
+        /// Outer `DISTINCT`.
+        distinct: bool,
+        /// Outer `LIMIT` (`None` = unbounded).
+        limit: Option<usize>,
+    },
+
     // ── Array (ND sparse) ─────────────────────────────────────
     /// `CREATE ARRAY <name> DIMS (...) ATTRS (...) TILE_EXTENTS (...)`.
     /// AST is engine-agnostic — the Origin converter builds the typed

nodedb-sql/src/visitor/plan_visitor/args.rs

@@ -32,6 +32,17 @@ pub struct ScanVisitArgs<'a> {
     pub temporal: &'a TemporalScope,
 }
 
+/// Parameters for [`super::trait_def::PlanVisitor::subquery`].
+pub struct SubqueryVisitArgs<'a> {
+    pub input: &'a SqlPlan,
+    pub filters: &'a [Filter],
+    pub projection: &'a [Projection],
+    pub sort_keys: &'a [SortKey],
+    pub offset: usize,
+    pub distinct: bool,
+    pub limit: Option<usize>,
+}
+
 /// Parameters for [`super::trait_def::PlanVisitor::document_index_lookup`].
 pub struct DocumentIndexLookupVisitArgs<'a> {
     pub collection: &'a str,

nodedb-sql/src/visitor/plan_visitor/dispatch_rest.rs

@@ -22,6 +22,23 @@ pub(super) fn dispatch_rest<V: PlanVisitor>(
         SqlPlan::Intersect { left, right, all } => visitor.intersect(left, right, *all),
         SqlPlan::Except { left, right, all } => visitor.except(left, right, *all),
         SqlPlan::Cte { definitions, outer } => visitor.cte(definitions, outer),
+        SqlPlan::Subquery {
+            input,
+            filters,
+            projection,
+            sort_keys,
+            offset,
+            distinct,
+            limit,
+        } => visitor.subquery(super::args::SubqueryVisitArgs {
+            input,
+            filters,
+            projection,
+            sort_keys,
+            offset: *offset,
+            distinct: *distinct,
+            limit: *limit,
+        }),
         SqlPlan::CreateArray {
             name,
             dims,

nodedb-sql/src/visitor/plan_visitor/trait_def.rs

@@ -9,8 +9,8 @@ use super::args::{
     AggregateVisitArgs, CreateArrayVisitArgs, DocumentIndexLookupVisitArgs,
     HybridSearchTripleVisitArgs, HybridSearchVisitArgs, InsertVisitArgs, JoinVisitArgs,
     LateralLoopVisitArgs, LateralTopKVisitArgs, MergeVisitArgs, RecursiveScanVisitArgs,
-    RecursiveValueVisitArgs, ScanVisitArgs, SpatialScanVisitArgs, TimeseriesScanVisitArgs,
-    UpdateFromVisitArgs, UpsertVisitArgs, VectorSearchVisitArgs,
+    RecursiveValueVisitArgs, ScanVisitArgs, SpatialScanVisitArgs, SubqueryVisitArgs,
+    TimeseriesScanVisitArgs, UpdateFromVisitArgs, UpsertVisitArgs, VectorSearchVisitArgs,
 };
 use crate::fts_types::FtsQuery;
 use crate::temporal::TemporalScope;
@@ -234,6 +234,9 @@ pub trait PlanVisitor {
         outer: &SqlPlan,
     ) -> Result<Self::Output, Self::Error>;
 
+    /// Handle [`SqlPlan::Subquery`].
+    fn subquery(&mut self, args: SubqueryVisitArgs<'_>) -> Result<Self::Output, Self::Error>;
+
     /// Handle [`SqlPlan::CreateArray`].
     fn create_array(&mut self, args: CreateArrayVisitArgs<'_>)
     -> Result<Self::Output, Self::Error>;

nodedb/src/control/exec_receiver/support.rs

@@ -64,6 +64,10 @@ pub(super) fn plan_contains_exchange(plan: &PhysicalPlan) -> bool {
                 .any(|child| plan_contains_exchange(child)),
             QueryOp::LateralTopK { outer_plan, .. } => plan_contains_exchange(outer_plan),
             QueryOp::LateralLoop { outer_plan, .. } => plan_contains_exchange(outer_plan),
+            // PostProcess wraps a materialized child that, before coordinator
+            // resolution, still carries an `Exchange{Gather}` — recurse so an
+            // unresolved PostProcess is correctly flagged as Exchange-bearing.
+            QueryOp::PostProcess { input, .. } => plan_contains_exchange(input),
             // Aggregate may carry a sub-plan input (catalog `ProviderScan`),
             // which could in principle nest an Exchange — recurse when present.
             QueryOp::Aggregate { input, .. } => {