[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

@@ -87,6 +87,7 @@
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
 import org.opensearch.sql.ast.tree.NoMv;
+import org.opensearch.sql.ast.tree.OutputLookup;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -563,6 +564,11 @@ public LogicalPlan visitNoMv(NoMv node, AnalysisContext context) {
     throw getOnlyForCalciteException("nomv");
   }
 
+  @Override
+  public LogicalPlan visitOutputLookup(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

@@ -76,6 +76,7 @@
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
 import org.opensearch.sql.ast.tree.NoMv;
+import org.opensearch.sql.ast.tree.OutputLookup;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -305,6 +306,10 @@ public T visitReverse(Reverse node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitOutputLookup(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,61 @@
+/*
+ * 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, overwrite by default. */
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@RequiredArgsConstructor
+@AllArgsConstructor
+public class OutputLookup extends UnresolvedPlan {
+
+  private UnresolvedPlan child;
+
+  /** The lookup name: the filtered alias published over the backing index. */
+  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 = 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

@@ -51,6 +51,7 @@
 import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.plan.ViewExpanders;
+import org.apache.calcite.prepare.Prepare;
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelHomogeneousShuttle;
@@ -146,6 +147,7 @@
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
 import org.opensearch.sql.ast.tree.NoMv;
+import org.opensearch.sql.ast.tree.OutputLookup;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -926,6 +928,58 @@ public RelNode visitReverse(
     return context.relBuilder.peek();
   }
 
+  @Override
+  public RelNode visitOutputLookup(OutputLookup node, CalcitePlanContext context) {
+    visitChildren(node, context);
+    String name = node.getIndexName();
+    if (name.startsWith(".")) {
+      throw new IllegalArgumentException(
+          "outputlookup lookup name [" + name + "] must not be dot-prefixed");
+    }
+    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()) {
+      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.RelOptSchema relOptSchema = context.relBuilder.getRelOptSchema();
+    if (!(relOptSchema instanceof Prepare.CatalogReader catalogReader)) {
+      throw new IllegalStateException("outputlookup could not obtain a Calcite catalog reader");
+    }
+    // Resolve a table from the schema (not a scan) so sourceless pipelines reach the client; any
+    // OpenSearch name yields the same client, and the supplied row type avoids a mapping fetch.
+    OpenSearchSchema openSearchSchema =
+        context.config.getDefaultSchema().unwrap(OpenSearchSchema.class);
+    org.apache.calcite.schema.Table clientTable = openSearchSchema.getTable(name);
+    RelOptTable destTable =
+        org.apache.calcite.prepare.RelOptTableImpl.create(
+            relOptSchema,
+            child.getRowType(),
+            clientTable,
+            ImmutableList.of(OpenSearchSchema.OPEN_SEARCH_SCHEMA_NAME, name));
+    RelNode sink =
+        OutputLookupTableModify.create(
+            child,
+            destTable,
+            catalogReader,
+            name,
+            node.isAppend(),
+            node.isOverrideIfEmpty(),
+            node.getKeyFields(),
+            node.getMax());
+    context.relBuilder.push(sink);
+    return context.relBuilder.peek();
+  }
+
   @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,115 @@
+/*
+ * 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.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.opensearch.sql.calcite.plan.AbstractOpenSearchTable;
+
+/**
+ * Terminal write node for outputlookup, a Calcite {@link TableModify} INSERT so the optimizer
+ * treats it as a mandatory side effect (never dropped or reordered). The referenced table is the
+ * destination backing index, used only so the lowering rule can reach the in-cluster client; the
+ * lookup is created and written at execution time. Subclasses {@link TableModify} rather than
+ * LogicalTableModify so the built-in EnumerableTableModifyRule (which needs a ModifiableTable) does
+ * not fire.
+ */
+@Getter
+public class OutputLookupTableModify extends TableModify {
+
+  private final String indexName;
+  private final boolean append;
+  private final boolean overrideIfEmpty;
+  private final 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 RelDataType deriveRowType() {
+    RelDataTypeFactory typeFactory = getCluster().getTypeFactory();
+    return typeFactory
+        .builder()
+        .add("rows_written", typeFactory.createSqlType(SqlTypeName.BIGINT))
+        .build();
+  }
+
+  /** Registers the write-lowering rule when no OpenSearch scan in the pipeline does. */
+  @Override
+  public void register(org.apache.calcite.plan.RelOptPlanner planner) {
+    AbstractOpenSearchTable osTable = table.unwrap(AbstractOpenSearchTable.class);
+    if (osTable != null) {
+      osTable.getWriteConversionRules().forEach(planner::addRule);
+    }
+  }
+
+  @Override
+  public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+    return new OutputLookupTableModify(
+        getCluster(),
+        traitSet,
+        getTable(),
+        getCatalogReader(),
+        inputs.get(0),
+        indexName,
+        append,
+        overrideIfEmpty,
+        keyFields,
+        max);
+  }
+}

core/src/main/java/org/opensearch/sql/calcite/plan/AbstractOpenSearchTable.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.sql.calcite.plan;
 
+import java.util.List;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.schema.TranslatableTable;
@@ -22,4 +23,11 @@ public abstract class AbstractOpenSearchTable extends AbstractTable
   public RelDataType getRowType(RelDataTypeFactory relDataTypeFactory) {
     return OpenSearchTypeFactory.convertSchema(this);
   }
+
+  /**
+   * Write-lowering rules a write node registers itself when the pipeline has no OpenSearch scan.
+   */
+  public java.util.List<org.apache.calcite.plan.RelOptRule> getWriteConversionRules() {
+    return List.of();
+  }
 }

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",