velesdb_server/handlers/query/
aggregation.rs

1//! Aggregation query dispatch and execution.
2//!
3//! Handles detection and execution of GROUP BY / aggregate function queries,
4//! routing them to `execute_aggregate` on the appropriate collection.
5
6use axum::{http::StatusCode, response::IntoResponse, Json};
7use std::sync::Arc;
8use velesdb_core::velesql::Query;
9
10use crate::handlers::helpers::notify_query_timing;
11use crate::types::{
12    AggregationResponse, QueryRequest, QueryResponseMeta, VELESQL_CONTRACT_VERSION,
13};
14use crate::AppState;
15
16use super::velesql_helpers::{parse_and_validate, velesql_collection_not_found, velesql_error};
17
18fn aggregation_result_count(result: &serde_json::Value) -> usize {
19    match result {
20        serde_json::Value::Array(rows) => rows.len(),
21        serde_json::Value::Object(_) => 1,
22        _ => 0,
23    }
24}
25
26pub(crate) fn execute_aggregation_query(
27    state: &Arc<AppState>,
28    collection_name: &str,
29    parsed: &Query,
30    params: &std::collections::HashMap<String, serde_json::Value>,
31    start: std::time::Instant,
32) -> axum::response::Response {
33    // Prefer typed vector collection for aggregation.
34    let result = if let Some(vc) = state.db.get_vector_collection(collection_name) {
35        vc.execute_aggregate(parsed, params)
36    } else if let Some(any) = state.db.get_any_collection(collection_name) {
37        any.execute_aggregate(parsed, params)
38    } else {
39        state.operational_metrics.inc_errors();
40        return velesql_collection_not_found(collection_name);
41    };
42
43    let result = match result {
44        Ok(r) => r,
45        Err(e) => {
46            state.operational_metrics.inc_errors();
47            return velesql_error(
48                StatusCode::UNPROCESSABLE_ENTITY,
49                "VELESQL_AGGREGATION_ERROR",
50                &e.to_string(),
51                "Verify GROUP BY/HAVING clauses and aggregate function arguments",
52                None,
53            );
54        }
55    };
56
57    let elapsed = start.elapsed();
58    let timing_ms = elapsed.as_secs_f64() * 1000.0;
59    notify_query_timing(state, collection_name, start);
60    state
61        .query_duration_histogram
62        .observe(elapsed.as_secs_f64());
63    let count = aggregation_result_count(&result);
64
65    Json(AggregationResponse {
66        result,
67        timing_ms,
68        meta: QueryResponseMeta {
69            velesql_contract_version: VELESQL_CONTRACT_VERSION.to_string(),
70            count,
71        },
72    })
73    .into_response()
74}
75
76/// Resolve the collection name for an aggregation query.
77#[allow(clippy::result_large_err)]
78pub(crate) fn resolve_aggregate_collection(
79    parsed: &Query,
80    req: &QueryRequest,
81) -> Result<String, axum::response::Response> {
82    if !parsed.select.from.is_empty() {
83        return Ok(parsed.select.from.clone());
84    }
85    req.collection
86        .as_ref()
87        .filter(|name| !name.is_empty())
88        .cloned()
89        .ok_or_else(|| {
90            velesql_error(
91                StatusCode::UNPROCESSABLE_ENTITY,
92                "VELESQL_MISSING_COLLECTION",
93                "Aggregation query requires a FROM collection or request-body `collection`",
94                "Add FROM <collection> to query or set `collection` in request JSON",
95                Some(serde_json::json!({
96                    "field": "collection",
97                    "endpoint": "/aggregate"
98                })),
99            )
100        })
101}
102
103/// Execute an aggregation-only VelesQL query.
104///
105/// This endpoint is explicit and stable for GROUP BY / HAVING / aggregate workloads.
106#[utoipa::path(
107    post,
108    path = "/aggregate",
109    tag = "query",
110    request_body = QueryRequest,
111    responses(
112        (status = 200, description = "Aggregation results", body = AggregationResponse),
113        (status = 400, description = "Query syntax error", body = crate::types::QueryErrorResponse),
114        (status = 422, description = "Aggregation validation/execution error", body = crate::types::VelesqlErrorResponse),
115        (status = 404, description = "Collection not found", body = crate::types::VelesqlErrorResponse)
116    )
117)]
118#[allow(clippy::unused_async)]
119pub async fn aggregate(
120    axum::extract::State(state): axum::extract::State<Arc<AppState>>,
121    Json(req): Json<QueryRequest>,
122) -> impl IntoResponse {
123    let start = std::time::Instant::now();
124    state.operational_metrics.inc_queries();
125
126    let parsed = match parse_and_validate(&req.query) {
127        Ok(q) => q,
128        Err(resp) => {
129            state.operational_metrics.inc_errors();
130            return resp;
131        }
132    };
133
134    if parsed.is_match_query() || !parsed.select.is_aggregation_query() {
135        state.operational_metrics.inc_errors();
136        return velesql_error(
137            StatusCode::UNPROCESSABLE_ENTITY,
138            "VELESQL_AGGREGATION_ERROR",
139            "Only aggregation queries are accepted on /aggregate",
140            "Use /query for row/search/graph queries; use /aggregate for GROUP BY/aggregate workloads.",
141            Some(serde_json::json!({ "endpoint": "/aggregate" })),
142        );
143    }
144
145    let collection_name = resolve_aggregate_collection(&parsed, &req);
146    let collection_name = match collection_name {
147        Ok(name) => name,
148        Err(resp) => {
149            state.operational_metrics.inc_errors();
150            return resp;
151        }
152    };
153
154    execute_aggregation_query(&state, &collection_name, &parsed, &req.params, start)
155}
velesdb_server/handlers/query/aggregation.rs

velesdb_server/handlers/query/
aggregation.rs
