Elasticsearch

Elasticsearch never holds a Wow log: every aggregate's log stays on PostgreSQL or MongoDB, the source of truth. Elasticsearch gets copies: the snapshots, so their queries run on a search engine, and the event streams, through a sink, so the events model can be searched across aggregates. Both can be rebuilt from the log at any time. The module is wow-elasticsearch.

A store of type elasticsearch

A store of type elasticsearch opens over the application's low-level Rest5Client, which Spring Boot's Elasticsearch support configures. Name it beside the log's store, add it to an aggregate's snapshot stores after the store its commands load from, and point the aggregate's queries at it:

wow:
  stores:
    main: { type: postgres }
    search:
      type: elasticsearch
      index-prefix: shop
  log:
    store: main
  aggregates:
    "[shop.order]":
      snapshot:
        stores: [main, search]
      query:
        snapshot:
          store: search
        events:
          store: search

Key	Sets	Default
wow.stores.<name>.client	The Rest5Client bean the store writes through	The only Rest5Client
wow.stores.<name>.query-client	A second Rest5Client, a connection pool of the queries' own	The store's client
wow.stores.<name>.index-prefix	A prefix for every index name: shop_wow_snapshot__…, shop_wow_events__…	None
wow.stores.<name>.refresh	When a snapshot write becomes searchable: wait-for, immediate or none	wait-for
wow.stores.<name>.ddl	Whether it creates its indices and maps the state's and the events' body fields	wow.ddl
wow.stores.<name>.reindex-requests-per-second	How fast an index's earlier documents are indexed again for a field mapped later (below); -1 for no limit	1000
wow.aggregates.[<context>.<aggregate>].snapshot.stores	The stores of an aggregate's snapshots	wow.snapshot.stores, else the log's store
wow.aggregates.[<context>.<aggregate>].snapshot.primary	The one of them its write side and state reads read; never Elasticsearch	The first of its own snapshot.stores when it lists them; otherwise wow.snapshot.primary, else the first of wow.snapshot.stores
wow.aggregates.[<context>.<aggregate>].query.snapshot.store	The store its snapshot queries read	Its primary snapshot store
wow.aggregates.[<context>.<aggregate>].query.events.store	The store its events queries read	The log's store

Each index records its layout in its _meta and refuses a store that expects another.

Snapshot queries

An aggregate whose query.snapshot.store names the Elasticsearch store has its snapshot queries served from there, with the same query language and capability descriptor as any backend. With the default refresh: wait-for, a snapshot is searchable when its write returns. The queries of a store run under its query bulkhead: wow.stores.<name>.query.max-concurrent, by default a quarter of the write pool's connections per route, at least two, or the whole pool of a query-client. A query beyond it waits briefly, then is answered 429 wow.query.busy.

A map in the state or in an event body (an object of any keys, such as a Map<String, Item> property) is mapped as one flattened field, so a value under one of its keys takes presence and exact-match filters only: no range, no relative time, no sort and no aggregation. The descriptor's dynamic patterns list exactly those operators.

A snapshot's document, and an event search document, also hold accessPresent: true under each access attribute the aggregate holds, a single one with a value, a multiple one empty or not. Elasticsearch indexes nothing for an empty array, so an access policy reads there whether an attribute is present, and grants on Elasticsearch the aggregates it grants on any store and to a state read. A document written before an attribute was declared lacks it, as the attribute is missing from it, until the aggregate's next commit or the access walk writes it again.

Full-text search

A string field of the state or of an event body is mapped as a keyword: matched by its whole value, as equality, contains, startsWith, sorts and groups match it. It is searched by words only where its schema marks it as prose: @QueryText on the property (Describe what types cannot), x-wow-text in the Manifest, text() in a declaration in code. A marked field gets a text sub-field, analyzed by the index's default analyzer, and the capability descriptor lists it among the fields a search may name: record.search.fields, or its element's.

• A search that names fields reads those, each of which must be marked; another is answered unsupported-capability.
• A search that names no field reads the marked fields of the record, outside its elements, the ones record.search.fields lists. Where none is marked, it is answered model-search-unsupported.
• A phrase needs every field it reads to take one, as every text sub-field does.

Mark only prose a reader searches by words, such as a name, a description, a note or a reason. Ids, enums, codes and references are matched by their whole value, and an index of each of their words only costs space and indexing time.

