Lesson 27 of 28 · Databases

Search with Elasticsearch

Elasticsearch is a search engine and document store: you give it JSON documents and it finds the ones that match a query, ranks them and counts them in any way you ask. This lesson builds a small library catalogue with the db.elasticsearch package, from the first connection to bulk loading and SQL.

In this lesson

A cluster to talk to
Connecting
Storing and reading documents
Building queries
Searching and reading the answer
Aggregations
Loading many documents
Paging through everything
SQL
Errors and retries
Where to go next

A cluster to talk to

The package speaks to Elasticsearch 7 and 8, to Elastic Cloud and to OpenSearch, over plain HTTP or TLS, with no outside library. For trying things out, one node in Docker is enough:

docker run -d -p 127.0.0.1:9200:9200 -e discovery.type=single-node -e xpack.security.enabled=false docker.elastic.co/elasticsearch/elasticsearch:8.15.3

That node answers on http://localhost:9200 without a password. The examples that need a running cluster are marked as only checked; the ones that build JSON or read answers run here, with their real output.

Connecting

elasticsearch.Connect(url) is the short form. For anything more, fill an elasticsearch.Config and pass it to NewClient: several nodes, a user and password, an API key, a bearer token, the CA file that signed the cluster certificate, a timeout and the number of retries. A node URL may also carry the credentials itself, as in https://user:pass@host:9200.

es-connect.salam

import db.elasticsearch

func main:
    mut cfg := elasticsearch.DefaultConfig()
    cfg.nodes = Vector {} as Vector<str>
    cfg.nodes.push("https://es-1.example.com:9200")
    cfg.nodes.push("https://es-2.example.com:9200")
    cfg.username = "elastic"
    cfg.password = "s3cret"
    cfg.ca_file = "/etc/elasticsearch/certs/http_ca.crt"
    cfg.timeout_ms = 10000
    mut c := elasticsearch.NewClient(cfg)
    if not elasticsearch.Ping(c):
        println "no answer:", elasticsearch.Error(c)
        ret
    end
    info := elasticsearch.GetInfo(c)
    println info.distribution, info.version, elasticsearch.ClusterHealth(c).status
    elasticsearch.Close(c)
end

Creating a client does not touch the network. Ping asks whether the cluster answers, GetInfo returns its name, version and distribution, and ClusterHealth its green, yellow or red status. When you are done, Close frees the client and the connections it keeps open.

You have	Use
A local node, no security	`Connect("http://localhost:9200")`
A user and password	`ConnectAuth(url, user, password)`
An API key	`ConnectApiKey(url, key)`, encoded or as `id:key`
An Elastic Cloud deployment	`ConnectCloud(cloud_id, api_key)`
A service account token	`Config` with `bearer`
A self-signed cluster certificate	`Config` with `ca_file`

Storing and reading documents

A document is a JSON object stored in an index under an id. Documents and query clauses are plain str values holding JSON, so you can write them by hand, build them with encoding.json or use the builders of this package. CreateIndex sets up an index with its mappings; Index stores a document; Get reads it back; Update merges fields into it and Delete removes it.

es-documents.salam

import db.elasticsearch

func main:
    mut c := elasticsearch.Connect("http://localhost:9200")
    mapping := `{"mappings":{"properties":{"title":{"type":"text"},"author":{"type":"keyword"},"year":{"type":"integer"}}}}`
    if not elasticsearch.IndexExists(c, "books"):
        elasticsearch.CreateIndex(c, "books", mapping)
    end

    w := elasticsearch.IndexWith(c, "books", "1", `{"title":"Shahnameh","author":"ferdowsi","year":1010}`, elasticsearch.Options { refresh = elasticsearch.REFRESH_WAIT_FOR })
    println w.ok, w.result, w.version

    d := elasticsearch.Get(c, "books", "1")
    if d.found:
        println d.source
    end

    u := elasticsearch.UpdateWith(c, "books", "1", `{"year":1011}`, elasticsearch.Options { if_seq_no = d.seq_no, if_primary_term = d.primary_term })
    if not u.ok:
        println "someone else changed it first:", elasticsearch.ErrorType(c)
    end

    println elasticsearch.Exists(c, "books", "1"), elasticsearch.Count(c, "books", "")
    elasticsearch.Delete(c, "books", "1")
    elasticsearch.Close(c)
end

