elasticctl_api/search/
esql.rs

1//! ES|QL columnar responses and the sync `/_query` runner.
2
3use elasticctl_core::{Error, ErrorKind, Result, Transport};
4use serde_json::Value;
5
6#[derive(Debug, Clone, PartialEq)]
7pub struct EsqlColumn {
8    pub name: String,
9    pub r#type: String,
10}
11
12#[derive(Debug, Clone, PartialEq)]
13pub struct EsqlResponse {
14    pub columns: Vec<EsqlColumn>,
15    pub values: Vec<Vec<Value>>,
16    pub is_partial: bool,
17}
18
19/// Strict decode of a `POST /_query` response. Unknown fields are accepted; a
20/// missing or mistyped `columns`/`values` is an error, never an empty result.
21pub fn decode(value: &Value) -> Result<EsqlResponse> {
22    let obj = value.as_object().ok_or_else(|| {
23        Error::new(
24            ErrorKind::Http,
25            "decoding esql response: expected an object",
26        )
27    })?;
28
29    let columns = obj
30        .get("columns")
31        .and_then(Value::as_array)
32        .ok_or_else(|| Error::new(ErrorKind::Http, "decoding esql response field `columns`"))?;
33    let mut out = Vec::with_capacity(columns.len());
34    for (i, col) in columns.iter().enumerate() {
35        let name = col.get("name").and_then(Value::as_str).ok_or_else(|| {
36            Error::new(
37                ErrorKind::Http,
38                format!("decoding esql response column {i} field `name`"),
39            )
40        })?;
41        let ty = col.get("type").and_then(Value::as_str).ok_or_else(|| {
42            Error::new(
43                ErrorKind::Http,
44                format!("decoding esql response column {i} field `type`"),
45            )
46        })?;
47        out.push(EsqlColumn {
48            name: name.to_string(),
49            r#type: ty.to_string(),
50        });
51    }
52
53    let values = obj
54        .get("values")
55        .and_then(Value::as_array)
56        .ok_or_else(|| Error::new(ErrorKind::Http, "decoding esql response field `values`"))?
57        .iter()
58        .map(|row| {
59            row.as_array().cloned().ok_or_else(|| {
60                Error::new(
61                    ErrorKind::Http,
62                    "decoding esql response: `values` rows must be arrays",
63                )
64            })
65        })
66        .collect::<Result<Vec<_>>>()?;
67
68    let is_partial = obj
69        .get("is_partial")
70        .and_then(Value::as_bool)
71        .unwrap_or(false);
72
73    Ok(EsqlResponse {
74        columns: out,
75        values,
76        is_partial,
77    })
78}
79
80/// Run a synchronous ES|QL query. `query` carries its own `FROM` and `LIMIT`.
81pub async fn run_sync(t: &Transport, query: &str) -> Result<EsqlResponse> {
82    let body = serde_json::json!({ "query": query });
83    let response = t.post_absolute_es("/_query", &body).await?;
84    decode(&response)
85}
86
87/// Run a query through the async API and poll until complete. ES|QL has no
88/// page-by-page cursor; the full result returns in one response.
89pub async fn run_async(t: &Transport, query: &str) -> Result<EsqlResponse> {
90    let start = t
91        .post_absolute_es(
92            "/_query/async",
93            &serde_json::json!({ "query": query, "wait_for_completion_timeout": "1ms", "columnar": false }),
94        )
95        .await?;
96    // A query finishing within wait_for_completion_timeout returns the inline
97    // result with is_running: false and no `id`; decode it directly.
98    let id = match start.get("id").and_then(Value::as_str) {
99        Some(id) => id.to_string(),
100        None => return decode(&start),
101    };
102    if start.get("is_running").and_then(Value::as_bool) == Some(false) {
103        return decode(&start);
104    }
105
106    // Poll at most 30 times; an async query that never finishes is a timeout,
107    // not an infinite loop. Clean up on the way out either way.
108    for _ in 0..30 {
109        let resp = t
110            .get_absolute_es(&format!(
111                "/_query/async/{id}?wait_for_completion_timeout=10s"
112            ))
113            .await?;
114        if resp.get("is_running").and_then(Value::as_bool) == Some(false) {
115            let _ = t.delete_absolute_es(&format!("/_query/async/{id}")).await;
116            return decode(&resp);
117        }
118    }
119    let _ = t.delete_absolute_es(&format!("/_query/async/{id}")).await;
120    Err(Error::new(
121        ErrorKind::Timeout,
122        format!("async query {id} still running after 30 polls"),
123    ))
124}
elasticctl_api/search/esql.rs

elasticctl_api/search/
esql.rs
