Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion nodedb-cluster-tests/tests/transport_security.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
2 changes: 2 additions & 0 deletions nodedb-physical/src/physical_plan/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
40 changes: 40 additions & 0 deletions nodedb-physical/src/physical_plan/query.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
7 changes: 7 additions & 0 deletions nodedb-physical/src/physical_plan/routing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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, .. }) => {
Expand Down
12 changes: 10 additions & 2 deletions nodedb-sql/src/parser/preprocess/pipeline.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down
Loading