Every write returns a WriteResult with ok, the result (created, updated, deleted), the new version and the sequence numbers. Passing if_seq_no and if_primary_term from a Get turns the update into a compare and swap: if someone changed the document in between, the update fails with a version conflict instead of overwriting their change.

A new document is searchable only after the index refreshes, about once a second. Options { refresh = elasticsearch.REFRESH_WAIT_FOR } makes the call wait for that, which is handy in tests and scripts.

Building queries

Each query builder returns one JSON clause: Match for full text, Term and Terms for exact values, Range for numbers and dates, and And, Or, Not and Filtered to combine them. A SearchRequest holds the query with the size, sort order and the fields to return, and SearchBody shows the JSON that will be sent.

es-queries.salam

import db.elasticsearch

func main:
    mut must := Vector {} as Vector<str>
    must.push(elasticsearch.Match("title", "book of kings"))
    mut filter := Vector {} as Vector<str>
    filter.push(elasticsearch.Range("year", "900", "1100"))
    filter.push(elasticsearch.Not(elasticsearch.Term("status", "draft")))
    query := elasticsearch.Filtered(elasticsearch.And(must), filter)
    println query

    mut sort := Vector {} as Vector<str>
    sort.push(elasticsearch.SortDesc("year"))
    mut shown := Vector {} as Vector<str>
    shown.push("title")
    shown.push("year")
    req := elasticsearch.SearchRequest { query = query, size = 10, sort = sort, source = shown }
    println elasticsearch.SearchBody(req)
end

Output{"bool": {"must": {"bool": {"must": [{"match": {"title": "book of kings"}}]}}, "filter": [{"range": {"year": {"gte": "900", "lte": "1100"}}}, {"bool": {"must_not": [{"term": {"status": "draft"}}]}}]}} {"query": {"bool": {"must": {"bool": {"must": [{"match": {"title": "book of kings"}}]}}, "filter": [{"range": {"year": {"gte": "900", "lte": "1100"}}}, {"bool": {"must_not": [{"term": {"status": "draft"}}]}}]}}, "size": 10, "sort": [{"year": {"order": "desc"}}], "_source": ["title", "year"]}

Clauses inside Filtered's filter list only decide whether a document matches; they do not change its score, and Elasticsearch can cache them. Put free text in the query and exact conditions such as dates, statuses and ids in the filter.

Searching and reading the answer

Search(c, index, request) runs the request and parses the answer into a SearchResult: the total number of matches, the hits of this page, each with its id, score, source and highlight, and the time it took.

es-search.salam

import db.elasticsearch

func main:
    mut c := elasticsearch.Connect("http://localhost:9200")
    mut fields := Vector {} as Vector<str>
    fields.push("title")
    req := elasticsearch.SearchRequest { query = elasticsearch.Match("title", "kings"), size = 5, highlight = elasticsearch.Highlight(fields) }
    res := elasticsearch.Search(c, "books", req)
    if not res.ok:
        println "search failed:", elasticsearch.Error(c)
        ret
    end
    println res.total, "matches in", res.took, "ms"
    each hit in res.hits:
        println hit.id, hit.score, hit.source, hit.highlight
    end
    elasticsearch.Close(c)
end

The same parsing is available on its own: ParseSearchResult reads any search answer, for example one you fetched with Perform.

es-results.salam

import db.elasticsearch

func main:
    body := `{"took":3,"hits":{"total":{"value":2,"relation":"eq"},"max_score":1.8,"hits":[{"_id":"1","_score":1.8,"_source":{"title":"Shahnameh","year":1010}},{"_id":"7","_score":0.9,"_source":{"title":"Garshasp-nama","year":1066}}]}}`
    res := elasticsearch.ParseSearchResult(body)
    println res.total, res.total_relation, res.max_score
    each hit in res.hits:
        println hit.id, hit.score, hit.source
    end
end

Output2 eq 1.8 1 1.8 {"title":"Shahnameh","year":1010} 7 0.9 {"title":"Garshasp-nama","year":1066}

Aggregations

Aggregations summarise the matching documents instead of listing them: how many per author, the average year, the oldest book. Build them with AddAgg(aggs, name, agg), nest one inside a bucket with SubAggs, and read them back with AggValue for a single number and AggBuckets for groups. A bucket's raw JSON goes to AggValueIn to read what is nested inside it.

es-aggs.salam

import db.elasticsearch

