[RFC] PPL outputlookup command (write search results to a lookup)

[noCharger]

Problem Statement

PPL can read lookups (lookup, source=<lookup_index>) but has no way to write a search result back into a lookup dataset. Users who want to derive a lookup (dimension) table from a query, such as a per-host rollup, a filtered subset, or a deduplicated key set, must build and refresh lookup indices out of band (frontend data importer, manual bulk indexing, external ETL). There is no in-query way to materialize a lookup from a pipeline result.

Current State

• lookup enriches rows from an existing lookup index; no PPL command produces or refreshes one.
• Lookups are created and maintained outside PPL.
• The already-merged Dashboards data importer (OpenSearch-Dashboards#11303) ships a lookup substrate: a filtered alias ({term:{__lookup:<uuid>}}) over a user-chosen backing index, written as a new __lookup slice and made visible by pointing the alias at it. The importer addresses its target index by name, with the filtered alias as an optional versioning layer on top. Related RFC #5074 proposed read/write lookup commands. This RFC adopts the same alias-plus-uuid-slice mechanism so PPL writes and importer writes share one shape, using a dedicated per-lookup backing index so lookups stay isolated from one another.

Long-Term Goals

• Give PPL a first-class way to persist a query result as a lookup, so downstream searches enrich against a small precomputed dataset instead of rescanning raw data.
• Keep the operation simple and predictable:
  • Overwrite is content-atomic and gap-free: a fresh slice is written and the alias is atomically repointed, so a concurrent reader always resolves the complete old lookup or the complete new lookup, never a partial one, and a crash mid-write leaves the old lookup serving and self-heals on re-run.
  • Append is incremental and at-least-once; a keyed write is idempotent on re-run.
• Run entirely under the caller's security context with no new privilege surface, so it is safe to expose on managed or multi-tenant clusters.
• Serve the lookup use case (small, bounded dimension datasets) and keep a single call's resource footprint bounded via an operator ceiling, so a large or unbounded query cannot OOM the coordinator or become a noisy neighbor.

Proposal

Add a terminal, synchronous PPL command:

... | outputlookup [append=<bool>] [override_if_empty=<bool>] [key_field=<f1>(,<f2>)*] [max=<int>] <name>

It materializes the current pipeline result into the lookup <name> and returns a single rows_written count. The command is single-name: <name> is the lookup, matching SPL outputlookup <name>; the physical backing index is derived and hidden from the user.

• Overwrite (append=false, default): replace the lookup with the result (atomic, gap-free).
• append=true: append to the existing lookup.
• override_if_empty (default true): an empty result clears the lookup; =false leaves it intact.
• key_field=<f...>: upsert by key (idempotent re-run, no duplicates); implies append=true. Every key field must be a field of the result; multivalue key values are rejected.
• max=<n>: per-query cap; truncates the result to at most n rows (user-opted, like SQL LIMIT).
• Terminal sink: returns a rows_written count rather than forwarding input rows. Multivalue fields are preserved as native arrays. The destination is created on demand.

A cluster setting plugins.ppl.outputlookup.max_rows (NodeScope, Dynamic, default 1_000_000, min 1) is an operator ceiling on how many rows a single call may write (see Operator ceiling below).

Approach

Substrate = one dedicated backing index per lookup. A lookup <name> is a __lookup=<uuid> slice inside an ordinary, non-hidden index <name>__lookup, behind a filtered alias <name> ({term:{__lookup:<uuid>}}), the same alias-plus-slice artifact #11303 produces. The backing index is created on demand and owns its own mapping via a uniform dynamic template (string maps to text plus keyword, date and numeric detection off), so the write honors that mapping rather than imposing a keyword-only one. Reads (lookup <name>, source=<name>) resolve the alias and see only that lookup's slice. Because each lookup has its own index, there is no shared mapping, no cross-lookup type conflict, and write authorization is per-index.

1. Overwrite (append=false): write a fresh __lookup=<new-uuid> slice, then atomically repoint the alias onto it in one _aliases request (remove the old filter, add the new). Reads are content-atomic and gap-free. The previous slice is left orphaned (a __lookup uuid referenced by no alias) for a follow-up reaper.
2. Append (append=true): resolve the current slice uuid from the alias filter and bulk into it. Incrementally visible, at-least-once.
3. Append to an absent lookup uses a deterministic per-lookup discriminant, so two concurrent first-appends converge into the same slice and install the same alias filter (idempotent), with no lost write and no orphan. This defines the same-name concurrency contract; overwrite stays last-writer-wins on the atomic repoint.
4. key_field upsert uses a deterministic _id salted with the slice uuid, so keyed rows never collide across slices; length-prefixed per-field canonical encoding distinguishes empty string from null, encodes BigDecimal and BigInteger without truncation, and rejects multivalue keys. A keyed append is idempotent on re-run; a plain append may duplicate.
5. The command is a Calcite TableModify (INSERT) node, so the optimizer treats it as a mandatory side effect (never dropped or reordered), with the rowcount row type named rows_written. It does not require a source scan: the in-cluster client is resolved from the schema, so a sourceless pipeline such as makeresults | outputlookup works. The source pages unbounded via PIT, so the whole result is written in full, subject only to the operator ceiling below (fail-loud, never a silent truncation).

Operator ceiling: plugins.ppl.outputlookup.max_rows

max_rows caps how many rows a single outputlookup call may write.

• It is fail-loud: if the pipeline produces more than the ceiling, the command throws (HTTP 400) and writes nothing (no slice, no alias change). It never writes a truncated slice.
• This preserves read/write consistency: a silent truncation would let the lookup diverge from its source with no error. Fail-loud makes it all-or-nothing. It is orthogonal to max=<int>, which is a user-opted truncation and therefore not a divergence. The command also validates 1 <= max <= max_rows before any write.
• It makes outputlookup a bounded dimension-table sink, not a bulk loader. Bounding the per-call row count also bounds the source PIT lifetime and the coordinator buffer, the most direct mitigation for noisy-neighbor impact; PIT count and lifetime are otherwise governed by core OpenSearch (search.max_open_pit_context, plugins.sql.cursor.keep_alive).
• A lookup larger than the ceiling is built by paged append, each call bounded and backpressured by indexing_pressure.

Safety / permissions: writes execute under the caller's security context (no privilege escalation). The caller needs indices:data/write/bulk, indices:admin/aliases, and indices:admin/get (to probe the target) on the relevant names, plus indices:admin/create the first time the backing index is created. No cluster-level permission is required. Because each lookup is an ordinary non-hidden index, ordinary index patterns cover it and write authorization is per-lookup: a grant on lookup A does not grant write on lookup B. Verified by a permissions integration test in which a read-privileged user reads a lookup through its alias.

Full resolution of <name> (every case). The target is in one of these states, and the backing index is <name>__lookup:

<name> current state	overwrite (append=false)	append (append=true)
absent	write new slice, add filtered alias	write new slice (deterministic discriminant so concurrent first-appends converge), add filtered alias
our lookup (filtered alias)	write new slice, atomically repoint the alias; old slice orphaned for the reaper	bulk into the current slice
concrete index of the same name	refuse (a concrete index cannot share a name with the alias)	refuse
#11303 filtered alias over another index	migrate: write a <name>__lookup slice and repoint the alias onto it; the legacy index is not deleted; the old slice is orphaned	refuse (overwrite once to migrate)
other (non-filtered) alias	refuse	refuse

Alternative

• Single shared .lookups index (co-locate all lookups as __lookup slices in one hidden index) is rejected. It re-inherits RFC [RFC] Discriminator-based lookups #5074's problems: mapping field explosion across all lookups (total_fields), first-writer-wins type conflicts on shared field names, a single-index blast radius, and a hidden dot-prefixed index that a role must be granted .* (or the exact name) to read. It also cannot scope write authorization per lookup, because a bulk grant on the shared index writes every slice and a filtered alias does not constrain writes. A dedicated per-lookup index avoids all of these while using the same alias-plus-uuid-slice mechanism as #11303.
• Plain index per lookup with delete-plus-recreate overwrite is rejected. It gives only a weak, non-atomic overwrite (an absent or rebuilding window, and a crash can lose the lookup). The alias-plus-uuid-slice with atomic repoint gives content-atomic gap-free overwrite at the same per-lookup granularity.
• Streaming-batch overwrite (page the read and bulk each page under one PIT, repoint once at the end) is rejected for v1. It lowers coordinator heap but holds the source PIT and indexing pressure on the data nodes for the whole write, trading a coordinator spike for a longer data-node noisy-neighbor window. Bounded writes (max_rows) plus paged append are preferred.
• Asynchronous or task-based write is rejected for v1: outputlookup is naturally synchronous and terminal.

Performance (measured)

3-node m5.xlarge, node-side, single run per cell (directional). The write is materialized like a build-time index: the pre-publish slice is invisible until the atomic alias repoint, so it loads with 0 replicas, async translog, and no auto refresh, all restored to serving settings before publish, and a single refresh replaces per-batch refreshes. With this, outputlookup write throughput is on par with or faster than a plain _bulk load of the same rows (overwrite, rows/s):

write	outputlookup	plain _bulk	ratio
100K / 2 cols	25,802	27,533	0.94x
100K / 5 cols	30,980	19,984	1.55x
100K / 20 cols	11,145	11,329	0.98x
1M / 2 cols	44,358	31,275	1.42x
1M / 5 cols	32,982	24,043	1.37x
1M / 20 cols	13,774	10,976	1.25x

The _bulk baseline used default index settings (1 replica, request durability), so these ratios reflect each path at its realistic write configuration, not identical durability; the point is that the per-batch refresh tax an earlier build carried (about 2.5x slower than bulk) is eliminated and the write is now bulk-class. The write is bounded by max_rows; 1M rows wrote un-truncated with no OOM even at 20 columns (20M cells) on an 8GB heap. Read tax is negligible (about 11 to 15 ms for a 100-row lookup) and flat across 0 / 9 / 99 orphan slices.

Scale

Each lookup is its own bounded index, so the shared-index ceilings (mapping explosion, cross-lookup type conflict, single-index blast radius) do not arise. max_rows bounds a single lookup and a single write. The tradeoff is the opposite one: a user who materializes very many lookups creates many small indices (shard and cluster-state overhead). outputlookup targets a bounded number of query-materialized dimension tables; bulk ingestion of many tiny files remains the data importer's job. The only accumulation within a lookup's own index is orphaned slices from overwrite, reclaimed by a follow-up reaper (enumerate the __lookup uuids, subtract those referenced by the alias, delete the remainder).

Implementation Discussion

Implemented in PR #5621: grammar, AST, and parse (single-name syntax); the TableModify node (row type named rows_written) plus physical write operator, bulk writer, and deterministic uuid-salted _id encoder (BigDecimal and BigInteger safe); the per-lookup backing-index lifecycle (ensureExists non-hidden with a uniform dynamic template; overwrite = new slice + atomic alias repoint; append = bulk into the current slice; append-to-absent = deterministic discriminant so concurrent first-appends converge); a client resolved from the schema so sourceless pipelines work; a structured XContentParser read of the alias __lookup filter; an indices-level target probe (no cluster:monitor/state); migrate-on-touch for a legacy #11303 filtered alias (repoint onto <name>__lookup, never delete the legacy index); key_field schema validation at plan time; 1 <= max <= max_rows validation; a bounded bulk-retry loop with an absolute timeout; and the plugins.ppl.outputlookup.max_rows operator ceiling (fail-loud, no partial write). Tests: CalcitePPLOutputLookupIT (20, including per-lookup isolation, sourceless makeresults|outputlookup, concurrent append-to-absent, large-source no-truncation, multi-field upsert, concrete-index-refused, importer-alias migration, and the ceiling), OutputLookupPermissionsIT (2), LookupIdEncoderTest (including a BigDecimal regression), AstBuilderTest, plus user docs.

Consistency contract. Overwrite is content-atomic and gap-free (write new slice, atomic alias repoint); a crash before the repoint leaves the old lookup serving and self-heals on re-run (the orphan is invisible through the filter and enumerable for the reaper). Append is at-least-once; a keyed append is idempotent. Concurrent appends to an absent lookup converge into one slice (deterministic discriminant) with no lost write; concurrent overwrites are last-writer-wins. Verified: a same-name concurrent-overwrite chaos run observed last-writer-wins with 13,532 reads and zero torn, partial, or mixed states across 30 atomic repoints; a concurrent append-to-absent integration test confirms both appends persist.

Coexistence with the importer (no deprecation planned). outputlookup and the #11303 importer both use the alias-plus-uuid-slice shape, so they coexist by construction; reads (lookup and source=<name>) resolve either transparently. The only conversion is migrate-on-collision: overwriting a name that is currently a legacy filtered alias repoints it onto a <name>__lookup slice, and only that one lookup.

Resolved design decisions:

1. A lookup is a dedicated per-lookup, non-hidden backing index (<name>__lookup), not a shared hidden .lookups. This removes the shared-mapping, type-conflict, hidden-index, and per-lookup-write-authz concerns by construction.
2. The command is single-name (<name>), matching SPL; there is no user-facing backing-index option, and the backing index is derived and hidden.
3. Same-name concurrency: append-to-absent uses a deterministic discriminant (concurrent convergence, no lost write); overwrite is last-writer-wins on the atomic repoint.
4. The target-state probe is indices-level (get-index and get-aliases) and needs no cluster:monitor/state.
5. No dedicated inputlookup command; source=<name> suffices.
6. max_rows is a fail-loud operator ceiling (no partial write), orthogonal to the per-query max=<int> truncation.
7. Orphaned __lookup slices (left by every overwrite, and by crash or concurrent-overwrite losers) are reclaimed by a periodic reaper, delivered as a separate PR; this PR leaves the slice in place with an observability log and documents the discriminant contract the reaper keys on.

Remaining for discussion:
8. Co-design with #11303: is a shared discovery and management contract for "what is a lookup" worth standardizing across the two writers, and does the migrate-on-collision plus permanent-coexistence model match the data-importer owners' expectations?