velesdb_server/handlers/query/
mod.rs

1//! VelesQL query execution handlers.
2
3pub mod aggregation;
4pub mod explain;
5pub(crate) mod velesql_helpers;
6
7pub use aggregation::__path_aggregate;
8pub use aggregation::aggregate;
9pub use explain::{__path_explain, explain};
10
11use axum::{extract::State, http::StatusCode, response::IntoResponse, Json};
12use std::sync::Arc;
13use velesdb_core::collection::search::query::projection;
14#[cfg(test)]
15use velesdb_core::velesql;
16use velesdb_core::velesql::{DmlStatement, Query, SelectColumns};
17
18use crate::types::{
19    QueryRequest, QueryResponse, QueryResponseMeta, QueryType, VELESQL_CONTRACT_VERSION,
20};
21use crate::AppState;
22
23use crate::handlers::helpers::run_blocking;
24use aggregation::execute_aggregation_query;
25use explain::condition_has_vector_search;
26use velesql_helpers::{parse_and_validate, velesql_collection_not_found, velesql_error};
27
28/// Returns `true` when the query should bypass collection resolution and go
29/// directly through `Database::execute_query` — DDL, introspection, admin,
30/// TRAIN, or graph/edge/delete DML that resolves its own collection from the AST.
31fn requires_mutation_dispatch(parsed: &Query) -> bool {
32    parsed.is_ddl_query()
33        || parsed.is_introspection_query()
34        || parsed.is_admin_query()
35        || parsed.is_train()
36        || is_ast_routed_dml(parsed)
37}
38
39/// Returns `true` for DML statements that resolve their collection name from
40/// the AST rather than from the request body's `FROM` clause:
41/// `INSERT EDGE`, `DELETE`, `DELETE EDGE`, `SELECT EDGES`, `INSERT NODE`.
42///
43/// `INSERT INTO`, `UPSERT`, and `UPDATE` return result rows and must flow
44/// through the standard query path (they use `stmt.table` which maps to
45/// the SELECT `FROM`).
46fn is_ast_routed_dml(parsed: &Query) -> bool {
47    matches!(
48        parsed.dml,
49        Some(
50            DmlStatement::InsertEdge(_)
51                | DmlStatement::Delete(_)
52                | DmlStatement::DeleteEdge(_)
53                | DmlStatement::SelectEdges(_)
54                | DmlStatement::InsertNode(_)
55        )
56    )
57}
58
59/// Execute a VelesQL query.
60///
61/// BUG-1 FIX: Automatically detects aggregation queries (GROUP BY, COUNT, SUM, etc.)
62/// and routes them to execute_aggregate for proper handling.
63///
64/// DDL statements (CREATE/DROP COLLECTION) are intercepted before collection
65/// resolution and dispatched directly through `Database::execute_query`.
66#[utoipa::path(
67    post,
68    path = "/query",
69    tag = "query",
70    request_body = QueryRequest,
71    responses(
72        (status = 200, description = "Query results", body = QueryResponse),
73        (status = 400, description = "Query syntax error", body = crate::types::QueryErrorResponse),
74        (status = 422, description = "Query validation/execution error", body = crate::types::VelesqlErrorResponse),
75        (status = 404, description = "Collection not found", body = crate::types::VelesqlErrorResponse)
76    )
77)]
78pub async fn query(
79    State(state): State<Arc<AppState>>,
80    Json(req): Json<QueryRequest>,
81) -> impl IntoResponse {
82    let start = std::time::Instant::now();
83    state.operational_metrics.inc_queries();
84
85    let parsed = match parse_and_validate(&req.query) {
86        Ok(q) => q,
87        Err(resp) => {
88            state.operational_metrics.inc_errors();
89            return resp;
90        }
91    };
92
93    // Query execution calls synchronous core code (locks, and fsync on the
94    // DML/DDL paths) — run it on the blocking pool so the async workers stay
95    // responsive, mirroring the points/search handler discipline.
96    let state_clone = Arc::clone(&state);
97    run_blocking(move || dispatch_parsed_query(&state_clone, &parsed, &req, start))
98        .await
99        .unwrap_or_else(|resp| resp)
100}
101
102/// Synchronous post-parse dispatch for [`query`]: routes the parsed query to
103/// the mutation, aggregation, or standard execution path and builds the
104/// response. Runs on the blocking pool.
105fn dispatch_parsed_query(
106    state: &Arc<AppState>,
107    parsed: &Query,
108    req: &QueryRequest,
109    start: std::time::Instant,
110) -> axum::response::Response {
111    // DDL/Introspection/Admin/graph-mutation bypass: these extract collection from
112    // the SQL AST, not from the request body.  INSERT INTO, UPSERT, and UPDATE flow
113    // through the standard path because they return meaningful result rows.
114    if requires_mutation_dispatch(parsed) {
115        return execute_mutation_query(state, parsed, &req.params, start);
116    }
117
118    let collection_name = match resolve_collection_name(parsed, req) {
119        Ok(name) => name,
120        Err(resp) => {
121            state.operational_metrics.inc_errors();
122            return resp;
123        }
124    };
125
126    // BUG-1 FIX: Detect aggregation queries and route to execute_aggregate
127    if parsed.select.is_aggregation_query() {
128        return execute_aggregation_query(state, &collection_name, parsed, &req.params, start);
129    }
130
131    let results = match execute_standard_query(state, parsed, &collection_name, req) {
132        Ok(r) => r,
133        Err(resp) => {
134            state.operational_metrics.inc_errors();
135            return resp;
136        }
137    };
138
139    build_query_response(state, start, results, &parsed.select.columns)
140}
141
142/// Execute a DDL, graph/delete DML, introspection, admin, or TRAIN query.
143///
144/// DDL (CREATE/DROP/ALTER/ANALYZE/TRUNCATE), graph/delete DML mutations
145/// (INSERT EDGE, DELETE, DELETE EDGE, SELECT EDGES, INSERT NODE),
146/// introspection (SHOW/DESCRIBE/EXPLAIN), admin (FLUSH), and TRAIN
147/// statements extract collection names from the SQL AST — no FROM clause
148/// needed.
149///
150/// Results from `Database::execute_query` are propagated into the response
151/// so that introspection (SHOW, DESCRIBE, EXPLAIN), ANALYZE, and SELECT
152/// EDGES return their data to the caller.
153fn execute_mutation_query(
154    state: &Arc<AppState>,
155    parsed: &Query,
156    params: &std::collections::HashMap<String, serde_json::Value>,
157    start: std::time::Instant,
158) -> axum::response::Response {
159    match state.db.execute_query(parsed, params) {
160        Ok(results) => build_query_response(state, start, results, &parsed.select.columns),
161        Err(e) => {
162            state.operational_metrics.inc_errors();
163            velesql_error(
164                StatusCode::UNPROCESSABLE_ENTITY,
165                "VELESQL_MUTATION_ERROR",
166                &e.to_string(),
167                "Check collection name, statement syntax, and target existence",
168                None,
169            )
170        }
171    }
172}
173
174/// Determine the target collection from the parsed query and request body.
175#[allow(clippy::result_large_err)]
176fn resolve_collection_name(
177    parsed: &Query,
178    req: &QueryRequest,
179) -> Result<String, axum::response::Response> {
180    if parsed.is_match_query() {
181        req.collection
182            .as_ref()
183            .filter(|name| !name.is_empty())
184            .cloned()
185            .ok_or_else(|| {
186                velesql_error(
187                    StatusCode::UNPROCESSABLE_ENTITY,
188                    "VELESQL_MISSING_COLLECTION",
189                    "MATCH query via /query requires `collection` in request body",
190                    "Add `collection` to the /query JSON body or use /collections/{name}/match",
191                    Some(serde_json::json!({
192                        "field": "collection",
193                        "endpoint": "/query",
194                        "query_type": "MATCH"
195                    })),
196                )
197            })
198    } else {
199        Ok(parsed.select.from.clone())
200    }
201}
202
203/// Execute a standard (non-aggregation) query, dispatching MATCH vs SELECT.
204#[allow(clippy::result_large_err)]
205fn execute_standard_query(
206    state: &Arc<AppState>,
207    parsed: &Query,
208    collection_name: &str,
209    req: &QueryRequest,
210) -> Result<Vec<velesdb_core::SearchResult>, axum::response::Response> {
211    let execute_result = if parsed.is_match_query() {
212        let mut params = req.params.clone();
213        params
214            .entry("_collection".to_string())
215            .or_insert_with(|| serde_json::json!(collection_name));
216        state.db.execute_query(parsed, &params)
217    } else {
218        state.db.execute_query(parsed, &req.params)
219    };
220
221    execute_result.map_err(|e| match e {
222        velesdb_core::Error::CollectionNotFound(name) => velesql_collection_not_found(&name),
223        other => velesql_error(
224            StatusCode::UNPROCESSABLE_ENTITY,
225            "VELESQL_EXECUTION_ERROR",
226            &other.to_string(),
227            "Validate query semantics and parameter types against the target collection",
228            None,
229        ),
230    })
231}
232
233/// Build the final query response with timing metrics and SQL projection.
234///
235/// Both callers ([`execute_standard_query`] and [`execute_mutation_query`])
236/// dispatch through [`Database::execute_query`](velesdb_core::Database::execute_query),
237/// which already fires the observer's `on_query` telemetry exactly once
238/// internally. This function must therefore NOT also call the deprecated
239/// `notify_query_timing`/`notify_query` shim — doing so would double-count
240/// every `/query` request for any registered `DatabaseObserver` (RBAC/audit/
241/// usage billing). Only the Prometheus histogram, which is unrelated to the
242/// observer, is recorded here.
243fn build_query_response(
244    state: &Arc<AppState>,
245    start: std::time::Instant,
246    results: Vec<velesdb_core::SearchResult>,
247    select_columns: &SelectColumns,
248) -> axum::response::Response {
249    let elapsed = start.elapsed();
250    let timing_ms = elapsed.as_secs_f64() * 1000.0;
251    #[allow(clippy::cast_possible_truncation)]
252    // Reason: timing_ms is always < u64::MAX (query durations < 585 millennia)
253    let took_ms = timing_ms.round() as u64;
254    state
255        .query_duration_histogram
256        .observe(elapsed.as_secs_f64());
257    let projected = projection::project_results(&results, select_columns);
258    let rows_returned = projected.len();
259
260    Json(QueryResponse {
261        results: projected,
262        timing_ms,
263        took_ms,
264        rows_returned,
265        meta: QueryResponseMeta {
266            velesql_contract_version: VELESQL_CONTRACT_VERSION.to_string(),
267            count: rows_returned,
268        },
269    })
270    .into_response()
271}
272
273/// Detect query type from parsed AST (EPIC-052 US-006).
274///
275/// Priority order:
276/// 1. DDL (CREATE/DROP COLLECTION) -> Ddl
277/// 2. DML (INSERT/UPDATE/DELETE) -> Dml
278/// 3. MATCH clause -> Graph
279/// 4. GROUP BY or aggregates -> Aggregation
280/// 5. Vector search -> Search
281/// 6. Default -> Rows
282#[allow(dead_code)] // Used in tests, will be used in unified handler
283pub fn detect_query_type(query: &Query) -> QueryType {
284    if query.is_ddl_query() {
285        return QueryType::Ddl;
286    }
287
288    if query.is_dml_query() {
289        return QueryType::Dml;
290    }
291
292    if query.is_match_query() {
293        return QueryType::Graph;
294    }
295
296    if query.select.is_aggregation_query() {
297        return QueryType::Aggregation;
298    }
299
300    let has_vector = query
301        .select
302        .where_clause
303        .as_ref()
304        .map(condition_has_vector_search)
305        .unwrap_or(false);
306
307    if has_vector {
308        return QueryType::Search;
309    }
310
311    QueryType::Rows
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317
318    #[test]
319    fn test_detect_query_type_search() {
320        let parsed = velesql::Parser::parse(
321            "SELECT * FROM docs WHERE similarity(embedding, $v) > 0.8 LIMIT 10",
322        )
323        .unwrap();
324        assert_eq!(detect_query_type(&parsed), QueryType::Search);
325    }
326
327    #[test]
328    fn test_detect_query_type_aggregation() {
329        let parsed =
330            velesql::Parser::parse("SELECT category, COUNT(*) FROM products GROUP BY category")
331                .unwrap();
332        assert_eq!(detect_query_type(&parsed), QueryType::Aggregation);
333    }
334
335    #[test]
336    fn test_detect_query_type_rows() {
337        let parsed =
338            velesql::Parser::parse("SELECT name, price FROM products WHERE price > 100").unwrap();
339        assert_eq!(detect_query_type(&parsed), QueryType::Rows);
340    }
341
342    #[test]
343    fn test_detect_query_type_graph() {
344        let parsed =
345            velesql::Parser::parse("MATCH (n:Person)-[:KNOWS]->(m) RETURN n.name, m.name LIMIT 10")
346                .unwrap();
347        assert_eq!(detect_query_type(&parsed), QueryType::Graph);
348    }
349
350    #[test]
351    fn test_detect_query_type_hybrid_vector_aggregation() {
352        // When both vector search and aggregation, aggregation takes priority
353        let parsed = velesql::Parser::parse(
354            "SELECT category, COUNT(*) FROM docs WHERE similarity(embedding, $v) > 0.7 GROUP BY category",
355        )
356        .unwrap();
357        assert_eq!(detect_query_type(&parsed), QueryType::Aggregation);
358    }
359
360    #[test]
361    fn test_detect_query_type_ddl_create() {
362        let parsed =
363            velesql::Parser::parse("CREATE COLLECTION docs (dimension = 768, metric = 'cosine');")
364                .unwrap();
365        assert_eq!(detect_query_type(&parsed), QueryType::Ddl);
366    }
367
368    #[test]
369    fn test_detect_query_type_ddl_drop() {
370        let parsed = velesql::Parser::parse("DROP COLLECTION docs;").unwrap();
371        assert_eq!(detect_query_type(&parsed), QueryType::Ddl);
372    }
373
374    #[test]
375    fn test_detect_query_type_dml_insert_edge() {
376        let parsed = velesql::Parser::parse(
377            "INSERT EDGE INTO kg (source = 1, target = 2, label = 'KNOWS');",
378        )
379        .unwrap();
380        assert_eq!(detect_query_type(&parsed), QueryType::Dml);
381    }
382
383    #[test]
384    fn test_detect_query_type_dml_delete() {
385        let parsed = velesql::Parser::parse("DELETE FROM docs WHERE id = 1;").unwrap();
386        assert_eq!(detect_query_type(&parsed), QueryType::Dml);
387    }
388}
velesdb_server/handlers/query/mod.rs

velesdb_server/handlers/query/
mod.rs
