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    pub id: Option<String>,
11    pub index: Option<String>,
12    pub score: Option<f64>,
13}
14
15#[derive(Debug, Clone, PartialEq)]
16pub struct DslPage {
17    pub hits: Vec<DslHit>,
18    pub total: Option<u64>,
19}
20
21pub fn decode(value: &Value) -> Result<DslPage> {
22    let hits = value
23        .pointer("/hits/hits")
24        .and_then(Value::as_array)
25        .ok_or_else(|| {
26            Error::new(
27                ErrorKind::Http,
28                "decoding search response field `hits.hits`",
29            )
30        })?;
31    let out = hits
32        .iter()
33        .map(|h| {
34            Ok(DslHit {
35                source: h.get("_source").cloned().unwrap_or(Value::Null),
36                sort: h.get("sort").and_then(Value::as_array).cloned(),
37                id: h.get("_id").and_then(Value::as_str).map(str::to_owned),
38                index: h.get("_index").and_then(Value::as_str).map(str::to_owned),
39                score: h.get("_score").and_then(Value::as_f64),
40            })
41        })
42        .collect::<Result<Vec<_>>>()?;
43    let total = value.pointer("/hits/total/value").and_then(Value::as_u64);
44    Ok(DslPage { hits: out, total })
45}
46
47/// Run one bounded `POST /<index>/_search` with the operator's body verbatim.
48/// This is the peek path — one request, no PIT.
49pub async fn run_sync(t: &Transport, index: &str, body: &Value) -> Result<DslPage> {
50    let response = t
51        .post_absolute_es(&format!("/{index}/_search"), body)
52        .await?;
53    decode(&response)
54}
55
56/// Normalize `sort` to a total order by appending `_shard_doc` (ascending)
57/// when absent. A non-total sort makes `search_after` skip or repeat documents
58/// across pages, so every export pages over a total order.
59fn total_sort(sort: &Value) -> Value {
60    let mut entries = match sort {
61        Value::Array(items) => items.clone(),
62        other => vec![other.clone()],
63    };
64    let has_tiebreaker = entries.iter().any(|entry| {
65        entry
66            .as_object()
67            .is_some_and(|obj| obj.contains_key("_shard_doc"))
68    });
69    if !has_tiebreaker {
70        entries.push(json!({"_shard_doc": "asc"}));
71    }
72    Value::Array(entries)
73}
74
75/// Open a point-in-time on `index` and page it fully with `search_after`.
76/// `query` is the operator's filter; `sort` is normalized to a total order by
77/// appending `_shard_doc` when absent. The PIT is closed on every exit path,
78/// success or error.
79pub async fn run_stream(
80    t: &Transport,
81    index: &str,
82    query: &Value,
83    sort: &Value,
84    limit: Option<usize>,
85) -> Result<Vec<DslHit>> {
86    let open: Value = t
87        .post_absolute_es(&format!("/{index}/_pit?keep_alive=1m"), &json!({}))
88        .await?;
89    let pit_id = open
90        .get("id")
91        .and_then(Value::as_str)
92        .ok_or_else(|| Error::new(ErrorKind::Http, "decoding _pit open response field `id`"))?
93        .to_string();
94
95    let sort = total_sort(sort);
96    let result = page_loop(t, &pit_id, query, &sort, limit).await;
97
98    // Close the PIT on every path. The `_pit` delete takes the id in the
99    // request body, so it uses the body-carrying DELETE. The `let _` swallow is
100    // deliberate: the PIT self-expires after `keep_alive`, and any `page_loop`
101    // error is the error to surface, not a best-effort close failure.
102    let _ = t
103        .delete_absolute_es_json("/_pit", &json!({ "id": pit_id }))
104        .await;
105    result
106}
107
108async fn page_loop(
109    t: &Transport,
110    pit_id: &str,
111    query: &Value,
112    sort: &Value,
113    limit: Option<usize>,
114) -> Result<Vec<DslHit>> {
115    let mut all = Vec::new();
116    let mut search_after: Option<Vec<Value>> = None;
117    let size = limit.map_or(1000, |n| n.min(1000));
118    loop {
119        let mut body = json!({
120            "size": size,
121            "sort": sort,
122            "pit": { "id": pit_id, "keep_alive": "1m" },
123            "query": query
124        });
125        if let Some(sa) = &search_after {
126            body["search_after"] = json!(sa);
127        }
128        let page = decode(&t.post_absolute_es("/_search", &body).await?)?;
129        if page.hits.is_empty() {
130            return Ok(all);
131        }
132        let last_sort = page.hits.last().and_then(|h| h.sort.clone());
133        all.extend(page.hits);
134        if let Some(limit) = limit
135            && all.len() >= limit
136        {
137            all.truncate(limit);
138            return Ok(all);
139        }
140        match last_sort {
141            Some(sa) => search_after = Some(sa),
142            None => return Ok(all),
143        }
144    }
145}
elasticctl_api/search/dsl.rs

elasticctl_api/search/
dsl.rs
