Search with OpenSearch
OpenSearch is a search engine: you give it documents as JSON, and it finds the ones that match a query, ranks them, and counts them by any field you like. This lesson builds a small book search with the db.opensearch package, one step at a time.
A server to talk to
Every example that talks to a cluster expects one at http://localhost:9200. The quickest way to get one is Docker: docker run -p 9200:9200 -e discovery.type=single-node -e DISABLE_SECURITY_PLUGIN=true opensearchproject/opensearch:2.17.1. There is no server behind this page, so those examples are checked by the compiler and show the output they printed when run, in order, against a real OpenSearch 2.17.1. The ones that only build requests or read answers are run every time the site is built.
The package is part of the standard library and needs nothing else installed. Import it as db.opensearch and call it as opensearch.
Connecting
Connect takes a node url and gives back a client. Nothing is sent until the first call, so check that the cluster answers with Ping, then ask it about itself with GetInfo and ClusterHealth.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
if not opensearch.Ping(c):
println "no answer:", opensearch.Error(c)
ret 1
end
info := opensearch.GetInfo(c)
println info.distribution, info.version, info.cluster_name
h := opensearch.ClusterHealth(c)
println "health:", h.status, h.nodes, "nodes"
ret 0
endA real deployment usually needs more: several nodes, a user and password, your own certificate authority, a time limit. Put them in a Config and pass it to NewClient. The client spreads requests over the nodes, skips one that refuses connections for a minute, and retries a request that got a 502, 503 or 504 on another node. ConnectInsecure is for the demo setup that ships with a self signed certificate, and ConnectAWS signs every request with AWS SigV4 for Amazon OpenSearch Service; use SERVICE_AOSS instead of SERVICE_ES for a Serverless collection.
import db.opensearch
func main:
mut cfg := opensearch.DefaultConfig()
cfg.nodes.push("https://node2.example.com:9200, https://node3.example.com:9200")
cfg.username = "app"
cfg.password = "s3cret-pass"
cfg.ca_file = "/etc/ssl/certs/opensearch-ca.pem"
cfg.timeout_ms = 10000
cfg.max_retries = 2
c := opensearch.NewClient(cfg)
println opensearch.Ok(c), opensearch.Nodes(c)
local := opensearch.ConnectInsecure("https://localhost:9200", "admin", "admin-pass")
println opensearch.Ok(local)
aws := opensearch.ConnectAWS(
"https://search-books.eu-west-1.es.amazonaws.com", "eu-west-1", opensearch.SERVICE_ES,
"AKIA...", "secret-key", ""
)
println opensearch.Ok(aws)
endAn index and its documents
An index is where documents live, and its mapping says what each field is: text is split into words for full text search, keyword is matched exactly, and numbers and dates can be compared and sorted. OpenSearch guesses a mapping if you give none, but a guess cannot be changed later, so it pays to create the index yourself. Property writes one field and Properties puts them together.
Documents are JSON text, and every write returns a WriteResult with the id, the result and the version. A new document is searchable after the next refresh, about a second later; REFRESH_WAIT_FOR makes the call wait for it, which is handy in tests and scripts. Get reads a document back by id, and JsonPath picks a value out of its source.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
mut props := Vector {} as Vector<str>
props.push(opensearch.Property("title", "text"))
props.push(opensearch.Property("author", "keyword"))
props.push(opensearch.Property("year", "integer"))
props.push(opensearch.Property("stock", "integer"))
println "created:", opensearch.CreateIndexWith(c, "books", 1, 0, opensearch.Properties(props))
w := opensearch.IndexWith(c, "books", "1", `{"title":"Shahnameh","author":"Ferdowsi","year":1010,"stock":3}`, opensearch.Options { refresh = opensearch.REFRESH_WAIT_FOR })
println "indexed:", w.ok, w.result, w.version
d := opensearch.Get(c, "books", "1")
if d.found:
println "title:", opensearch.JsonPath(d.source, "title")
end
u := opensearch.Update(c, "books", "1", `{"stock":2}`)
println "updated:", u.result, u.version
del := opensearch.Delete(c, "books", "1")
println "deleted:", del.result
endMany documents at once
Sending documents one request at a time is slow. A Bulk collects index, create, update and delete actions and sends them together. This example only builds the batch and prints the body it would send, so it runs here:
import db.opensearch
func main:
mut b := opensearch.NewBulk()
opensearch.BulkIndex(b, "books", "1", `{"title":"Shahnameh","author":"Ferdowsi","year":1010}`)
opensearch.BulkIndex(b, "books", "2", `{"title":"Masnavi","author":"Rumi","year":1258}`)
opensearch.BulkIndex(b, "books", "3", `{"title":"Golestan","author":"Saadi","year":1258}`)
opensearch.BulkUpdate(b, "books", "2", `{"stock":4}`)
opensearch.BulkDelete(b, "books", "99")
println opensearch.BulkLen(b), "actions"
print opensearch.BulkBody(b)
opensearch.BulkFree(b)
endBulkSend sends it. A bulk request succeeds as a whole even when some actions fail, so look at each item: ok, status and the error type and reason say what went wrong for that one document.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
mut b := opensearch.NewBulk()
opensearch.BulkIndex(b, "books", "1", `{"title":"Shahnameh, the Book of Kings","author":"Ferdowsi","year":1010}`)
opensearch.BulkIndex(b, "books", "2", `{"title":"Masnavi","author":"Rumi","year":1258}`)
opensearch.BulkIndex(b, "books", "3", `{"title":"Golestan","author":"Saadi","year":1258}`)
opensearch.BulkIndex(b, "books", "4", `{"title":"Kings and Heroes","author":"Anonymous","year":1900}`)
res := opensearch.BulkSendWith(c, b, opensearch.REFRESH_TRUE)
opensearch.BulkFree(b)
println res.succeeded, "stored,", res.failed, "failed"
each it in res.items:
if not it.ok:
println it.id, it.error_type, it.error_reason
end
end
endBuilding queries
A query is JSON as well, and the package has a builder for each common kind. Match searches the words of a text field, Term wants an exact value, and Range, RangeInt and DateRange compare. To combine them, fill a BoolQuery: must has to match and adds to the score, filter has to match but does not score, should raises the score, and must_not excludes. A SearchRequest adds paging, sorting and which fields to return; SearchBody shows what will be sent.
import db.opensearch
func main:
println opensearch.Match("title", "book of kings")
println opensearch.Term("author", "Ferdowsi")
println opensearch.RangeInt("year", 1000, 1300)
mut q := opensearch.BoolQuery {}
q.must.push(opensearch.Match("title", "kings"))
q.filter.push(opensearch.RangeInt("year", 1000, 1300))
q.must_not.push(opensearch.Term("author", "Anonymous"))
println opensearch.BoolOf(q)
mut req := opensearch.SearchRequest { query = opensearch.BoolOf(q), size = 5 }
req.sort.push(opensearch.SortDesc("year"))
req.source.push("title")
println opensearch.SearchBody(req)
endAnything the builders do not cover can be written by hand as a backtick string, since a query is only text. The builders, the request body and the answer can all be mixed with your own JSON.
Reading the hits
SearchWith runs the request and returns a SearchResult: the total number of matches, the time it took, and the hits of this page, best first. Each Hit has its id, score and source, plus the highlighted fragments you asked for.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
mut fields := Vector {} as Vector<str>
fields.push("title")
mut req := opensearch.SearchRequest { query = opensearch.Match("title", "kings"), size = 10 }
req.highlight = opensearch.Highlight(fields)
r := opensearch.SearchWith(c, "books", req)
if not r.ok:
println r.error
ret 1
end
println r.total, "matches"
each h in r.hits:
println h.id, h.score, h.source
end
ret 0
endThe same reader works on an answer you already have, through ParseSearch. That is how this example runs here without a server:
import db.opensearch
func main:
answer := `{"took":3,"timed_out":false,"hits":{"total":{"value":2,"relation":"eq"},"max_score":2.1,"hits":[{"_index":"books","_id":"1","_score":2.1,"_source":{"title":"Shahnameh","author":"Ferdowsi"},"highlight":{"title":["<em>Shahnameh</em>"]}},{"_index":"books","_id":"7","_score":0.8,"_source":{"title":"Tales from the Shahnameh","author":"Anonymous"}}]}}`
r := opensearch.ParseSearch(answer)
println r.total, "matches in", r.took, "ms"
each h in r.hits:
println h.id, h.score, opensearch.JsonPath(h.source, "title"), "by", opensearch.JsonPath(h.source, "author")
end
println opensearch.JsonPath(r.hits.get(0).highlight, "title.0")
endCounting with aggregations
Aggregations answer questions about all the matches at once: how many books per author, the average year, a histogram by month. Add them to a request with AddAgg, set size to 0 if you only want the numbers, and read the answer with Buckets for groups and AggValue for single values. Other readers cover the rest: AggField for one number of a stats result, AggPercentile for percentiles.
import db.opensearch
func main:
mut aggs := opensearch.AddAgg("{}", "authors", opensearch.TermsAgg("author", 3))
aggs = opensearch.AddAgg(aggs, "avg_year", opensearch.AvgAgg("year"))
println opensearch.SearchBody(opensearch.SearchRequest { size = 0, aggs = aggs })
answer := `{"took":2,"hits":{"total":{"value":40,"relation":"eq"},"hits":[]},"aggregations":{"authors":{"buckets":[{"key":"Saadi","doc_count":12},{"key":"Rumi","doc_count":9},{"key":"Hafez","doc_count":7}]},"avg_year":{"value":1265.5}}}`
r := opensearch.ParseSearch(answer)
each b in opensearch.Buckets(r.aggregations, "authors"):
println b.key, b.doc_count
end
println "average year:", opensearch.AggValue(r.aggregations, "avg_year")
endPaging through everything
from and size are fine for the first pages a person looks at, but not for walking through a whole index. For that, open a point in time with CreatePit, which freezes a view of the index, sort by a field, and ask for each next page with search_after set to the sort values of the last hit. An empty page means you are done; close the point in time with DeletePit.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
pit := opensearch.CreatePit(c, "books", "2m")
mut req := opensearch.SearchRequest { query = opensearch.MatchAll(), size = 100, pit_id = pit, pit_keep_alive = "2m" }
req.sort.push(opensearch.SortAsc("year"))
mut seen := 0
mut page := opensearch.SearchWith(c, "", req)
until page.hits.len() > 0:
seen += page.hits.len()
req.search_after = page.hits.get(page.hits.len() - 1).sort
page = opensearch.SearchWith(c, "", req)
end
opensearch.DeletePit(c, pit)
println seen, "books"
endSQL and PPL
OpenSearch also answers SQL. Conn turns a client into a db.Connection, the same interface the SQL engines use, so db.QueryInt and the row loop from the db package work unchanged. PPL is OpenSearch's piped query language, handy for logs; Ppl returns rows the same way. Long results come in pages, and the rows fetch the next page as you go.
import db
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
conn := opensearch.Conn(c)
println db.QueryInt(conn, "SELECT COUNT(*) FROM books WHERE year > 1200")
rows := conn.Query("SELECT author, COUNT(*) AS n FROM books GROUP BY author ORDER BY n DESC")
until rows.Next():
println rows.Text(0), rows.Int(1)
end
rows.Close()
p := opensearch.Ppl(c, "source=books | where year > 1200 | stats count() by author")
until p.Next():
println p.Text(1), p.Int(0)
end
p.Close()
endVector search
A knn_vector field holds an embedding, a list of numbers that places a text or an image in space so that similar things are close together. KnnIndexBody creates an index with one, and Knn finds the documents nearest to a vector. Neural and Hybrid go one step further and let OpenSearch make the vector from text with a model it hosts.
import db.opensearch
func main:
c := opensearch.Connect("http://localhost:9200")
method := opensearch.KnnMethod { engine = "lucene", space_type = "cosinesimil" }
mut extra := Vector {} as Vector<str>
extra.push(opensearch.Property("name", "text"))
opensearch.CreateIndex(c, "places", opensearch.KnnIndexBody("embedding", 3, method, extra))
opensearch.Index(c, "places", "1", `{"name":"Naderi Cafe","embedding":[0.1,0.9,0.2]}`)
opensearch.Refresh(c, "places")
mut v := Vector {} as Vector<f64>
v.push(0.1)
v.push(0.8)
v.push(0.3)
r := opensearch.Search(c, "places", opensearch.Knn("embedding", v, 3))
each h in r.hits:
println h.id, h.score, h.source
end
endWhen something goes wrong
No call stops your program. A call that fails returns false, -1 or an empty result, and the client remembers what happened: Status is the HTTP status of the last request (0 when no answer came), Error is a readable message, and ErrorType and ErrorReason are what OpenSearch said, such as version_conflict_engine_exception. A bad url is caught when the client is made, and Ok tells you.
import db.opensearch
func main:
mut cfg := opensearch.DefaultConfig()
cfg.nodes = Vector {} as Vector<str>
cfg.nodes.push("http://127.0.0.1:1")
cfg.max_retries = 0
c := opensearch.NewClient(cfg)
println "config ok:", opensearch.Ok(c)
d := opensearch.Get(c, "books", "1")
println "found:", d.found
println "status:", opensearch.Status(c)
println "error:", opensearch.Error(c)
bad := opensearch.Connect("ftp://example.com")
println "config ok:", opensearch.Ok(bad), "|", opensearch.Error(bad)
endWhere to go next
The db/opensearch reference lists every function with an example, including index templates, aliases, snapshots, ingest pipelines, ISM policies, the security plugin, alerting and ML Commons. For anything the package has no function for, Perform sends any request you write.