refactor(pgwire): split ddl/dsl.rs into per-concern modules

dsl.rs has grown to cover DDL parsing for search indexes, CRDT merge,
FTS, vector, sparse, and fusion in a single flat file. Split it into a
dsl/ directory with one module per concern:

  crdt_merge, fulltext_index, search_fusion, search_index,
  search_vector, sparse_index, vector_index, helpers, mod

No logic changes — pure file reorganisation.

Author: farhan-syah

nodedb/src/control/server/pgwire/ddl/dsl.rs

@@ -0,0 +1,95 @@
+//! `CRDT MERGE INTO` DSL handler.
+
+use std::time::Duration;
+
+use pgwire::api::results::{Response, Tag};
+use pgwire::error::PgWireResult;
+
+use crate::bridge::envelope::PhysicalPlan;
+use crate::bridge::physical_plan::CrdtOp;
+use crate::control::security::identity::AuthenticatedIdentity;
+use crate::control::server::pgwire::types::sqlstate_error;
+use crate::control::state::SharedState;
+
+/// CRDT MERGE INTO <collection> FROM '<source_id>' TO '<target_id>'
+pub async fn crdt_merge(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    parts: &[&str],
+) -> PgWireResult<Vec<Response>> {
+    if parts.len() < 7 {
+        return Err(sqlstate_error(
+            "42601",
+            "syntax: CRDT MERGE INTO <collection> FROM '<source_id>' TO '<target_id>'",
+        ));
+    }
+
+    let collection = parts[3];
+    let tenant_id = identity.tenant_id;
+
+    let from_idx = parts
+        .iter()
+        .position(|p| p.eq_ignore_ascii_case("FROM"))
+        .ok_or_else(|| sqlstate_error("42601", "expected FROM keyword"))?;
+    let to_idx = parts
+        .iter()
+        .position(|p| p.eq_ignore_ascii_case("TO"))
+        .ok_or_else(|| sqlstate_error("42601", "expected TO keyword"))?;
+
+    let source_id = parts
+        .get(from_idx + 1)
+        .map(|s| s.trim_matches('\'').trim_matches('"'))
+        .ok_or_else(|| sqlstate_error("42601", "missing source document ID"))?;
+    let target_id = parts
+        .get(to_idx + 1)
+        .map(|s| s.trim_matches('\'').trim_matches('"'))
+        .ok_or_else(|| sqlstate_error("42601", "missing target document ID"))?;
+
+    let source_plan = PhysicalPlan::Crdt(CrdtOp::Read {
+        collection: collection.to_string(),
+        document_id: source_id.to_string(),
+    });
+
+    let source_bytes = crate::control::server::pgwire::ddl::sync_dispatch::dispatch_async(
+        state,
+        tenant_id,
+        collection,
+        source_plan,
+        Duration::from_secs(state.tuning.network.default_deadline_secs),
+    )
+    .await
+    .map_err(|e| sqlstate_error("XX000", &e.to_string()))?;
+    if source_bytes.is_empty() {
+        return Err(sqlstate_error(
+            "02000",
+            &format!("source document '{source_id}' not found"),
+        ));
+    }
+
+    let apply_plan = PhysicalPlan::Crdt(CrdtOp::Apply {
+        collection: collection.to_string(),
+        document_id: target_id.to_string(),
+        delta: source_bytes,
+        peer_id: identity.user_id,
+        mutation_id: 0,
+    });
+
+    crate::control::server::pgwire::ddl::sync_dispatch::dispatch_async(
+        state,
+        tenant_id,
+        collection,
+        apply_plan,
+        Duration::from_secs(state.tuning.network.default_deadline_secs),
+    )
+    .await
+    .map_err(|e| sqlstate_error("XX000", &e.to_string()))?;
+
+    state.audit_record(
+        crate::control::security::audit::AuditEvent::AdminAction,
+        Some(tenant_id),
+        &identity.username,
+        &format!("CRDT merge: {source_id} → {target_id} in '{collection}'"),
+    );
+
+    Ok(vec![Response::Execution(Tag::new("CRDT MERGE"))])
+}