func main:
    per_author := elasticsearch.SubAggs(elasticsearch.TermsAgg("author", 5), elasticsearch.AddAgg("", "avg_year", elasticsearch.AvgAgg("year")))
    mut aggs := elasticsearch.AddAgg("", "by_author", per_author)
    aggs = elasticsearch.AddAgg(aggs, "oldest", elasticsearch.MinAgg("year"))
    println elasticsearch.SearchBody(elasticsearch.SearchRequest { size = 0, aggs = aggs })

    answer := `{"hits":{"total":{"value":120,"relation":"eq"},"hits":[]},"aggregations":{"oldest":{"value":940.0},"by_author":{"buckets":[{"key":"ferdowsi","doc_count":70,"avg_year":{"value":1010.0}},{"key":"saadi","doc_count":50,"avg_year":{"value":1258.0}}]}}}`
    res := elasticsearch.ParseSearchResult(answer)
    println "oldest:", elasticsearch.AggValue(res, "oldest")
    each b in elasticsearch.AggBuckets(res, "by_author"):
        println b.key, b.doc_count, elasticsearch.AggValueIn(b.raw, "avg_year")
    end
end

Output{"size": 0, "aggs": {"by_author": {"terms": {"field": "author", "size": 5}, "aggs": {"avg_year": {"avg": {"field": "year"}}}}, "oldest": {"min": {"field": "year"}}}} oldest: 940 ferdowsi 70 1010 saadi 50 1258

Setting size = 0 skips the hits when only the summary matters, which makes the request much cheaper.

Loading many documents

Sending documents one by one costs a round trip each. A Bulk collects index, create, update and delete actions and sends them in one request. The answer says per item whether it worked, so a partial failure is never silent.

es-bulk.salam

import str
import db.elasticsearch

func main:
    mut b := elasticsearch.NewBulk()
    repeat 3 in i:
        id := str.FromInt(i as i64)
        elasticsearch.BulkIndex(b, "logs", id, `{"line":` + id + `,"level":"info"}`)
    end
    elasticsearch.BulkUpdate(b, "logs", "1", `{"level":"warn"}`)
    elasticsearch.BulkDelete(b, "logs", "2")
    println elasticsearch.BulkLen(b), "actions"
    print elasticsearch.BulkBody(b)

    answer := `{"took":5,"errors":true,"items":[{"index":{"_id":"0","status":201,"result":"created"}},{"index":{"_id":"1","status":201,"result":"created"}},{"index":{"_id":"2","status":201,"result":"created"}},{"update":{"_id":"1","status":200,"result":"updated"}},{"delete":{"_id":"2","status":429,"error":{"type":"es_rejected_execution_exception","reason":"queue is full"}}}]}`
    res := elasticsearch.ParseBulkResult(answer)
    println res.succeeded, "ok,", res.failed, "failed"
    each it in res.items:
        if not it.ok:
            println "retry", it.action, it.id, it.error_type
        end
    end
end

Output5 actions {"index": {"_index": "logs", "_id": "0"}} {"line":0,"level":"info"} {"index": {"_index": "logs", "_id": "1"}} {"line":1,"level":"info"} {"index": {"_index": "logs", "_id": "2"}} {"line":2,"level":"info"} {"update": {"_index": "logs", "_id": "1"}} {"doc": {"level":"warn"}} {"delete": {"_index": "logs", "_id": "2"}} 4 ok, 1 failed retry delete 2 es_rejected_execution_exception

For large loads, send a batch every thousand or so actions and start over with BulkReset. The client keeps the connection to each node open between calls, so over HTTPS the TLS handshake is paid once, not once per request.

es-bulk-run.salam

import str
import db.elasticsearch

func main:
    mut c := elasticsearch.Connect("http://localhost:9200")
    mut b := elasticsearch.NewBulk()
    repeat 10000 in i:
        id := str.FromInt(i as i64)
        elasticsearch.BulkIndex(b, "logs", id, `{"line":` + id + `}`)
        if elasticsearch.BulkLen(b) == 1000:
            res := elasticsearch.RunBulk(c, b)
            println res.succeeded, res.failed
            elasticsearch.BulkReset(b)
        end
    end
    if elasticsearch.BulkLen(b) > 0:
        res := elasticsearch.RunBulk(c, b)
        println res.succeeded, res.failed
    end
    elasticsearch.Refresh(c, "logs")
    println elasticsearch.Count(c, "logs", "")
    elasticsearch.Close(c)
end

Paging through everything

