linkrs_wire/
query.rs

1//! Query contract DTOs (HTTP `/query` endpoints).
2//!
3//! Previously mirrored between `api/server/http/handlers/query_types.rs` and
4//! `cli/client/{types,request_types,response_types}.rs`; this is the single
5//! source of truth for both sides.
6
7use serde::{Deserialize, Serialize};
8use std::collections::HashMap;
9
10/// Query request
11#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
12pub struct QueryRequest {
13    pub query: String,
14    pub session_id: i64,
15    /// Query parameters bound to `@name` references in the statement.
16    #[serde(default)]
17    pub parameters: HashMap<String, serde_json::Value>,
18    /// Session variables bound to `$name` references in the statement.
19    /// When omitted, the session-managed snapshot (set via `LET $name = expr`)
20    /// is used.
21    #[serde(default)]
22    pub session_variables: HashMap<String, serde_json::Value>,
23    /// Consistency level: None = eventual (default), Some("read_your_writes") enables RYW.
24    #[serde(default)]
25    pub consistency: Option<String>,
26    #[serde(default)]
27    pub consistency_timeout_ms: Option<u64>,
28    #[serde(default)]
29    pub minimum_lsn: Option<u64>,
30}
31
32/// Batch query request: multiple auto-commit DML statements executed inside a
33/// single shared auto-commit batch window.
34#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
35pub struct BatchQueryRequest {
36    pub session_id: i64,
37    pub statements: Vec<String>,
38    /// Query parameters bound to `@name` references in every statement.
39    #[serde(default)]
40    pub parameters: HashMap<String, serde_json::Value>,
41    /// Session variables bound to `$name` references in every statement.
42    /// When omitted, the session-managed snapshot is used.
43    #[serde(default)]
44    pub session_variables: HashMap<String, serde_json::Value>,
45}
46
47/// Batch query response: one [`QueryResponse`] per input statement, in order.
48#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
49pub struct BatchQueryResponse {
50    pub results: Vec<QueryResponse>,
51}
52
53/// Query response (structured)
54#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
55pub struct QueryResponse {
56    pub success: bool,
57    #[serde(default)]
58    pub data: Option<QueryData>,
59    #[serde(default)]
60    pub error: Option<QueryError>,
61    #[serde(default)]
62    pub metadata: QueryMetadata,
63}
64
65/// Query data
66#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
67pub struct QueryData {
68    #[serde(default)]
69    pub columns: Vec<String>,
70    #[serde(default)]
71    pub rows: Vec<HashMap<String, serde_json::Value>>,
72    #[serde(default)]
73    pub row_count: usize,
74}
75
76/// Per-stage timings in milliseconds for query portraits.
77#[derive(Debug, Clone, Serialize, Deserialize, Default, utoipa::ToSchema)]
78pub struct QueryStageTimings {
79    #[serde(default)]
80    pub parse_ms: f64,
81    #[serde(default)]
82    pub validate_ms: f64,
83    #[serde(default)]
84    pub plan_ms: f64,
85    #[serde(default)]
86    pub optimize_ms: f64,
87    #[serde(default)]
88    pub execute_ms: f64,
89}
90
91/// Query metadata
92#[derive(Debug, Clone, Serialize, Deserialize, Default, utoipa::ToSchema)]
93pub struct QueryMetadata {
94    #[serde(default)]
95    pub execution_time_ms: u64,
96    #[serde(default)]
97    pub rows_scanned: u64,
98    #[serde(default)]
99    pub rows_returned: usize,
100    #[serde(default)]
101    pub space_id: Option<u64>,
102    /// The result was cut at the configured row ceiling; `rows_returned`
103    /// holds the rows actually delivered.
104    #[serde(default)]
105    pub truncated: bool,
106    #[serde(default, skip_serializing_if = "Option::is_none")]
107    pub trace_id: Option<String>,
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub stages: Option<QueryStageTimings>,
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub plan_node_count: Option<usize>,
112    #[serde(default, skip_serializing_if = "Option::is_none")]
113    pub result_row_count: Option<usize>,
114}
115
116/// Error position within the query text (1-based line and column).
117#[derive(Debug, Clone, Copy, Serialize, Deserialize, utoipa::ToSchema)]
118pub struct ErrorPosition {
119    pub line: usize,
120    pub column: usize,
121}
122
123/// Query error.
124#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
125pub struct QueryError {
126    pub code: String,
127    pub message: String,
128    #[serde(default)]
129    pub details: Option<String>,
130    #[serde(default, skip_serializing_if = "Option::is_none")]
131    pub position: Option<ErrorPosition>,
132}
133
134/// Verify the response.
135#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
136pub struct ValidateRequest {
137    pub query: String,
138    pub session_id: i64,
139    #[serde(default)]
140    pub need_estimate: bool,
141}
142
143/// Explain request: plan a statement without executing it.
144#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
145pub struct ExplainRequest {
146    pub query: String,
147    pub session_id: i64,
148    /// Query parameters bound to `@name` references in the statement.
149    #[serde(default)]
150    pub parameters: HashMap<String, serde_json::Value>,
151    /// Session variables bound to `$name` references in the statement.
152    #[serde(default)]
153    pub session_variables: HashMap<String, serde_json::Value>,
154}
155
156/// Verify the response.
157#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
158pub struct ValidateResponse {
159    pub valid: bool,
160    pub message: String,
161    /// Planner root-operator row estimate for automatic routing.
162    /// Absent when the statement is not stream-shaped or the plan
163    /// carries no estimate; callers treat a missing value as "stream".
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub estimated_rows: Option<u64>,
166}
167
168/// Streaming query request (SSE `/stream` endpoint).
169///
170/// Single-statement mode uses `query`. Batch-streaming mode uses
171/// `statements` (with shared `parameters` / `session_variables`, mirroring
172/// [`BatchQueryRequest`]) and emits one `statement_begin` /
173/// `statement_end` event pair per statement. The two modes are mutually
174/// exclusive: a non-empty `query` together with non-empty `statements` is
175/// rejected by the server.
176#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
177pub struct StreamQueryRequest {
178    pub query: String,
179    pub session_id: i64,
180    #[serde(default = "default_buffer_capacity")]
181    pub event_buffer_capacity: usize,
182    /// Batch-streaming statements. Empty means single-statement mode.
183    #[serde(default)]
184    pub statements: Vec<String>,
185    /// Query parameters bound to `@name` references in every statement
186    /// (batch-streaming mode only).
187    #[serde(default)]
188    pub parameters: HashMap<String, serde_json::Value>,
189    /// Session variables bound to `$name` references in every statement
190    /// (batch-streaming mode only). When omitted, the session-managed
191    /// snapshot is used, same as [`BatchQueryRequest`].
192    #[serde(default)]
193    pub session_variables: HashMap<String, serde_json::Value>,
194    /// Stop the batch at the first failing statement (default true).
195    /// Already produced statements are kept either way.
196    #[serde(default = "default_fail_fast")]
197    pub fail_fast: bool,
198}
199
200fn default_fail_fast() -> bool {
201    true
202}
203
204/// Open a forward-only cursor over a single statement's result.
205#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
206pub struct OpenCursorRequest {
207    pub session_id: i64,
208    pub query: String,
209}
210
211/// Open cursor response: server-assigned id plus upfront column names.
212#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
213pub struct OpenCursorResponse {
214    pub cursor_id: u64,
215    #[serde(default)]
216    pub columns: Vec<String>,
217}
218
219/// Fetch one page from a cursor.
220#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
221pub struct FetchCursorRequest {
222    pub session_id: i64,
223    pub cursor_id: u64,
224    #[serde(default = "default_cursor_page_size")]
225    pub page_size: usize,
226}
227
228/// One fetched page: rows in column order plus exhaustion flag.
229#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
230pub struct FetchCursorResponse {
231    #[serde(default)]
232    pub columns: Vec<String>,
233    #[serde(default)]
234    pub rows: Vec<HashMap<String, serde_json::Value>>,
235    pub has_more: bool,
236    pub returned: usize,
237    #[serde(default, skip_serializing_if = "Option::is_none")]
238    pub trace_id: Option<String>,
239    #[serde(default, skip_serializing_if = "Option::is_none")]
240    pub stages: Option<QueryStageTimings>,
241}
242
243/// Release a cursor.
244#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
245pub struct CloseCursorRequest {
246    pub session_id: i64,
247    pub cursor_id: u64,
248}
249
250/// Cursor release outcome: false when the id was already gone.
251#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
252pub struct CloseCursorResponse {
253    pub closed: bool,
254}
255
256fn default_cursor_page_size() -> usize {
257    500
258}
259
260fn default_buffer_capacity() -> usize {
261    100
262}
263
264impl QueryResponse {
265    /// A successful response has been created.
266    pub fn success(data: QueryData, metadata: QueryMetadata) -> Self {
267        Self {
268            success: true,
269            data: Some(data),
270            error: None,
271            metadata,
272        }
273    }
274
275    /// Creating an error response
276    pub fn error(code: String, message: String, details: Option<String>) -> Self {
277        Self::error_with_position(code, message, details, None)
278    }
279
280    /// Creating an error response with an optional source position.
281    pub fn error_with_position(
282        code: String,
283        message: String,
284        details: Option<String>,
285        position: Option<ErrorPosition>,
286    ) -> Self {
287        Self {
288            success: false,
289            data: None,
290            error: Some(QueryError {
291                code,
292                message,
293                details,
294                position,
295            }),
296            metadata: QueryMetadata::default(),
297        }
298    }
299}
300
301impl QueryData {
302    /// Create empty query data.
303    pub fn empty() -> Self {
304        Self {
305            columns: Vec::new(),
306            rows: Vec::new(),
307            row_count: 0,
308        }
309    }
310
311    /// Create query data from columns and rows.
312    pub fn new(columns: Vec<String>, rows: Vec<HashMap<String, serde_json::Value>>) -> Self {
313        let row_count = rows.len();
314        Self {
315            columns,
316            rows,
317            row_count,
318        }
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325
326    fn parse(json: &str) -> serde_json::Result<QueryResponse> {
327        serde_json::from_str(json)
328    }
329
330    #[test]
331    fn deserialize_success_envelope() {
332        let result = parse(
333            r#"{
334                "success": true,
335                "data": {
336                    "columns": ["name", "age"],
337                    "rows": [{"name": "Alice", "age": 30}],
338                    "row_count": 1
339                },
340                "error": null,
341                "metadata": {
342                    "execution_time_ms": 12,
343                    "rows_scanned": 42,
344                    "rows_returned": 1,
345                    "space_id": null
346                }
347            }"#,
348        )
349        .expect("success envelope should parse");
350
351        let data = result.data.expect("data should be present");
352        assert_eq!(data.columns, vec!["name", "age"]);
353        assert_eq!(data.row_count, 1);
354        assert_eq!(data.rows.len(), 1);
355        assert_eq!(result.metadata.execution_time_ms, 12);
356        assert_eq!(result.metadata.rows_scanned, 42);
357        assert!(result.error.is_none());
358        assert!(result.success);
359    }
360
361    #[test]
362    fn deserialize_error_envelope() {
363        let result = parse(
364            r#"{
365                "success": false,
366                "data": null,
367                "error": {
368                    "code": "QUERY_ERROR",
369                    "message": "syntax error",
370                    "details": null
371                },
372                "metadata": {
373                    "execution_time_ms": 0,
374                    "rows_scanned": 0,
375                    "rows_returned": 0,
376                    "space_id": null
377                }
378            }"#,
379        )
380        .expect("error envelope should parse");
381
382        assert!(result.data.is_none());
383        let error = result.error.expect("error should be present");
384        assert_eq!(error.code, "QUERY_ERROR");
385        assert_eq!(error.message, "syntax error");
386        assert!(!result.success);
387    }
388
389    #[test]
390    fn deserialize_use_statement_space_id() {
391        // USE results carry a space_id in the metadata group.
392        let result = parse(
393            r#"{
394                "success": true,
395                "data": {
396                    "columns": ["space_name", "space_id", "vid_type"],
397                    "rows": [
398                        {
399                            "space_name": "test_space",
400                            "space_id": 1,
401                            "vid_type": "INT64"
402                        }
403                    ],
404                    "row_count": 1
405                },
406                "error": null,
407                "metadata": {
408                    "execution_time_ms": 1,
409                    "rows_scanned": 0,
410                    "rows_returned": 1,
411                    "space_id": 1
412                }
413            }"#,
414        )
415        .expect("USE envelope should parse");
416
417        let data = result.data.expect("data should be present");
418        assert_eq!(data.row_count, 1);
419        assert_eq!(
420            data.rows[0].get("space_id"),
421            Some(&serde_json::Value::from(1))
422        );
423        assert_eq!(result.metadata.space_id, Some(1));
424    }
425
426    #[test]
427    fn stream_batch_request_defaults() {
428        let request: StreamQueryRequest = serde_json::from_str(r#"{"query": "", "session_id": 3}"#)
429            .expect("stream request should parse");
430        assert!(request.statements.is_empty());
431        assert!(request.parameters.is_empty());
432        assert!(request.session_variables.is_empty());
433        assert!(request.fail_fast);
434
435        let batch: StreamQueryRequest = serde_json::from_str(
436            r#"{
437                "query": "",
438                "session_id": 3,
439                "statements": ["RETURN 1", "RETURN 2"],
440                "parameters": {"p": 1},
441                "fail_fast": false
442            }"#,
443        )
444        .expect("batch stream request should parse");
445        assert_eq!(batch.statements.len(), 2);
446        assert_eq!(batch.parameters.get("p"), Some(&serde_json::Value::from(1)));
447        assert!(!batch.fail_fast);
448    }
449
450    #[test]
451    fn validate_request_estimate_defaults_off() {
452        let request: ValidateRequest =
453            serde_json::from_str(r#"{"query": "RETURN 1", "session_id": 3}"#)
454                .expect("validate request should parse");
455        assert!(!request.need_estimate);
456
457        let with_estimate: ValidateRequest = serde_json::from_str(
458            r#"{"query": "RETURN 1", "session_id": 3, "need_estimate": true}"#,
459        )
460        .expect("validate request with estimate should parse");
461        assert!(with_estimate.need_estimate);
462    }
463
464    #[test]
465    fn metadata_extension_defaults() {
466        let meta: QueryMetadata = serde_json::from_str(
467            r#"{"execution_time_ms": 5, "rows_scanned": 1, "rows_returned": 1}"#,
468        )
469        .expect("old metadata must parse");
470        assert!(meta.trace_id.is_none());
471        assert!(meta.stages.is_none());
472        assert!(meta.plan_node_count.is_none());
473        let full = QueryMetadata {
474            execution_time_ms: 5,
475            rows_scanned: 1,
476            rows_returned: 1,
477            space_id: None,
478            truncated: false,
479            trace_id: Some("trace-1".to_string()),
480            stages: Some(QueryStageTimings {
481                parse_ms: 0.1,
482                validate_ms: 0.1,
483                plan_ms: 0.2,
484                optimize_ms: 0.1,
485                execute_ms: 4.5,
486            }),
487            plan_node_count: Some(3),
488            result_row_count: Some(1),
489        };
490        let json = serde_json::to_string(&full).unwrap();
491        let back: QueryMetadata = serde_json::from_str(&json).unwrap();
492        assert_eq!(back.trace_id.as_deref(), Some("trace-1"));
493        assert_eq!(back.plan_node_count, Some(3));
494    }
495
496    #[test]
497    fn request_roundtrip() {
498        let request = QueryRequest {
499            query: "MATCH (n) RETURN n".to_string(),
500            session_id: 7,
501            parameters: HashMap::from([("p".to_string(), serde_json::json!(42))]),
502            session_variables: HashMap::new(),
503            consistency: None,
504            consistency_timeout_ms: None,
505            minimum_lsn: None,
506        };
507        let json = serde_json::to_string(&request).unwrap();
508        let back: QueryRequest = serde_json::from_str(&json).unwrap();
509        assert_eq!(back.query, request.query);
510        assert_eq!(back.session_id, 7);
511        assert_eq!(back.parameters.get("p"), Some(&serde_json::json!(42)));
512    }
513}
linkrs_wire/query.rs

linkrs_wire/
query.rs