nodedb/src/control/server/pgwire/ddl/dsl/crdt_merge.rs

@@ -0,0 +1,47 @@
+//! `CREATE FULLTEXT INDEX` DSL handler.
+
+use pgwire::api::results::{Response, Tag};
+use pgwire::error::PgWireResult;
+
+use crate::control::security::identity::AuthenticatedIdentity;
+use crate::control::server::pgwire::types::sqlstate_error;
+use crate::control::state::SharedState;
+
+/// CREATE FULLTEXT INDEX <name> ON <collection> (<field>)
+pub fn create_fulltext_index(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    parts: &[&str],
+) -> PgWireResult<Vec<Response>> {
+    if parts.len() < 7 {
+        return Err(sqlstate_error(
+            "42601",
+            "syntax: CREATE FULLTEXT INDEX <name> ON <collection> (<field>)",
+        ));
+    }
+
+    let index_name = parts[3];
+    if !parts[4].eq_ignore_ascii_case("ON") {
+        return Err(sqlstate_error("42601", "expected ON after index name"));
+    }
+    let collection = parts[5];
+    let field = parts[6].trim_matches(|c| c == '(' || c == ')');
+    let tenant_id = identity.tenant_id;
+
+    super::super::owner_propose::propose_owner(
+        state,
+        "fulltext_index",
+        tenant_id,
+        index_name,
+        &identity.username,
+    )?;
+
+    state.audit_record(
+        crate::control::security::audit::AuditEvent::AdminAction,
+        Some(tenant_id),
+        &identity.username,
+        &format!("created fulltext index '{index_name}' on '{collection}' ({field})"),
+    );
+
+    Ok(vec![Response::Execution(Tag::new("CREATE FULLTEXT INDEX"))])
+}

nodedb/src/control/server/pgwire/ddl/dsl/fulltext_index.rs

@@ -0,0 +1,35 @@
+//! Shared parameter-extraction helpers for DSL handlers.
+
+pub(super) fn extract_param(upper: &str, name: &str) -> Option<usize> {
+    let idx = upper.find(name)?;
+    let rest = &upper[idx + name.len()..];
+    rest.split(|c: char| !c.is_ascii_digit())
+        .find(|s| !s.is_empty())
+        .and_then(|s| s.parse().ok())
+}
+
+pub(super) fn extract_string_param(sql: &str, name: &str) -> Option<String> {
+    let upper = sql.to_uppercase();
+    let idx = upper.find(name)?;
+    let rest = &sql[idx + name.len()..];
+    let rest = rest.trim();
+    if rest.starts_with('\'') || rest.starts_with('"') {
+        let quote = rest.chars().next()?;
+        let end = rest[1..].find(quote)?;
+        Some(rest[1..end + 1].to_string())
+    } else {
+        rest.split_whitespace().next().map(|s| s.to_string())
+    }
+}
+
+pub(super) fn find_param_str(upper_parts: &[String], name: &str) -> Option<String> {
+    let idx = upper_parts.iter().position(|p| p == name)?;
+    upper_parts.get(idx + 1).cloned()
+}
+
+pub(super) fn find_param_usize(upper_parts: &[String], name: &str) -> Option<usize> {
+    let idx = upper_parts.iter().position(|p| p == name)?;
+    upper_parts
+        .get(idx + 1)
+        .and_then(|s| s.parse::<usize>().ok())
+}

nodedb/src/control/server/pgwire/ddl/dsl/helpers.rs

