rustlavel-search 0.4.0

Elasticsearch and OpenSearch for Rustlavel: indexing, search, and bulk operations
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
//! Reading a search response.
//!
//! Two things in this shape are routinely misread, and both lose information
//! silently rather than loudly:
//!
//! - **The total is a relation, not a number.** Since Elasticsearch 7 the
//!   counting stops at 10,000 by default and the response says "at least
//!   10,000". A caller that reads `value` alone will print "10,000 results"
//!   above a list of millions. [`TotalHits`] refuses to be an integer for that
//!   reason.
//! - **A search can succeed on some shards and fail on others.** The status is
//!   200, the hits are there, and a slice of the index was never consulted.
//!   [`SearchResults::is_partial`] is the only thing that says so.

use crate::error::{Result, SearchError};
use rustlavel_core::Json;
use std::collections::BTreeMap;

/// Everything a search came back with.
#[derive(Debug, Clone, PartialEq)]
pub struct SearchResults {
    /// Milliseconds the cluster spent, excluding the network.
    pub took: u64,
    /// The search hit its time limit and returned what it had.
    pub timed_out: bool,
    pub shards: Shards,
    pub total: TotalHits,
    /// The best score, or `None` when the search was sorted by a field — in
    /// that case nothing was scored at all.
    pub max_score: Option<f64>,
    pub hits: Vec<Hit>,
    pub aggregations: Aggregations,
}

impl SearchResults {
    pub fn parse(body: &Json) -> Result<SearchResults> {
        let hits = body.get("hits").ok_or_else(|| SearchError::Malformed {
            context: "_search".to_string(),
            message: "the reply has no `hits`".to_string(),
        })?;

        Ok(SearchResults {
            took: u64_at(body, "took"),
            timed_out: body.get("timed_out").and_then(Json::as_bool).unwrap_or(false),
            shards: Shards::parse(body.get("_shards")),
            total: TotalHits::parse(hits.get("total")),
            max_score: hits.get("max_score").and_then(Json::as_f64),
            hits: hits
                .get("hits")
                .and_then(Json::as_array)
                .map(|items| items.iter().map(Hit::parse).collect())
                .unwrap_or_default(),
            aggregations: Aggregations::parse(body.get("aggregations")),
        })
    }

    /// How many documents came back on this page — not how many matched.
    pub fn len(&self) -> usize {
        self.hits.len()
    }

    pub fn is_empty(&self) -> bool {
        self.hits.is_empty()
    }

    /// Whether part of the index was not consulted.
    ///
    /// A search over a cluster with a failing shard answers 200 with fewer
    /// results, and nothing in the hits distinguishes that from a query that
    /// genuinely matched less. Anything acting on absence — "no orders, so
    /// close the account" — has to check this first.
    pub fn is_partial(&self) -> bool {
        self.timed_out || self.shards.failed > 0
    }

    /// The `_source` of every hit, in order.
    pub fn sources(&self) -> impl Iterator<Item = &Json> {
        self.hits.iter().map(|hit| &hit.source)
    }

    /// The `_id` of every hit, in order.
    pub fn ids(&self) -> Vec<&str> {
        self.hits.iter().map(|hit| hit.id.as_str()).collect()
    }

    /// The first hit, for a search that expects at most one.
    pub fn first(&self) -> Option<&Hit> {
        self.hits.first()
    }
}

/// How many shards answered.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Shards {
    pub total: u32,
    pub successful: u32,
    pub skipped: u32,
    pub failed: u32,
    /// Why the failed ones failed, where the cluster said.
    pub failures: Vec<String>,
}

impl Shards {
    fn parse(body: Option<&Json>) -> Shards {
        let Some(body) = body else {
            return Shards::default();
        };

        Shards {
            total: u32_at(body, "total"),
            successful: u32_at(body, "successful"),
            skipped: u32_at(body, "skipped"),
            failed: u32_at(body, "failed"),
            failures: body
                .get("failures")
                .and_then(Json::as_array)
                .map(|items| {
                    items
                        .iter()
                        .map(|failure| {
                            failure
                                .get("reason.reason")
                                .or_else(|| failure.get("reason"))
                                .and_then(Json::as_str)
                                .unwrap_or("no reason given")
                                .to_string()
                        })
                        .collect()
                })
                .unwrap_or_default(),
        }
    }
}

