Turns a search into an Elixir Stream, for walking more documents than fit in one response.

%{query: %{match_all: %{}}, size: 1_000, sort: [%{"_id" => "asc"}]}
|> Dowser.Opensearch.Streamer.stream(index: "posts")
|> Stream.map(& &1["_source"])
|> Enum.each(&process/1)

Each element is a hit — _index, _id, _source and the rest — cast exactly as Dowser.Opensearch.Search.search/2 would cast it.

How it walks

A point in time pins the index against concurrent writes, and search_after pages through it, taking each page's cursor from the last hit of the page before it.

search_after is sequential by construction, so one stream cannot fetch pages in parallel. stream_slices/4 is how you get parallelism; see below.

The sort is yours to give

:sort is required in the query, and its last key must be unique per document.

This is the one place this module differs from dowser_elasticsearch, and the difference comes from OpenSearch rather than from taste. Elasticsearch has a _shard_doc sort field — shard index plus Lucene document id — that exists precisely to tiebreak a point-in-time walk, so the Elasticsearch streamer appends it for you and needs no sort from the caller. OpenSearch forked before _shard_doc was introduced and never added it, and there is nothing in its sort vocabulary that plays the same role: _doc is a per-shard document id, so the same value occurs once per shard, and search_after on a non-unique final key silently skips or repeats the documents that share it.

A guessed default would therefore be a walk that quietly loses documents on any multi-shard index, so this module asks instead. _id is always unique and needs no mapping change:

%{query: %{match_all: %{}}, sort: [%{"_id" => "asc"}]}

Sorting on _id loads its field data, which costs memory on a large index; where you have a unique, indexed field of your own — a monotonic id, a timestamp plus a tiebreaker — prefer it:

%{query: ..., sort: [%{"created_at" => "asc"}, %{"_id" => "asc"}]}

If your cluster does serve _shard_doc, pass it yourself — nothing here rejects it.

Options

• :index — the index to open the point in time on. Required unless :pit is given. It is not sent with the searches themselves: OpenSearch rejects a search that names both an index and a PIT.
• :pit — an existing point-in-time id to walk instead of opening one. It is checked once before the walk starts, and left open afterwards — whoever opened it closes it. Without it, the stream opens its own and closes it when enumeration ends, however it ends.
• :keep_alive — how long OpenSearch holds the point in time, extended on every search. Defaults to "1m". Long enough to cover the gap between two pages, not the whole walk.
• :verify_pit — whether a :pit you passed is checked before the walk starts. true by default; false when you just opened it and know it is alive.

Everything else is forwarded to Dowser.Opensearch.Search, so :context, :codec, :keys and :http_opts all work as usual.

A query with no size gets 1000, not OpenSearch's default of 10 — at 10 hits per round trip a million documents is a hundred thousand requests. Set your own when you have a reason to.

Slicing

stream_slices/4 splits the point in time into disjoint subsets and walks them at once. OpenSearch divides first across shards, then within each shard by contiguous ranges of Lucene document ids, so the natural ceiling is your shard count — more slices than shards subdivides a shard rather than adding parallelism.

%{query: %{match_all: %{}}, size: 1_000, sort: [%{"_id" => "asc"}]}
|> Dowser.Opensearch.Streamer.stream_slices(4, &Enum.count/1, index: "posts")
|> Enum.sum()

Note what it takes: a function, not a stream. Each slice is consumed inside its own task, because a lazy stream handed back out of a task would run every page in the caller — concurrent in name only.

To spread a walk across nodes instead, open the point in time yourself and give each node one slice. slice is an ordinary search body field, so it goes in the query rather than the options:

stream(%{query: ..., sort: ..., slice: %{id: 2, max: 8}}, pit: pit_id)

Failure

There is no stream!/2. Every other function in this package comes in a pair because it returns {:ok, result} or {:error, error} and the bang variant unwraps it; a stream has nothing to unwrap, and nothing has happened yet when it is built. Errors surface on enumeration, by raising — which is what the bang variant would have done anyway.

A missing :sort is the exception: that is a mistake in the call rather than a failure of the cluster, so it raises when the stream is built, not when it is walked.

The stream raises on the first failed request, and closes a point in time it opened on the way out — on normal completion, on Enum.take/2, and on an exception. It cannot close one if the enumerating process is killed outright; :keep_alive is the backstop there.