[Feature] Add PPL outputlookup command (synchronous terminal write sink)

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -36,6 +36,7 @@ public enum Key {
     PPL_SYNTAX_LEGACY_PREFERRED("plugins.ppl.syntax.legacy.preferred"),
     PPL_SUBSEARCH_MAXOUT("plugins.ppl.subsearch.maxout"),
     PPL_JOIN_SUBSEARCH_MAXOUT("plugins.ppl.join.subsearch_maxout"),
+    OUTPUTLOOKUP_MAX_ROWS("plugins.ppl.outputlookup.max_rows"),
 
     /** Enable Calcite as execution engine */
     CALCITE_ENGINE_ENABLED("plugins.calcite.enabled"),

core/src/main/java/org/opensearch/sql/analysis/Analyzer.java

@@ -563,6 +563,12 @@ public LogicalPlan visitNoMv(NoMv node, AnalysisContext context) {
     throw getOnlyForCalciteException("nomv");
   }
 
+  @Override
+  public LogicalPlan visitOutputLookup(
+      org.opensearch.sql.ast.tree.OutputLookup node, AnalysisContext context) {
+    throw getOnlyForCalciteException("outputlookup");
+  }
+
   @Override
   public LogicalPlan visitMakeResults(MakeResults node, AnalysisContext context) {
     throw getOnlyForCalciteException("makeresults");

core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java

@@ -305,6 +305,10 @@ public T visitReverse(Reverse node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitOutputLookup(org.opensearch.sql.ast.tree.OutputLookup node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitTranspose(Transpose node, C context) {
     return visitChildren(node, context);
   }

core/src/main/java/org/opensearch/sql/ast/tree/OutputLookup.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.tree;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+
+/**
+ * AST node for the {@code outputlookup} command: a terminal write sink that materializes pipeline
+ * rows into a lookup index. Overwrite-by-default (set {@code append=true} to append instead). See
+ * the outputlookup PPL design.
+ */
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@RequiredArgsConstructor
+@AllArgsConstructor
+public class OutputLookup extends UnresolvedPlan {
+
+  private UnresolvedPlan child;
+
+  /** Destination lookup name. */
+  private final String indexName;
+
+  /** false (default) overwrites the destination; true appends to it. */
+  private boolean append = false;
+
+  /** true (default) clears the destination on empty results; false keeps it. */
+  private boolean overrideIfEmpty = true;
+
+  /**
+   * Fields whose values form the document {@code _id} for upsert; empty means auto-generated id.
+   */
+  private List<String> keyFields = java.util.List.of();
+
+  /** Cap on the number of rows written; null means unbounded. */
+  private Integer max;
+
+  @Override
+  public OutputLookup attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  @Override
+  public List<UnresolvedPlan> getChild() {
+    return this.child == null ? ImmutableList.of() : ImmutableList.of(this.child);
+  }
+
+  @Override
+  public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) {
+    return nodeVisitor.visitOutputLookup(this, context);
+  }
+}

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -926,6 +926,67 @@ public RelNode visitReverse(
     return context.relBuilder.peek();
   }
 
+  @Override
+  public RelNode visitOutputLookup(
+      org.opensearch.sql.ast.tree.OutputLookup node, CalcitePlanContext context) {
+    visitChildren(node, context);
+    String indexName = node.getIndexName();
+    if (indexName.startsWith(".")) {
+      throw new IllegalArgumentException(
+          "outputlookup destination [" + indexName + "] must not be a system (dot-prefixed) index");
+    }
+    RelNode child = context.relBuilder.build();
+    // Validate key_field names against the result schema: a missing/misspelled key field would
+    // otherwise encode identically for every row and collapse them into a single document.
+    if (!node.getKeyFields().isEmpty()) {
+      java.util.List<String> resultFields = child.getRowType().getFieldNames();
+      for (String keyField : node.getKeyFields()) {
+        if (!resultFields.contains(keyField)) {
+          throw new IllegalArgumentException(
+              "outputlookup key_field ["
+                  + keyField
+                  + "] is not a field of the result; available fields: "
+                  + resultFields);
+        }
+      }
+    }
+    org.apache.calcite.plan.RelOptTable sourceTable = findSourceTable(child);
+    if (sourceTable == null) {
+      throw new IllegalArgumentException(
+          "outputlookup requires an OpenSearch source scan in the pipeline");
+    }
+    org.apache.calcite.plan.RelOptSchema relOptSchema = context.relBuilder.getRelOptSchema();
+    if (!(relOptSchema instanceof org.apache.calcite.prepare.Prepare.CatalogReader)) {
+      throw new IllegalStateException("outputlookup could not obtain a Calcite catalog reader");
+    }
+    RelNode sink =
+        OutputLookupTableModify.create(
+            child,
+            sourceTable,
+            (org.apache.calcite.prepare.Prepare.CatalogReader) relOptSchema,
+            indexName,
+            node.isAppend(),
+            node.isOverrideIfEmpty(),
+            node.getKeyFields(),
+            node.getMax());
+    context.relBuilder.push(sink);
+    return context.relBuilder.peek();
+  }
+
+  /** Walk down to the first scan carrying a table, so the write rule can reach the client. */
+  private static org.apache.calcite.plan.RelOptTable findSourceTable(RelNode rel) {
+    if (rel.getTable() != null) {
+      return rel.getTable();
+    }
+    for (RelNode input : rel.getInputs()) {
+      org.apache.calcite.plan.RelOptTable found = findSourceTable(input);
+      if (found != null) {
+        return found;
+      }
+    }
+    return null;
+  }
+
   @Override
   public RelNode visitTranspose(
       org.opensearch.sql.ast.tree.Transpose node, CalcitePlanContext context) {

core/src/main/java/org/opensearch/sql/calcite/OutputLookupTableModify.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite;
+
+import java.util.List;
+import lombok.Getter;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.prepare.Prepare.CatalogReader;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.TableModify;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Terminal write node for outputlookup, modeled as a Calcite {@link TableModify} INSERT so the
+ * optimizer treats it as a mandatory table-modifying side effect (never dropped or reordered) and
+ * it exposes the standard rowcount row type. The referenced table is the source scan, used only so
+ * the lowering rule can reach the in-cluster client; the real destination lookup index is created
+ * and written at execution time by the physical operator, so a preexisting destination table is not
+ * required. Subclasses {@link TableModify} rather than LogicalTableModify so the built-in
+ * EnumerableTableModifyRule (which needs a ModifiableTable) does not fire on it.
+ */
+@Getter
+public class OutputLookupTableModify extends TableModify {
+
+  private final String indexName;
+  private final boolean append;
+  private final boolean overrideIfEmpty;
+  private final java.util.List<String> keyFields;
+  private final @Nullable Integer max;
+
+  public OutputLookupTableModify(
+      RelOptCluster cluster,
+      RelTraitSet traits,
+      RelOptTable table,
+      CatalogReader catalogReader,
+      RelNode input,
+      String indexName,
+      boolean append,
+      boolean overrideIfEmpty,
+      java.util.List<String> keyFields,
+      @Nullable Integer max) {
+    super(cluster, traits, table, catalogReader, input, Operation.INSERT, null, null, false);
+    this.indexName = indexName;
+    this.append = append;
+    this.overrideIfEmpty = overrideIfEmpty;
+    this.keyFields = keyFields;
+    this.max = max;
+  }
+
+  public static OutputLookupTableModify create(
+      RelNode input,
+      RelOptTable table,
+      CatalogReader catalogReader,
+      String indexName,
+      boolean append,
+      boolean overrideIfEmpty,
+      java.util.List<String> keyFields,
+      @Nullable Integer max) {
+    RelOptCluster cluster = input.getCluster();
+    RelTraitSet traits = cluster.traitSetOf(Convention.NONE);
+    return new OutputLookupTableModify(
+        cluster,
+        traits,
+        table,
+        catalogReader,
+        input,
+        indexName,
+        append,
+        overrideIfEmpty,
+        keyFields,
+        max);
+  }
+
+  @Override
+  public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+    return new OutputLookupTableModify(
+        getCluster(),
+        traitSet,
+        getTable(),
+        getCatalogReader(),
+        inputs.get(0),
+        indexName,
+        append,
+        overrideIfEmpty,
+        keyFields,
+        max);
+  }
+}

docs/category.json

@@ -25,6 +25,7 @@
     "user/ppl/cmd/head.md",
     "user/ppl/cmd/join.md",
     "user/ppl/cmd/lookup.md",
+    "user/ppl/cmd/outputlookup.md",
     "user/ppl/cmd/mvcombine.md",
     "user/ppl/cmd/nomv.md",
     "user/ppl/cmd/mvexpand.md",

docs/user/ppl/cmd/outputlookup.md

@@ -0,0 +1,96 @@
+
+# outputlookup
+
+The `outputlookup` command is a terminal write sink: it materializes the current pipeline result into a lookup index and returns a single `rows_written` count. Use it to build or refresh a lookup (dimension) dataset from a search, which can then be read back with `source=<name>` or enriched into other searches with the `lookup` command.
+
+A lookup name refers to a plain index. Overwrite replaces that index with the current result (its schema is replaced each run); append adds the result to it. Writes are eventually consistent: a reader during an overwrite may briefly see the lookup being rebuilt.
+
+## Syntax
+
+The `outputlookup` command has the following syntax:
+
+```syntax
+outputlookup [append=<bool>] [override_if_empty=<bool>] [key_field=<field>(, <field>)*] [max=<int>] <name>
+```
+
+The following are examples of the `outputlookup` command syntax:
+
+```syntax
+source = table | outputlookup my_lookup
+source = table | where status = 'active' | outputlookup active_hosts
+source = table | stats count by region | outputlookup region_counts
+source = table | fields id, name | outputlookup append=true my_lookup
+source = table | outputlookup override_if_empty=false my_lookup
+source = table | fields id, name | outputlookup key_field=id my_lookup
+source = table | fields region, host, val | outputlookup key_field=region, host my_lookup
+source = table | outputlookup max=1000 sample_lookup
+```
+
+## Parameters
+
+The `outputlookup` command supports the following parameters.
+
+| Parameter | Required/Optional | Description |
+| --- | --- | --- |
+| `<name>` | Required | The lookup name to write to. It is a plain index, created on demand if it does not exist and replaced on overwrite. Lookup indices are tagged with a `_meta.lookup` marker; if the name is an existing **non-lookup** index (no marker), the command refuses rather than overwriting it — delete it first to reuse the name. If the name is a filtered alias created by the OpenSearch Dashboards data importer, overwrite migrates it onto a dedicated plain index (the shared index is never deleted). |
+| `append` | Optional | When `false` (default), overwrites the lookup with the result. When `true`, appends the result to the existing lookup. Because OpenSearch is schemaless, appended rows may introduce new fields. Default is `false`. |
+| `override_if_empty` | Optional | When `true` (default), an empty result clears the existing lookup. When `false`, an empty result leaves the existing lookup intact. Default is `true`. |
+| `key_field` | Optional | One or more fields (comma-separated) used as the upsert key. Rows are written by a deterministic `_id` derived from the key values, so re-running the same command updates matching rows in place instead of creating duplicates. Setting `key_field` implies `append=true`. Every field listed must be a field of the result; a multivalue key value is rejected. |
+| `max` | Optional | Caps the number of rows written. |
+
+## Example 1: Building a lookup from an aggregation
+
+The following query writes per-region counts into a lookup and returns the number of rows written:
+
+```ppl ignore
+source = events
+  | stats count as cnt by region
+  | outputlookup region_counts
+```
+
+The query returns the following result:
+
+```text
++--------------+
+| rows_written |
+|--------------|
+| 3            |
++--------------+
+```
+
+The lookup can then be read back:
+
+```ppl ignore
+source = region_counts
+```
+
+## Example 2: Idempotent upsert with `key_field`
+
+The following query upserts by `id`, so re-running it updates existing rows instead of duplicating them:
+
+```ppl ignore
+source = users
+  | fields id, name, department
+  | outputlookup key_field=id users_lookup
+```
+
+Running the command a second time with overlapping `id` values leaves the row count unchanged for the overlapping keys and only inserts genuinely new keys.
+
+## Example 3: Preserving a lookup on an empty result
+
+The following query refreshes `error_hosts` only when the search returns rows; an empty result keeps the previous lookup intact:
+
+```ppl ignore
+source = logs
+  | where level = 'ERROR'
+  | fields host
+  | outputlookup override_if_empty=false error_hosts
+```
+
+## Limitations
+
+- `outputlookup` is a terminal command: it returns a `rows_written` count rather than forwarding the input rows.
+- The destination is always a lookup index; there is no file output target.
+- Overwrite is weak/eventually consistent but gap-free: it writes a fresh slice and atomically repoints the alias, so a concurrent read always sees the whole old slice or the whole new one, never a partial state. The previous slice is orphaned until reclaimed.
+- `outputlookup` is for bounded lookup (dimension) tables, not a bulk data sink. The number of rows a single write may produce is capped by `plugins.ppl.outputlookup.max_rows` (NodeScope, Dynamic, default `1000000`, minimum `1`). Exceeding it fails the query (no slice is written) rather than silently truncating. The optional `max=<int>` command argument truncates to at most N rows and must not exceed the setting. To write a larger lookup you can raise `plugins.ppl.outputlookup.max_rows` (it is dynamic; a higher ceiling costs more coordinator heap per write), or ingest large data through a bulk indexing path instead.
+- The write executes under the caller's security context. The caller needs write and get privileges on the destination and alias privileges to publish or repoint the filtered alias, plus create-index the first time the shared `.lookups` index is created; no cluster-level privilege is required.