/// How the total relates to the real number of matches.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TotalRelation {
    /// The number is the answer.
    Exact,
    /// Counting stopped early; the real number is this or larger.
    AtLeast,
}

/// How many documents matched.
///
/// Not a `u64`, deliberately. Elasticsearch stops counting at
/// `track_total_hits` — 10,000 unless asked otherwise — and reports the cutoff
/// as the value with a `gte` relation. Anything that renders a count, decides
/// how many pages there are, or compares against an expected number has to
/// know which of the two it is holding.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TotalHits {
    pub value: u64,
    pub relation: TotalRelation,
}

impl TotalHits {
    fn parse(body: Option<&Json>) -> TotalHits {
        let Some(body) = body else {
            return TotalHits { value: 0, relation: TotalRelation::Exact };
        };

        // Elasticsearch 6, and OpenSearch asked for `rest_total_hits_as_int`,
        // send a bare number here. It has no cutoff, so it is always exact.
        if let Some(value) = body.as_i64() {
            return TotalHits { value: value.max(0) as u64, relation: TotalRelation::Exact };
        }

        TotalHits {
            value: body.get("value").and_then(Json::as_i64).unwrap_or(0).max(0) as u64,
            relation: match body.get("relation").and_then(Json::as_str) {
                Some("gte") => TotalRelation::AtLeast,
                _ => TotalRelation::Exact,
            },
        }
    }

    /// Whether the number can be trusted as the whole answer.
    pub fn is_exact(&self) -> bool {
        self.relation == TotalRelation::Exact
    }

    /// The count, but only when it is really the count.
    ///
    /// Use this where being wrong matters — a "1 of N" label, a page count.
    /// [`TotalHits::value`] is there for a progress bar, where a floor is fine.
    pub fn exact(&self) -> Option<u64> {
        self.is_exact().then_some(self.value)
    }
}

impl std::fmt::Display for TotalHits {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.relation {
            TotalRelation::Exact => write!(f, "{}", self.value),
            TotalRelation::AtLeast => write!(f, "at least {}", self.value),
        }
    }
}

/// One matching document.
#[derive(Debug, Clone, PartialEq)]
pub struct Hit {
    pub index: String,
    pub id: String,
    /// `None` when the search was sorted by a field rather than by relevance.
    pub score: Option<f64>,
    /// The document, or the part of it `_source` filtering asked for.
    pub source: Json,
    /// The values this hit sorted on, for paging with `search_after`.
    pub sort: Vec<Json>,
}

impl Hit {
    fn parse(body: &Json) -> Hit {
        Hit {
            index: string_at(body, "_index"),
            id: string_at(body, "_id"),
            score: body.get("_score").and_then(Json::as_f64),
            source: body.get("_source").cloned().unwrap_or(Json::Null),
            sort: body.get("sort").and_then(Json::as_array).map(<[Json]>::to_vec).unwrap_or_default(),
        }
    }

    /// A field of the document, by dotted path.
    pub fn field(&self, path: &str) -> Option<&Json> {
        self.source.get(path)
    }

    /// A string field of the document, by dotted path.
    pub fn string(&self, path: &str) -> Option<&str> {
        self.source.get(path).and_then(Json::as_str)
    }
}

/// The aggregation results, keyed by the names the search gave them.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Aggregations(BTreeMap<String, Json>);