from and size are fine for the first few pages of a result list. To walk through every match, open a point in time with OpenPit, sort by the fields you need, and pass the whole sort array of the last hit as search_after to get the next page. A point-in-time search adds a unique _shard_doc value to the end of every hit's sort, so no two hits tie and you do not need to sort by _id. The point in time keeps the view stable while you read. SearchScroll and Scroll do the same in the older style.

es-paging.salam

import db.elasticsearch

func main:
    mut c := elasticsearch.Connect("http://localhost:9200")
    mut sort := Vector {} as Vector<str>
    sort.push(elasticsearch.SortAsc("year"))

    pit := elasticsearch.OpenPit(c, "books", "1m")
    mut after := ""
    mut more := true
    until more:
        page := elasticsearch.Search(c, "", elasticsearch.SearchRequest { size = 500, sort = sort, pit_id = pit, pit_keep_alive = "1m", search_after = after })
        if page.hits.len() == 0:
            more = false
        else:
            each hit in page.hits:
                println hit.id
            end
            after = page.hits.get(page.hits.len() - 1).sort
        end
    end
    elasticsearch.ClosePit(c, pit)
    elasticsearch.Close(c)
end

SQL

Elasticsearch, and the SQL plugin of OpenSearch, also answer SQL. Sql and SqlWith return rows that implement db.Rows and fetch further pages by themselves as you call Next. Conn(c) wraps the client as a db.Connection, so a function written for SQLite or PostgreSQL can read from Elasticsearch too. That connection is read only: writes go through Index, Bulk and DeleteByQuery.

es-sql.salam

import db
import db.elasticsearch

func authors(conn: dyn db.Connection):
    rows := conn.Query("SELECT author, COUNT(*) AS books FROM books GROUP BY author ORDER BY author")
    until rows.Next():
        println rows.Text(0), rows.Int(1)
    end
    rows.Close()
end

func main:
    mut c := elasticsearch.Connect("http://localhost:9200")
    mut params := Vector {} as Vector<str>
    params.push("1000")
    mut rows := elasticsearch.SqlWith(c, "SELECT title, year FROM books WHERE year > ? ORDER BY year", params, 100)
    until rows.Next():
        println rows.Text(0), rows.Int(1)
    end
    rows.Close()
    authors(elasticsearch.Conn(c))
    elasticsearch.Close(c)
end

Errors and retries

A call that fails returns its sentinel (false, -1, an empty string or a result with ok false), and the client remembers why: Error(c) gives the message, ErrorType(c) the Elasticsearch error type such as version_conflict_engine_exception, and Status(c) the HTTP status, 0 when no node answered. With several nodes the client rotates over them, skips a node that refused a connection for a minute, and retries on 502, 503 and 504.

es-errors.salam

import db.elasticsearch

func main:
    mut cfg := elasticsearch.Config { max_retries = 3, timeout_ms = 5000, keep_alive = true }
    cfg.nodes.push("http://es-1:9200")
    cfg.nodes.push("http://es-2:9200")
    mut c := elasticsearch.NewClient(cfg)
    w := elasticsearch.Create(c, "books", "1", `{"title":"Shahnameh"}`)
    if not w.ok:
        switch elasticsearch.ErrorType(c):
            "version_conflict_engine_exception":
                println "already there"
                break
            end
            "":
                println "no node answered:", elasticsearch.Error(c)
                break
            end
            else:
                println elasticsearch.Status(c), elasticsearch.Error(c)
            end
        end
    end
    r := elasticsearch.Perform(c, "GET", "/_cat/indices?format=json", "")
    println r.status, r.body
    elasticsearch.Close(c)
end

Anything the package has no function for is one Perform(c, method, path, body) away; it returns the status and the raw body.

Where to go next

Need	Reach for
Index settings, mappings, aliases and templates	`PutMapping`, `PutSettings`, `SwapAlias`, `PutIndexTemplate`
Copying an index	`Reindex`, or `StartReindex` and `GetTask` in the background
Cleaning documents on the way in	`PutPipeline` and `Options { pipeline = ... }`
Logs and other time series	`CreateDataStream`, `Rollover`, `PutLifecyclePolicy`
Backups	`PutRepository`, `CreateSnapshot`, `RestoreSnapshot`
Long searches	`SubmitAsyncSearch` and `GetAsyncSearch`
Vector search	`Knn` in `SearchRequest.knn`

Every function, type and field is described, with examples, on the db/elasticsearch reference page.