Marking a field later adds the text sub-field to the index and indexes the documents written before it again (below). Removing a marker leaves the sub-field in the mapping, and the field is no longer searched by words. When an index name covers several indices, an alias over rollover indices, and only some of them map the sub-field, the field keeps its whole-value matches and is searched by words once every index maps it.

A field mapped later

The state's and the events' body fields are mapped when a query first loads the aggregate's schema (under ddl: auto), so a field added to a schema later is mapped on an index that already holds documents. Those documents are indexed again in the background, by an _update_by_query task that keeps each document as it is, so that they are found by the new field too. Until the task ends, a query that filters, sorts or groups on the new field misses the documents written before the field was mapped: its results are incomplete.

The task runs at reindex-requests-per-second documents a second, 1,000 by default, in as many slices as the index has shards. At the default a million documents take about 17 minutes, and ten million about 3 hours. Lower it for a cluster under load; raise it, or set -1 for no limit, when the field is needed sooner.

The store logs the task's id as it starts, Elasticsearch index [<index>] mapped [<fields>]: … task [<node>:<id>]. It does not log the task's end, which only polling the task would tell. Follow the task, and change its rate while it runs, with Elasticsearch's own APIs:

curl "$ELASTICSEARCH_URL/_tasks/<node>:<id>"
curl -X POST "$ELASTICSEARCH_URL/_update_by_query/<node>:<id>/_rethrottle?requests_per_second=5000"

GET _tasks/<id> answers "completed": true once every earlier document is indexed again.

Event search

The event search sink writes each commit as one document of [<prefix>_]wow_events__<context>__<aggregate>: the events model's record, its system fields, and its events as the nested array events, each with its name, revision and body as committed. A document's id is the commit's identity, tenant, aggregate id and version, so a commit written again changes nothing. The body fields are mapped from the Manifest's event schemas; a field two events declare with different types is left unmapped.

An aggregate whose query.events.store names the Elasticsearch store has its events queries served from that index: an event condition is a nested query, an expand("events") aggregation a nested aggregation, and a query that names no aggregate searches the index instead of scanning the log. The index shows what the sink has written, a commit becoming searchable once the sink has written it and the index refreshed, one second by default.

Each sink runs as a subscription named wow.elasticsearch-sink.<aggregate>, so it resumes from its checkpoint and can start again from the beginning of the log. The host starts the sinks by itself, with no setting of their own: for each Elasticsearch store that an aggregate's query.events.store names, it writes exactly those aggregates' streams into that store, whether or not the service publishes them.

Store combinations

Each aggregate type picks its stores: its log and its primary snapshots on PostgreSQL or MongoDB (wow.log.store, wow.aggregates.[<context>.<aggregate>].log.store), a copy of its snapshots and its events search on Elasticsearch, as the sample above sets them. Two consequences:

• Commands and state reads stay on the primary. A command loads its state from the primary snapshot store and the log after it, and the state reads (GET …/state) read the primary, so neither waits for Elasticsearch's refresh. Each snapshot store is written by a subscription of its own from the log, wow.snapshot for the default snapshot store (the first of wow.snapshot.stores, else the log's store) and wow.snapshot:<store> for each other, and keeps the highest version it was given, so the copy only lags and can be rebuilt. The host refuses Elasticsearch as an aggregate's primary snapshot store, or its only one, naming the key.
• Events queries follow the sink. An aggregate's events queries read Elasticsearch once its query.events.store names the store, so they see what the sink has written: a commit is searchable shortly after it commits, not at once. Its snapshots may stay on the log's store; the two settings are independent.

Rebuild

Both copies come from the log. A snapshot index is rebuilt by rebuilding the aggregates' snapshots (Rebuild and Replay). The events index is filled again by starting its sink subscription from the beginning of the log. Neither needs a backup of its own (Backup and Recovery).

From 9.x

The 9.x Elasticsearch event store is not a migration source: migrate-data reads a 9.x MongoDB event store only, so move the streams to MongoDB within 9.x first (Migrate the Data). 9.x indices are not read either. Give Wow 10 its own index-prefix or cluster: once the log is migrated, the snapshots are written from it, and the event search sink, started from the beginning, fills the events index.