impl Aggregations {
    fn parse(body: Option<&Json>) -> Aggregations {
        Aggregations(
            body.and_then(Json::as_object).cloned().unwrap_or_default(),
        )
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    pub fn names(&self) -> Vec<&str> {
        self.0.keys().map(String::as_str).collect()
    }

    /// One result as the cluster sent it, for an aggregation this module does
    /// not model.
    pub fn raw(&self, name: &str) -> Option<&Json> {
        self.0.get(name)
    }

    /// A `terms` result.
    pub fn terms(&self, name: &str) -> Option<TermsResult> {
        let body = self.0.get(name)?;
        let buckets = body.get("buckets")?.as_array()?;

        Some(TermsResult {
            doc_count_error_upper_bound: u64_at(body, "doc_count_error_upper_bound"),
            sum_other_doc_count: u64_at(body, "sum_other_doc_count"),
            buckets: buckets.iter().map(Bucket::parse).collect(),
        })
    }

    /// A `stats` result.
    pub fn stats(&self, name: &str) -> Option<StatsResult> {
        let body = self.0.get(name)?;

        Some(StatsResult {
            count: u64_at(body, "count"),
            // Null on every one of these when the aggregation saw no document
            // — there is no minimum of nothing, and reporting 0 would be a
            // number somebody charts.
            min: body.get("min").and_then(Json::as_f64),
            max: body.get("max").and_then(Json::as_f64),
            avg: body.get("avg").and_then(Json::as_f64),
            sum: body.get("sum").and_then(Json::as_f64).unwrap_or(0.0),
        })
    }

    /// The number from a single-value metric — `min`, `max`, `avg`, `sum`,
    /// `cardinality`, `value_count`.
    pub fn value(&self, name: &str) -> Option<f64> {
        self.0.get(name)?.get("value")?.as_f64()
    }
}

/// What a `terms` aggregation returned.
#[derive(Debug, Clone, PartialEq)]
pub struct TermsResult {
    pub buckets: Vec<Bucket>,
    /// The largest number of documents any *returned* bucket may be missing.
    pub doc_count_error_upper_bound: u64,
    /// Documents in the values that did not make the list.
    pub sum_other_doc_count: u64,
}

impl TermsResult {
    /// Whether values are missing from this answer.
    ///
    /// A `terms` aggregation returns the top few — ten by default — and says
    /// nothing else about the rest. Charting the buckets as "the tags" when
    /// this is true draws a picture that is quietly wrong, and raising `size`
    /// is the fix.
    pub fn is_approximate(&self) -> bool {
        self.sum_other_doc_count > 0 || self.doc_count_error_upper_bound > 0
    }

    /// A bucket by its key, rendered as a string.
    pub fn get(&self, key: &str) -> Option<&Bucket> {
        self.buckets.iter().find(|bucket| bucket.key_string() == key)
    }

    /// The count for one key, or zero if it is not in the answer.
    ///
    /// Zero is honest only when [`TermsResult::is_approximate`] is false.
    pub fn count(&self, key: &str) -> u64 {
        self.get(key).map(|bucket| bucket.doc_count).unwrap_or(0)
    }
}

/// One bucket of an aggregation.
#[derive(Debug, Clone, PartialEq)]
pub struct Bucket {
    /// The value, which is a string for a `keyword` field and a number for a
    /// numeric or date one.
    pub key: Json,
    pub doc_count: u64,
    /// Whatever was nested inside this bucket.
    pub aggregations: Aggregations,
}

impl Bucket {
    fn parse(body: &Json) -> Bucket {
        Bucket {
            key: body.get("key").cloned().unwrap_or(Json::Null),
            doc_count: u64_at(body, "doc_count"),
            // The bucket object holds its sub-aggregations alongside `key` and
            // `doc_count`; those two extra names cost nothing and keeping the
            // whole object means a sub-aggregation is read exactly the way a
            // top-level one is.
            aggregations: Aggregations::parse(Some(body)),
        }
    }

    /// The key as text.
    ///
    /// Prefers `key_as_string`, which a date histogram sends alongside the raw
    /// epoch milliseconds and which is the only readable form of it.
    pub fn key_string(&self) -> String {
        if let Some(Json::String(text)) = self.aggregations.raw("key_as_string") {
            return text.clone();
        }
        match &self.key {
            Json::String(text) => text.clone(),
            other => other.to_string(),
        }
    }
}

/// What a `stats` aggregation returned.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct StatsResult {
    pub count: u64,
    /// `None` when nothing was aggregated — not zero.
    pub min: Option<f64>,
    pub max: Option<f64>,
    pub avg: Option<f64>,
    pub sum: f64,
}

fn string_at(body: &Json, path: &str) -> String {
    body.get(path).and_then(Json::as_str).unwrap_or_default().to_string()
}

