elasticctl_api/search/
dsl.rs

1//! Query DSL responses and PIT + `search_after` pagination.
2
3use elasticctl_core::{Error, ErrorKind, Result, Transport};
4use serde_json::{Value, json};
5
6#[derive(Debug, Clone, PartialEq)]
7pub struct DslHit {
8    pub source: Value,
9    pub sort: Option<Vec<Value>>,
10}
11
12#[derive(Debug, Clone, PartialEq)]
13pub struct DslPage {
14    pub hits: Vec<DslHit>,
15    pub total: Option<u64>,
16}
17
18pub fn decode(value: &Value) -> Result<DslPage> {
19    let hits = value
20        .pointer("/hits/hits")
21        .and_then(Value::as_array)
22        .ok_or_else(|| {
23            Error::new(
24                ErrorKind::Http,
25                "decoding search response field `hits.hits`",
26            )
27        })?;
28    let out = hits
29        .iter()
30        .map(|h| {
31            Ok(DslHit {
32                source: h.get("_source").cloned().unwrap_or(Value::Null),
33                sort: h.get("sort").and_then(Value::as_array).cloned(),
34            })
35        })
36        .collect::<Result<Vec<_>>>()?;
37    let total = value.pointer("/hits/total/value").and_then(Value::as_u64);
38    Ok(DslPage { hits: out, total })
39}
40
41/// Run one bounded `POST /<index>/_search` with the operator's body verbatim.
42/// This is the peek path — one request, no PIT.
43pub async fn run_sync(t: &Transport, index: &str, body: &Value) -> Result<DslPage> {
44    let response = t
45        .post_absolute_es(&format!("/{index}/_search"), body)
46        .await?;
47    decode(&response)
48}
49
50/// Open a point-in-time on `index` and page it fully with `search_after`.
51/// `query` is the operator's filter; `sort` must be a total order ending in
52/// `_shard_doc`. The PIT is closed on every exit path, success or error.
53pub async fn run_stream(
54    t: &Transport,
55    index: &str,
56    query: &Value,
57    sort: &Value,
58    limit: Option<usize>,
59) -> Result<Vec<DslHit>> {
60    let open: Value = t
61        .post_absolute_es(&format!("/{index}/_pit?keep_alive=1m"), &json!({}))
62        .await?;
63    let pit_id = open
64        .get("id")
65        .and_then(Value::as_str)
66        .ok_or_else(|| Error::new(ErrorKind::Http, "decoding _pit open response field `id`"))?
67        .to_string();
68
69    let result = page_loop(t, &pit_id, query, sort, limit).await;
70
71    // Close the PIT on every path. The `_pit` delete takes the id in the
72    // request body, so it uses the body-carrying DELETE. The `let _` swallow is
73    // deliberate: the PIT self-expires after `keep_alive`, and any `page_loop`
74    // error is the error to surface, not a best-effort close failure.
75    let _ = t
76        .delete_absolute_es_json("/_pit", &json!({ "id": pit_id }))
77        .await;
78    result
79}
80
81async fn page_loop(
82    t: &Transport,
83    pit_id: &str,
84    query: &Value,
85    sort: &Value,
86    limit: Option<usize>,
87) -> Result<Vec<DslHit>> {
88    let mut all = Vec::new();
89    let mut search_after: Option<Vec<Value>> = None;
90    loop {
91        let mut body = json!({
92            "size": 1000,
93            "sort": sort,
94            "pit": { "id": pit_id, "keep_alive": "1m" },
95            "query": query
96        });
97        if let Some(sa) = &search_after {
98            body["search_after"] = json!(sa);
99        }
100        let page = decode(&t.post_absolute_es("/_search", &body).await?)?;
101        if page.hits.is_empty() {
102            return Ok(all);
103        }
104        let last_sort = page.hits.last().and_then(|h| h.sort.clone());
105        all.extend(page.hits);
106        if let Some(limit) = limit
107            && all.len() >= limit
108        {
109            all.truncate(limit);
110            return Ok(all);
111        }
112        match last_sort {
113            Some(sa) => search_after = Some(sa),
114            None => return Ok(all),
115        }
116    }
117}
elasticctl_api/search/dsl.rs

elasticctl_api/search/
dsl.rs
