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

velesdb_server/handlers/query/
mod.rs