fn u64_at(body: &Json, path: &str) -> u64 {
    body.get(path).and_then(Json::as_i64).unwrap_or(0).max(0) as u64
}

fn u32_at(body: &Json, path: &str) -> u32 {
    u64_at(body, path) as u32
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Copied from a running Elasticsearch 8.15, not written from memory.
    fn parse(body: &str) -> SearchResults {
        SearchResults::parse(&Json::parse(body).expect("valid JSON in a test")).unwrap()
    }

    #[test]
    fn reads_the_hits_a_real_search_returns() {
        let results = parse(
            r#"{"took":4,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":2,"relation":"eq"},"max_score":0.60706794,"hits":[{"_index":"posts","_id":"1","_score":0.60706794,"_source":{"title":"Rust web frameworks","tags":["rust","web"],"views":120}},{"_index":"posts","_id":"2","_score":0.35667494,"_source":{"title":"Rust and Tokio","tags":["rust"],"views":45}}]}}"#,
        );

        assert_eq!(results.took, 4);
        assert_eq!(results.total.value, 2);
        assert!(results.total.is_exact());
        assert_eq!(results.max_score, Some(0.60706794));
        assert_eq!(results.ids(), vec!["1", "2"]);
        assert_eq!(results.hits[0].string("title"), Some("Rust web frameworks"));
        assert_eq!(results.hits[0].field("views").and_then(Json::as_i64), Some(120));
        assert!(!results.is_partial());
    }

    #[test]
    fn a_capped_total_is_never_reported_as_the_answer() {
        // The trap: `value` is 10000 and the real number is far larger. This is
        // the default for every index, so it is the common case rather than an
        // edge one.
        let results = parse(
            r#"{"took":12,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":10000,"relation":"gte"},"max_score":1.0,"hits":[]}}"#,
        );

        assert_eq!(results.total.relation, TotalRelation::AtLeast);
        assert!(!results.total.is_exact());
        assert_eq!(results.total.exact(), None, "a floor must not be handed out as a count");
        assert_eq!(results.total.to_string(), "at least 10000");
    }

    #[test]
    fn an_exact_total_reads_as_a_plain_number() {
        let results = parse(
            r#"{"took":1,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":7,"relation":"eq"},"max_score":null,"hits":[]}}"#,
        );

        assert_eq!(results.total.exact(), Some(7));
        assert_eq!(results.total.to_string(), "7");
        assert_eq!(results.max_score, None, "a sorted search scores nothing");
    }

    #[test]
    fn an_old_style_integer_total_is_still_read() {
        // OpenSearch with `rest_total_hits_as_int`, and Elasticsearch 6.
        let results = parse(
            r#"{"took":1,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":42,"max_score":1.0,"hits":[]}}"#,
        );

        assert_eq!(results.total.exact(), Some(42));
    }

    #[test]
    fn a_search_that_succeeded_on_only_some_shards_says_so() {
        // Status 200, hits present, and part of the index was never read. A
        // caller acting on absence has to know.
        let results = parse(
            r#"{"took":8,"timed_out":false,"_shards":{"total":2,"successful":1,"skipped":0,"failed":1,"failures":[{"shard":0,"index":"posts","node":"abc","reason":{"type":"query_shard_exception","reason":"failed to create query"}}]},"hits":{"total":{"value":1,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"posts","_id":"1","_score":1.0,"_source":{}}]}}"#,
        );

        assert!(results.is_partial(), "one shard failed and the results are incomplete");
        assert_eq!(results.shards.failed, 1);
        assert_eq!(results.shards.failures, vec!["failed to create query".to_string()]);
    }

    #[test]
    fn a_timed_out_search_is_partial_too() {
        let results = parse(
            r#"{"took":500,"timed_out":true,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":3,"relation":"eq"},"max_score":1.0,"hits":[]}}"#,
        );

        assert!(results.is_partial());
    }

    #[test]
    fn reads_a_terms_aggregation_and_the_stats_nested_inside_it() {
        let results = parse(
            r#"{"took":6,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":3,"relation":"eq"},"max_score":null,"hits":[]},"aggregations":{"by_tag":{"doc_count_error_upper_bound":0,"sum_other_doc_count":0,"buckets":[{"key":"rust","doc_count":2,"views":{"count":2,"min":45.0,"max":120.0,"avg":82.5,"sum":165.0}},{"key":"web","doc_count":1,"views":{"count":1,"min":120.0,"max":120.0,"avg":120.0,"sum":120.0}}]}}}"#,
        );

        let tags = results.aggregations.terms("by_tag").unwrap();
        assert_eq!(tags.buckets.len(), 2);
        assert_eq!(tags.buckets[0].key_string(), "rust");
        assert_eq!(tags.count("rust"), 2);
        assert_eq!(tags.count("nothing"), 0);
        assert!(!tags.is_approximate());

        let views = tags.buckets[0].aggregations.stats("views").unwrap();
        assert_eq!(views.count, 2);
        assert_eq!(views.avg, Some(82.5));
        assert_eq!(views.sum, 165.0);
    }

    #[test]
    fn a_truncated_terms_aggregation_admits_that_it_is_truncated() {
        // Ten buckets and 4,000 documents in the ones that did not fit. A chart
        // drawn from this without checking is wrong and looks fine.
        let results = parse(
            r#"{"took":9,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":5000,"relation":"eq"},"max_score":null,"hits":[]},"aggregations":{"by_tag":{"doc_count_error_upper_bound":12,"sum_other_doc_count":4000,"buckets":[{"key":"rust","doc_count":1000}]}}}"#,
        );

        let tags = results.aggregations.terms("by_tag").unwrap();
        assert!(tags.is_approximate());
        assert_eq!(tags.sum_other_doc_count, 4000);
    }

    #[test]
    fn stats_over_nothing_are_null_rather_than_zero() {
        let results = parse(
            r#"{"took":2,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":0,"relation":"eq"},"max_score":null,"hits":[]},"aggregations":{"price":{"count":0,"min":null,"max":null,"avg":null,"sum":0.0}}}"#,
        );

        let price = results.aggregations.stats("price").unwrap();
        assert_eq!(price.count, 0);
        assert_eq!(price.min, None, "there is no minimum of nothing");
        assert_eq!(price.avg, None);
        assert_eq!(price.sum, 0.0);
    }

    #[test]
    fn a_single_value_metric_is_read_by_name() {
        let results = parse(
            r#"{"took":3,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":3,"relation":"eq"},"max_score":null,"hits":[]},"aggregations":{"authors":{"value":2},"total_views":{"value":285.0}}}"#,
        );

        assert_eq!(results.aggregations.value("authors"), Some(2.0));
        assert_eq!(results.aggregations.value("total_views"), Some(285.0));
        assert_eq!(results.aggregations.value("missing"), None);
        assert_eq!(results.aggregations.names(), vec!["authors", "total_views"]);
    }

    #[test]
    fn a_date_histogram_bucket_reads_as_its_date_and_not_as_epoch_millis() {
        let results = parse(
            r#"{"took":5,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":2,"relation":"eq"},"max_score":null,"hits":[]},"aggregations":{"per_day":{"buckets":[{"key_as_string":"2024-05-01T00:00:00.000Z","key":1714521600000,"doc_count":2}]}}}"#,
        );

        let days = results.aggregations.terms("per_day").unwrap();
        assert_eq!(days.buckets[0].key_string(), "2024-05-01T00:00:00.000Z");
        assert_eq!(days.buckets[0].doc_count, 2);
    }

    #[test]
    fn sort_values_come_back_for_paging_past_the_result_window() {
        let results = parse(
            r#"{"took":2,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":2,"relation":"eq"},"max_score":null,"hits":[{"_index":"posts","_id":"2","_score":null,"_source":{"title":"b"},"sort":[1714521600000,"2"]}]}}"#,
        );

        assert_eq!(results.hits[0].sort.len(), 2);
        assert_eq!(results.hits[0].sort[1].as_str(), Some("2"));
    }

    #[test]
    fn a_reply_with_no_hits_object_is_a_shape_we_do_not_understand() {
        let error = SearchResults::parse(&Json::parse(r#"{"took":1}"#).unwrap()).unwrap_err();
        assert!(matches!(error, SearchError::Malformed { .. }), "got {error:?}");
    }
}