@@ -0,0 +1,27 @@
+//! NodeDB DSL extensions — custom SQL-like commands beyond standard SQL.
+//!
+//! - SEARCH <collection> USING VECTOR(<field>, ARRAY[...], <k>)
+//! - SEARCH <collection> USING FUSION(vector=..., graph=..., top_k=...)
+//! - CREATE VECTOR INDEX <name> ON <collection> [METRIC ...] [M ...] [EF_CONSTRUCTION ...] [DIM ...]
+//!   [INDEX_TYPE hnsw|hnsw_pq|ivf_pq] [PQ_M ...] [IVF_CELLS ...] [IVF_NPROBE ...]
+//! - CREATE FULLTEXT INDEX <name> ON <collection> (<field>)
+//! - CREATE SEARCH INDEX ON <collection> FIELDS ...
+//! - CREATE SPARSE INDEX [name] ON <collection> (<field>)
+//! - CRDT MERGE INTO <collection> FROM <source_id> TO <target_id>
+
+mod crdt_merge;
+mod fulltext_index;
+mod helpers;
+mod search_fusion;
+mod search_index;
+mod search_vector;
+mod sparse_index;
+mod vector_index;
+
+pub use crdt_merge::crdt_merge;
+pub use fulltext_index::create_fulltext_index;
+pub use search_fusion::search_fusion;
+pub use search_index::create_search_index;
+pub use search_vector::search_vector;
+pub use sparse_index::create_sparse_index;
+pub use vector_index::create_vector_index;

nodedb/src/control/server/pgwire/ddl/dsl/mod.rs

@@ -0,0 +1,100 @@
+//! `SEARCH <collection> USING FUSION(...)` DSL (vector + graph fusion).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures::stream;
+use pgwire::api::results::{DataRowEncoder, QueryResponse, Response};
+use pgwire::error::PgWireResult;
+
+use crate::bridge::envelope::PhysicalPlan;
+use crate::bridge::physical_plan::GraphOp;
+use crate::control::security::identity::AuthenticatedIdentity;
+use crate::control::server::pgwire::types::{sqlstate_error, text_field};
+use crate::control::state::SharedState;
+
+use super::helpers::{extract_param, extract_string_param};
+
+/// SEARCH <collection> USING FUSION(VECTOR(ARRAY[...], <k>), GRAPH(<label>, <depth>), TOP <n>)
+pub async fn search_fusion(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    sql: &str,
+) -> PgWireResult<Vec<Response>> {
+    let parts: Vec<&str> = sql.split_whitespace().collect();
+    if parts.len() < 4 {
+        return Err(sqlstate_error(
+            "42601",
+            "syntax: SEARCH <collection> USING FUSION(...)",
+        ));
+    }
+    let collection = parts[1];
+    let tenant_id = identity.tenant_id;
+
+    let array_start = sql.find("ARRAY[").or_else(|| sql.find("array["));
+    let array_start = match array_start {
+        Some(i) => i + 6,
+        None => {
+            return Err(sqlstate_error("42601", "expected ARRAY[...] in FUSION"));
+        }
+    };
+    let array_end = sql[array_start..].find(']').map(|i| i + array_start);
+    let array_end = match array_end {
+        Some(i) => i,
+        None => {
+            return Err(sqlstate_error("42601", "unterminated ARRAY["));
+        }
+    };
+
+    let vector_str = &sql[array_start..array_end];
+    let query_vector: Vec<f32> = vector_str
+        .split(',')
+        .filter_map(|s| s.trim().parse::<f32>().ok())
+        .collect();
+
+    if query_vector.is_empty() {
+        return Err(sqlstate_error("42601", "empty query vector in FUSION"));
+    }
+
+    let upper = sql.to_uppercase();
+    let vector_top_k = extract_param(&upper, "VECTOR_TOP_K").unwrap_or(20);
+    let expansion_depth = extract_param(&upper, "DEPTH").unwrap_or(2);
+    let final_top_k = extract_param(&upper, "TOP").unwrap_or(10);
+
+    let edge_label = extract_string_param(sql, "LABEL");
+
+    let plan = PhysicalPlan::Graph(GraphOp::RagFusion {
+        collection: collection.to_string(),
+        query_vector: Arc::from(query_vector.as_slice()),
+        vector_top_k,
+        edge_label,
+        direction: crate::engine::graph::edge_store::Direction::Out,
+        expansion_depth,
+        final_top_k,
+        rrf_k: (60.0, 60.0),
+        options: crate::engine::graph::traversal_options::GraphTraversalOptions::default(),
+    });
+
+    let payload = crate::control::server::pgwire::ddl::sync_dispatch::dispatch_async(
+        state,
+        tenant_id,
+        collection,
+        plan,
+        Duration::from_secs(state.tuning.network.default_deadline_secs),
+    )
+    .await
+    .map_err(|e| sqlstate_error("XX000", &e.to_string()))?;
+
+    let schema = Arc::new(vec![text_field("result")]);
+    let text = crate::data::executor::response_codec::decode_payload_to_json(&payload);
+    let mut encoder = DataRowEncoder::new(schema.clone());
+    encoder
+        .encode_field(&text)
+        .map_err(|e| sqlstate_error("XX000", &e.to_string()))?;
+    let row = encoder.take_row();
+
+    Ok(vec![Response::Query(QueryResponse::new(
+        schema,
+        stream::iter(vec![Ok(row)]),
+    ))])
+}

