rustlavel_search/
index.rs

1//! Creating indices and describing what goes in them.
2//!
3//! An index can be created by writing a document to it, which is why so many
4//! projects never write a mapping at all — and then discover that every number
5//! is a `long` because the first document happened to have no decimals, that a
6//! title cannot be sorted or aggregated because it was analysed, and that a
7//! misspelled field name was accepted in silence and is now a permanent column
8//! in the mapping. A mapping cannot be changed after the fact; the index has to
9//! be rebuilt. So writing one first is not ceremony.
10//!
11//! ```ignore
12//! client
13//!     .create_index(
14//!         "posts",
15//!         &IndexDefinition::new()
16//!             .field("title", Field::text().with_keyword())
17//!             .field("tags", Field::keyword())
18//!             .field("views", Field::long())
19//!             .field("published_at", Field::date())
20//!             .dynamic_strict()
21//!             .shards(1)
22//!             .replicas(0),
23//!     )
24//!     .await?;
25//! ```
26
27use crate::client::{SearchClient, encode};
28use crate::error::{Result, SearchError};
29use rustlavel_core::Json;
30use rustlavel_http::Method;
31use std::collections::BTreeMap;
32
33/// How one field is stored and searched.
34#[derive(Debug, Clone, PartialEq)]
35pub struct Field {
36    options: BTreeMap<String, Json>,
37    sub_fields: BTreeMap<String, Field>,
38    properties: BTreeMap<String, Field>,
39}
40
41impl Field {
42    /// Analysed free text: broken into terms, searchable with
43    /// [`crate::Query::matching`].
44    ///
45    /// A `text` field cannot be sorted, aggregated on, or matched exactly —
46    /// the original string is not kept as a whole. [`Field::with_keyword`] is
47    /// how you get both, and is what dynamic mapping does for a string it
48    /// discovers on its own.
49    pub fn text() -> Field {
50        Field::of("text")
51    }
52
53    /// An exact string: an identifier, a status, a tag.
54    ///
55    /// Not analysed, so it sorts, aggregates and filters — and matches only
56    /// the whole value, case included.
57    pub fn keyword() -> Field {
58        Field::of("keyword")
59    }
60
61    pub fn long() -> Field {
62        Field::of("long")
63    }
64
65    pub fn integer() -> Field {
66        Field::of("integer")
67    }
68
69    pub fn double() -> Field {
70        Field::of("double")
71    }
72
73    pub fn float() -> Field {
74        Field::of("float")
75    }
76
77    pub fn boolean() -> Field {
78        Field::of("boolean")
79    }
80
81    /// A date. Accepts ISO 8601 and epoch milliseconds unless
82    /// [`Field::format`] says otherwise.
83    pub fn date() -> Field {
84        Field::of("date")
85    }
86
87    pub fn ip() -> Field {
88        Field::of("ip")
89    }
90
91    pub fn geo_point() -> Field {
92        Field::of("geo_point")
93    }
94
95    /// A nested structure, flattened into the parent document.
96    ///
97    /// An array of objects mapped this way loses the association between its
98    /// fields — `{"a":1,"b":2}` and `{"a":2,"b":1}` become `a:[1,2] b:[1,2]`
99    /// and a query for `a=1 and b=1` matches. [`Field::nested`] is what keeps
100    /// each object whole, at the cost of a hidden document per entry.
101    pub fn object() -> Field {
102        Field::of("object")
103    }
104
105    /// A structure whose objects stay separately queryable.
106    pub fn nested() -> Field {
107        Field::of("nested")
108    }
109
110    /// A type this module does not name.
111    pub fn of(kind: &str) -> Field {
112        Field {
113            options: BTreeMap::from([("type".to_string(), Json::String(kind.to_string()))]),
114            sub_fields: BTreeMap::new(),
115            properties: BTreeMap::new(),
116        }
117    }
118
119    /// Index the whole value as a `keyword` alongside the analysed form, under
120    /// `<field>.keyword`.
121    ///
122    /// The single most useful line in a mapping: without it, aggregating or
123    /// sorting on a `text` field fails with an error suggesting you enable
124    /// fielddata, which loads every term into heap and is almost never what
125    /// anybody wants. `ignore_above` matches what dynamic mapping uses — a
126    /// value longer than 256 characters is not worth a sort key.
127    pub fn with_keyword(mut self) -> Field {
128        self.sub_fields.insert(
129            "keyword".to_string(),
130            Field::keyword().option("ignore_above", Json::from(256)),
131        );
132        self
133    }
134
135    /// Which analyser to break the text with — `"english"` for stemming and
136    /// stop words, a custom one defined in the index settings.
137    pub fn analyzer(self, analyzer: &str) -> Field {
138        self.option("analyzer", Json::String(analyzer.to_string()))
139    }
140
141    /// The date formats this field accepts, `||`-separated.
142    pub fn format(self, format: &str) -> Field {
143        self.option("format", Json::String(format.to_string()))
144    }
145
146    /// Store the field but never search it, which saves the index for
147    /// something only ever read back with the document.
148    pub fn not_indexed(self) -> Field {
149        self.option("index", Json::Bool(false))
150    }
151
152    /// Accept a value that does not parse instead of rejecting the document.
153    ///
154    /// The field is left out of the index for that document and stays in
155    /// `_source`. Worth considering for ingest of data you do not control,
156    /// where one bad row should not fail a whole bulk request — and worth
157    /// avoiding elsewhere, because the value becomes invisible to search with
158    /// nothing to say so.
159    pub fn ignore_malformed(self) -> Field {
160        self.option("ignore_malformed", Json::Bool(true))
161    }
162
163    /// A field of an `object` or `nested` field.
164    pub fn field(mut self, name: impl Into<String>, field: Field) -> Field {
165        self.properties.insert(name.into(), field);
166        self
167    }
168
169    /// Another sub-field beside the main one.
170    pub fn sub_field(mut self, name: impl Into<String>, field: Field) -> Field {
171        self.sub_fields.insert(name.into(), field);
172        self
173    }
174
175    /// Any other mapping parameter.
176    pub fn option(mut self, name: impl Into<String>, value: Json) -> Field {
177        self.options.insert(name.into(), value);
178        self
179    }
180
181    /// The JSON this field maps to.
182    pub fn json(&self) -> Json {
183        let mut body = self.options.clone();
184
185        if !self.sub_fields.is_empty() {
186            body.insert("fields".to_string(), properties_of(&self.sub_fields));
187        }
188        if !self.properties.is_empty() {
189            body.insert("properties".to_string(), properties_of(&self.properties));
190        }
191
192        Json::Object(body)
193    }
194}
195
196/// An index about to be created.
197#[derive(Debug, Clone, Default, PartialEq)]
198pub struct IndexDefinition {
199    fields: BTreeMap<String, Field>,
200    settings: BTreeMap<String, Json>,
201    dynamic: Option<Json>,
202    aliases: Vec<String>,
203}
204
205impl IndexDefinition {
206    pub fn new() -> IndexDefinition {
207        IndexDefinition::default()
208    }
209
210    pub fn field(mut self, name: impl Into<String>, field: Field) -> IndexDefinition {
211        self.fields.insert(name.into(), field);
212        self
213    }
214
215    /// How many primary shards. Fixed for the life of the index.
216    pub fn shards(self, shards: u32) -> IndexDefinition {
217        self.setting("number_of_shards", Json::from(shards))
218    }
219
220    /// How many copies of each shard.
221    ///
222    /// Zero is right for a test or a single-node cluster and wrong everywhere
223    /// else: with no replica, one lost node is lost data. The default is one,
224    /// which is also why a single-node cluster reports yellow.
225    pub fn replicas(self, replicas: u32) -> IndexDefinition {
226        self.setting("number_of_replicas", Json::from(replicas))
227    }
228
229    /// How often new documents become searchable, as a duration like `"30s"`,
230    /// or `"-1"` to refresh only on demand while bulk-loading.
231    pub fn refresh_interval(self, interval: &str) -> IndexDefinition {
232        self.setting("refresh_interval", Json::String(interval.to_string()))
233    }
234
235    pub fn setting(mut self, name: impl Into<String>, value: Json) -> IndexDefinition {
236        self.settings.insert(name.into(), value);
237        self
238    }
239
240    /// Reject a document containing a field the mapping does not declare.
241    ///
242    /// The default is to accept it and add it to the mapping permanently, so a
243    /// typo in one document becomes a field in the index for good, and the
244    /// misspelled data is invisible to every query written against the correct
245    /// name. Strict turns that into an error at write time, where it can be
246    /// fixed.
247    pub fn dynamic_strict(mut self) -> IndexDefinition {
248        self.dynamic = Some(Json::String("strict".to_string()));
249        self
250    }
251
252    /// Accept unmapped fields but do not index them: they stay readable in
253    /// `_source` and are not searchable.
254    pub fn dynamic_false(mut self) -> IndexDefinition {
255        self.dynamic = Some(Json::Bool(false));
256        self
257    }
258
259    /// Another name this index answers to.
260    pub fn alias(mut self, alias: impl Into<String>) -> IndexDefinition {
261        self.aliases.push(alias.into());
262        self
263    }
264
265    /// The mappings half on its own, which is what `_mapping` takes.
266    pub fn mappings(&self) -> Json {
267        let mut body = BTreeMap::new();
268
269        if let Some(dynamic) = &self.dynamic {
270            body.insert("dynamic".to_string(), dynamic.clone());
271        }
272        if !self.fields.is_empty() {
273            body.insert("properties".to_string(), properties_of(&self.fields));
274        }
275
276        Json::Object(body)
277    }
278
279    /// The whole create-index body.
280    pub fn body(&self) -> Json {
281        let mut body = BTreeMap::new();
282
283        if !self.settings.is_empty() {
284            body.insert("settings".to_string(), Json::Object(self.settings.clone()));
285        }
286
287        let mappings = self.mappings();
288        if !mappings.as_object().map(BTreeMap::is_empty).unwrap_or(true) {
289            body.insert("mappings".to_string(), mappings);
290        }
291
292        if !self.aliases.is_empty() {
293            let aliases = self
294                .aliases
295                .iter()
296                .map(|name| (name.clone(), Json::Object(BTreeMap::new())))
297                .collect();
298            body.insert("aliases".to_string(), Json::Object(aliases));
299        }
300
301        Json::Object(body)
302    }
303}
304
305fn properties_of(fields: &BTreeMap<String, Field>) -> Json {
306    Json::Object(fields.iter().map(|(name, field)| (name.clone(), field.json())).collect())
307}
308
309impl SearchClient {
310    /// Create an index.
311    ///
312    /// Fails with [`SearchError::IndexAlreadyExists`] if it is already there.
313    pub async fn create_index(&self, name: &str, definition: &IndexDefinition) -> Result<()> {
314        let path = encode(name);
315        self.json(Method::Put, &path, name, Some(definition.body())).await?;
316        Ok(())
317    }
318
319    /// Create an index unless it exists, answering whether it was created.
320    ///
321    /// Implemented by attempting the create and forgiving the one failure that
322    /// means "already there" — not by checking first. Two application
323    /// instances booting together both pass a check and one still loses the
324    /// create, so the check would turn a harmless race into an intermittent
325    /// crash at startup.
326    pub async fn create_index_if_missing(
327        &self,
328        name: &str,
329        definition: &IndexDefinition,
330    ) -> Result<bool> {
331        match self.create_index(name, definition).await {
332            Ok(()) => Ok(true),
333            Err(SearchError::IndexAlreadyExists { .. }) => Ok(false),
334            Err(error) => Err(error),
335        }
336    }
337
338    /// Delete an index and everything in it, answering whether there was one.
339    pub async fn delete_index(&self, name: &str) -> Result<bool> {
340        let path = encode(name);
341        match self.json(Method::Delete, &path, name, None).await {
342            Ok(_) => Ok(true),
343            Err(SearchError::IndexNotFound { .. }) => Ok(false),
344            Err(error) => Err(error),
345        }
346    }
347
348    /// Whether the index — or an alias by that name — exists.
349    ///
350    /// `filter_path` matches nothing on purpose: the whole reply would be the
351    /// index's settings and mappings, and the only part being asked about is
352    /// the status code.
353    pub async fn index_exists(&self, name: &str) -> Result<bool> {
354        self.exists_at(&format!("{}?filter_path=none", encode(name)), name).await
355    }
356
357    /// Add fields to an existing index's mapping.
358    ///
359    /// Adding is all this can do. Changing the type of a field that already
360    /// exists is refused, because the data is already indexed the old way;
361    /// that needs a new index and a reindex.
362    pub async fn put_mapping(&self, name: &str, definition: &IndexDefinition) -> Result<()> {
363        let path = format!("{}/_mapping", encode(name));
364        self.json(Method::Put, &path, name, Some(definition.mappings())).await?;
365        Ok(())
366    }
367
368    /// The mapping as the cluster holds it, including whatever dynamic mapping
369    /// added along the way.
370    pub async fn get_mapping(&self, name: &str) -> Result<Json> {
371        let path = format!("{}/_mapping", encode(name));
372        self.json(Method::Get, &path, name, None).await
373    }
374
375    /// Make everything written so far searchable, now.
376    ///
377    /// Elasticsearch is near-real-time, not real-time: an indexed document is
378    /// durable immediately and *searchable* only after the next refresh, which
379    /// is one second away by default. That gap is the single most common cause
380    /// of a test that passes alone and fails in a suite, and of a "save then
381    /// redirect to the list" that shows the old list.
382    ///
383    /// Call it after writing and before searching in a test, and after a bulk
384    /// load. Do not call it after every write in production — each refresh
385    /// makes a new segment, and doing it per document is how an index ends up
386    /// spending its day merging.
387    pub async fn refresh(&self, name: &str) -> Result<()> {
388        let path = format!("{}/_refresh", encode(name));
389        self.json(Method::Post, &path, name, None).await?;
390        Ok(())
391    }
392}
393
394#[cfg(test)]
395mod tests {
396    use super::*;
397    use rustlavel_client::{Fake, FakeResponse};
398
399    fn json(body: &str) -> Json {
400        Json::parse(body).expect("valid JSON in a test")
401    }
402
403    #[test]
404    fn a_definition_sends_only_the_sections_that_have_something_in_them() {
405        assert_eq!(IndexDefinition::new().body().to_string(), "{}");
406
407        let definition = IndexDefinition::new().field("title", Field::text());
408        assert_eq!(
409            definition.body().to_string(),
410            r#"{"mappings":{"properties":{"title":{"type":"text"}}}}"#
411        );
412    }
413
414    #[test]
415    fn a_text_field_with_a_keyword_gets_the_sub_field_aggregations_need() {
416        // Without this, aggregating on `title` fails and the error suggests
417        // fielddata, which is the wrong fix.
418        let definition = IndexDefinition::new().field("title", Field::text().with_keyword());
419
420        assert_eq!(
421            definition.mappings().to_string(),
422            r#"{"properties":{"title":{"fields":{"keyword":{"ignore_above":256,"type":"keyword"}},"type":"text"}}}"#
423        );
424    }
425
426    #[test]
427    fn settings_mappings_and_aliases_all_land_in_the_create_body() {
428        let definition = IndexDefinition::new()
429            .shards(1)
430            .replicas(0)
431            .refresh_interval("30s")
432            .dynamic_strict()
433            .alias("posts-current")
434            .field("views", Field::long());
435
436        assert_eq!(
437            definition.body().to_string(),
438            r#"{"aliases":{"posts-current":{}},"mappings":{"dynamic":"strict","properties":{"views":{"type":"long"}}},"settings":{"number_of_replicas":0,"number_of_shards":1,"refresh_interval":"30s"}}"#
439        );
440    }
441
442    #[test]
443    fn a_nested_field_carries_its_own_properties() {
444        let field = Field::nested()
445            .field("street", Field::text())
446            .field("postcode", Field::keyword());
447
448        assert_eq!(
449            field.json().to_string(),
450            r#"{"properties":{"postcode":{"type":"keyword"},"street":{"type":"text"}},"type":"nested"}"#
451        );
452    }
453
454    #[test]
455    fn field_options_reach_the_mapping() {
456        assert_eq!(
457            Field::text().analyzer("english").json().to_string(),
458            r#"{"analyzer":"english","type":"text"}"#
459        );
460        assert_eq!(
461            Field::date().format("yyyy-MM-dd").json().to_string(),
462            r#"{"format":"yyyy-MM-dd","type":"date"}"#
463        );
464        assert_eq!(
465            Field::keyword().not_indexed().json().to_string(),
466            r#"{"index":false,"type":"keyword"}"#
467        );
468        assert_eq!(
469            Field::long().ignore_malformed().json().to_string(),
470            r#"{"ignore_malformed":true,"type":"long"}"#
471        );
472    }
473
474    #[tokio::test]
475    async fn creating_an_index_puts_the_body_at_the_index_url() {
476        let client = SearchClient::new("http://localhost:9200").faking(
477            Fake::new().fallback(FakeResponse::json(json(
478                r#"{"acknowledged":true,"shards_acknowledged":true,"index":"posts"}"#,
479            ))),
480        );
481
482        client
483            .create_index("posts", &IndexDefinition::new().field("title", Field::text()))
484            .await
485            .unwrap();
486
487        let sent = &client.fake().unwrap().recorded()[0];
488        assert_eq!(sent.method, Method::Put);
489        assert_eq!(sent.url, "http://localhost:9200/posts");
490        assert!(sent.body_text().contains(r#""title":{"type":"text"}"#));
491    }
492
493    #[tokio::test]
494    async fn creating_an_index_that_exists_is_forgiven_only_when_it_was_asked_to_be() {
495        let body = r#"{"error":{"root_cause":[{"type":"resource_already_exists_exception","reason":"index [posts/9jNXbCyMS4WTRDMCsQiSuQ] already exists","index_uuid":"9jNXbCyMS4WTRDMCsQiSuQ","index":"posts"}],"type":"resource_already_exists_exception","reason":"index [posts/9jNXbCyMS4WTRDMCsQiSuQ] already exists","index_uuid":"9jNXbCyMS4WTRDMCsQiSuQ","index":"posts"},"status":400}"#;
496        let definition = IndexDefinition::new();
497
498        let client = SearchClient::new("http://localhost:9200")
499            .faking(Fake::new().fallback(FakeResponse::text(body).status(400)));
500        assert!(matches!(
501            client.create_index("posts", &definition).await,
502            Err(SearchError::IndexAlreadyExists { .. })
503        ));
504
505        let client = SearchClient::new("http://localhost:9200")
506            .faking(Fake::new().fallback(FakeResponse::text(body).status(400)));
507        assert!(
508            !client.create_index_if_missing("posts", &definition).await.unwrap(),
509            "it existed, so nothing was created"
510        );
511    }
512
513    #[tokio::test]
514    async fn deleting_an_index_that_is_not_there_is_not_an_error() {
515        let body = r#"{"error":{"root_cause":[{"type":"index_not_found_exception","reason":"no such index [gone]","resource.type":"index_or_alias","resource.id":"gone","index_uuid":"_na_","index":"gone"}],"type":"index_not_found_exception","reason":"no such index [gone]","resource.type":"index_or_alias","resource.id":"gone","index_uuid":"_na_","index":"gone"},"status":404}"#;
516        let client = SearchClient::new("http://localhost:9200")
517            .faking(Fake::new().fallback(FakeResponse::text(body).status(404)));
518
519        assert!(!client.delete_index("gone").await.unwrap());
520    }
521
522    #[tokio::test]
523    async fn existence_asks_for_as_little_as_possible_and_reads_a_404_as_an_answer() {
524        // Deliberately a GET: a real cluster answers a HEAD with a chunked
525        // framing it then never terminates. See `SearchClient::exists_at`.
526        let present = SearchClient::new("http://localhost:9200")
527            .faking(Fake::new().fallback(FakeResponse::json(json("{}"))));
528        assert!(present.index_exists("posts").await.unwrap());
529
530        let sent = &present.fake().unwrap().recorded()[0];
531        assert_eq!(sent.method, Method::Get);
532        assert_eq!(sent.url, "http://localhost:9200/posts?filter_path=none");
533
534        let absent = SearchClient::new("http://localhost:9200")
535            .faking(Fake::new().fallback(FakeResponse::text("").status(404)));
536        assert!(!absent.index_exists("posts").await.unwrap());
537    }
538
539    #[tokio::test]
540    async fn a_mapping_update_sends_only_the_mappings_half() {
541        let client = SearchClient::new("http://localhost:9200")
542            .faking(Fake::new().fallback(FakeResponse::json(json(r#"{"acknowledged":true}"#))));
543
544        // The settings are ignored rather than sent: `_mapping` answers 400 for
545        // a body containing them.
546        client
547            .put_mapping(
548                "posts",
549                &IndexDefinition::new().shards(3).field("summary", Field::text()),
550            )
551            .await
552            .unwrap();
553
554        let sent = &client.fake().unwrap().recorded()[0];
555        assert_eq!(sent.url, "http://localhost:9200/posts/_mapping");
556        assert_eq!(sent.body_text(), r#"{"properties":{"summary":{"type":"text"}}}"#);
557    }
558
559    #[tokio::test]
560    async fn refreshing_posts_to_the_refresh_endpoint() {
561        let client = SearchClient::new("http://localhost:9200").faking(Fake::new().fallback(
562            FakeResponse::json(json(r#"{"_shards":{"total":2,"successful":1,"failed":0}}"#)),
563        ));
564
565        client.refresh("posts").await.unwrap();
566
567        let sent = &client.fake().unwrap().recorded()[0];
568        assert_eq!(sent.method, Method::Post);
569        assert_eq!(sent.url, "http://localhost:9200/posts/_refresh");
570    }
571}
rustlavel_search/index.rs

rustlavel_search/
index.rs