nodedb/src/control/server/pgwire/ddl/dsl/search_fusion.rs

@@ -0,0 +1,71 @@
+//! `CREATE SEARCH INDEX` DSL handler (higher-level alias for fulltext).
+
+use pgwire::api::results::{Response, Tag};
+use pgwire::error::PgWireResult;
+
+use crate::control::security::identity::AuthenticatedIdentity;
+use crate::control::server::pgwire::types::sqlstate_error;
+use crate::control::state::SharedState;
+
+/// CREATE SEARCH INDEX ON <collection> FIELDS <field1>[, <field2>...] [ANALYZER '<name>'] [FUZZY true|false]
+pub fn create_search_index(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    sql: &str,
+) -> PgWireResult<Vec<Response>> {
+    let upper = sql.to_uppercase();
+
+    let on_pos = upper.find(" ON ").ok_or_else(|| {
+        sqlstate_error(
+            "42601",
+            "syntax: CREATE SEARCH INDEX ON <collection> FIELDS <field> [ANALYZER 'name'] [FUZZY true]",
+        )
+    })?;
+    let after_on = sql[on_pos + 4..].trim_start();
+    let fields_pos = upper.find(" FIELDS ").ok_or_else(|| {
+        sqlstate_error(
+            "42601",
+            "syntax: CREATE SEARCH INDEX ON <collection> FIELDS <field> [ANALYZER 'name'] [FUZZY true]",
+        )
+    })?;
+
+    let collection = after_on[..fields_pos - on_pos - 4].trim().to_lowercase();
+    if collection.is_empty() {
+        return Err(sqlstate_error("42601", "missing collection name"));
+    }
+
+    let after_fields = &sql[fields_pos + 8..];
+    let fields_end = upper[fields_pos + 8..]
+        .find(" ANALYZER ")
+        .or_else(|| upper[fields_pos + 8..].find(" FUZZY "))
+        .unwrap_or(after_fields.len());
+    let fields_str = after_fields[..fields_end].trim();
+    let fields: Vec<&str> = fields_str.split(',').map(|s| s.trim()).collect();
+
+    if fields.is_empty() || fields[0].is_empty() {
+        return Err(sqlstate_error("42601", "missing field list"));
+    }
+
+    let tenant_id = identity.tenant_id;
+
+    for field in &fields {
+        let index_name = format!("fts_{}_{}", collection, field);
+
+        super::super::owner_propose::propose_owner(
+            state,
+            "fulltext_index",
+            tenant_id,
+            &index_name,
+            &identity.username,
+        )?;
+
+        state.audit_record(
+            crate::control::security::audit::AuditEvent::AdminAction,
+            Some(tenant_id),
+            &identity.username,
+            &format!("created search index '{index_name}' on '{collection}' ({field})"),
+        );
+    }
+
+    Ok(vec![Response::Execution(Tag::new("CREATE SEARCH INDEX"))])
+}