Add PPL multikv command (fixed-schema)

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

@@ -83,6 +83,7 @@
 import org.opensearch.sql.ast.tree.Lookup;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.MakeResults;
+import org.opensearch.sql.ast.tree.Multikv;
 import org.opensearch.sql.ast.tree.Multisearch;
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
@@ -569,6 +570,11 @@ public LogicalPlan visitMakeResults(MakeResults node, AnalysisContext context) {
     throw getOnlyForCalciteException("makeresults");
   }
 
+  @Override
+  public LogicalPlan visitMultikv(Multikv node, AnalysisContext context) {
+    throw getOnlyForCalciteException("multikv");
+  }
+
   @Override
   public LogicalPlan visitMvExpand(MvExpand node, AnalysisContext context) {
     throw getOnlyForCalciteException("mvexpand");

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

@@ -72,6 +72,7 @@
 import org.opensearch.sql.ast.tree.Lookup;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.MakeResults;
+import org.opensearch.sql.ast.tree.Multikv;
 import org.opensearch.sql.ast.tree.Multisearch;
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
@@ -518,6 +519,10 @@ public T visitMvExpand(MvExpand node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitMultikv(Multikv node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitGraphLookup(GraphLookup node, C context) {
     return visitChildren(node, context);
   }

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

@@ -0,0 +1,108 @@
+/*
+ * 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 javax.annotation.Nullable;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.expression.Field;
+import org.opensearch.sql.data.type.ExprCoreType;
+
+/**
+ * AST node representing the {@code multikv} PPL command.
+ *
+ * <p>{@code multikv} extracts field values from an input field (default {@code _raw}) and emits one
+ * row per source row. The input field is either table-formatted text (split into columns) or an
+ * array of objects (one row per element, each declared column read from the element). This is a
+ * one-to-many (row-multiplying) command.
+ *
+ * <p>The output column names must be determinable at plan time, sourced from the declared {@link
+ * #fields} list, a literal {@link #forceHeader} line, or positional naming when {@link #noHeader}
+ * is set. Runtime header auto-detection (no fields, no forceheader, no noheader) is not supported;
+ * such a query is rejected at the field-resolution phase with a message directing the author to add
+ * a {@code fields} clause.
+ */
+@ToString
+@EqualsAndHashCode(callSuper = false)
+@Getter
+public class Multikv extends UnresolvedPlan {
+
+  /** Default input field name used when {@code field=} is omitted. */
+  public static final String DEFAULT_INPUT_FIELD = "_raw";
+
+  private UnresolvedPlan child;
+
+  /** Input field carrying the table text. Defaults to {@code _raw}. */
+  private final String inField;
+
+  /** Declared output columns (the {@code fields} option). Null when not declared. */
+  @Nullable private final List<Field> fields;
+
+  /**
+   * Per-column declared types aligned with {@link #fields} (the {@code col:type} syntax). Null when
+   * none declared.
+   */
+  @Nullable private final List<ExprCoreType> fieldTypes;
+
+  /** Filter terms; a table row is kept only if it contains at least one term. Null when absent. */
+  @Nullable private final List<String> filterTerms;
+
+  /** 1-based header line to force (the {@code forceheader} option). Null when absent. */
+  @Nullable private final Integer forceHeader;
+
+  /** When true, columns are named positionally (Column_1, Column_2, ...). */
+  private final boolean noHeader;
+
+  /** When true (default), the original event is dropped from the output. */
+  private final boolean rmOrig;
+
+  public Multikv(
+      String inField,
+      @Nullable List<Field> fields,
+      @Nullable List<String> filterTerms,
+      @Nullable Integer forceHeader,
+      boolean noHeader,
+      boolean rmOrig) {
+    this(inField, fields, null, filterTerms, forceHeader, noHeader, rmOrig);
+  }
+
+  public Multikv(
+      String inField,
+      @Nullable List<Field> fields,
+      @Nullable List<ExprCoreType> fieldTypes,
+      @Nullable List<String> filterTerms,
+      @Nullable Integer forceHeader,
+      boolean noHeader,
+      boolean rmOrig) {
+    this.inField = inField;
+    this.fields = fields;
+    this.fieldTypes = fieldTypes;
+    this.filterTerms = filterTerms;
+    this.forceHeader = forceHeader;
+    this.noHeader = noHeader;
+    this.rmOrig = rmOrig;
+  }
+
+  @Override
+  public Multikv 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.visitMultikv(this, context);
+  }
+}

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

@@ -143,6 +143,7 @@
 import org.opensearch.sql.ast.tree.Lookup.OutputStrategy;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.MakeResults;
+import org.opensearch.sql.ast.tree.Multikv;
 import org.opensearch.sql.ast.tree.Multisearch;
 import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.MvExpand;
@@ -199,6 +200,7 @@
 import org.opensearch.sql.expression.function.BuiltinFunctionName;
 import org.opensearch.sql.expression.function.PPLBuiltinOperators;
 import org.opensearch.sql.expression.function.PPLFuncImpTable;
+import org.opensearch.sql.expression.function.multikv.MultikvParser;
 import org.opensearch.sql.expression.parse.RegexCommonUtils;
 import org.opensearch.sql.utils.ParseUtils;
 import org.opensearch.sql.utils.WildcardRenameUtils;
@@ -4663,6 +4665,182 @@ public RelNode visitMvExpand(MvExpand mvExpand, CalcitePlanContext context) {
     return relBuilder.peek();
   }
 
+  /**
+   * multikv (fixed-schema): rewrite to an equivalent pipeline using existing operators.
+   *
+   * <pre>
+   *   eval __multikv_record__ = MULTIKV_SPLIT(inField, forceHeader, noHeader, filter)  // array&lt;varchar&gt;
+   *   | mvexpand __multikv_record__                                                     // 1 -&gt; N rows
+   *   | eval &lt;col&gt; = MULTIKV_EXTRACT(__multikv_record__, '&lt;col&gt;')  for each declared field
+   *   | fields &lt;col1&gt;, &lt;col2&gt;, ...                                                      // declared output schema
+   * </pre>
+   *
+   * The output column names come from the declared {@code fields} list and are therefore known at
+   * plan time. When no {@code fields} clause is declared, the output schema is not determinable at
+   * plan time and is rejected with guidance to add a {@code fields} clause.
+   *
+   * <p>When the input field is an object or an array of objects instead of text, the command
+   * dispatches to a native rewrite: {@code mvexpand} an array (a single object needs no explosion),
+   * then read each declared column from the object with {@code ITEM}. The extracted columns are
+   * typed {@code ANY}: object and nested fields collapse to ANY-valued containers in the type
+   * layer, so the mapped scalar type is not recovered (cast downstream). Shares the {@code fields}
+   * contract.
+   */
+  @Override
+  public RelNode visitMultikv(Multikv node, CalcitePlanContext context) {
+    List<Field> fields = node.getFields();
+    boolean noFields = (fields == null || fields.isEmpty());
+
+    // Fixed-schema: output columns must come from the fields clause. The only supported no-fields
+    // form is positional noheader (row-explosion, no named columns). Any other no-fields form
+    // (bare auto-header, or forceheader without fields) has no plan-time schema and is rejected.
+    if (noFields && !node.isNoHeader()) {
+      throw ErrorReport.wrap(
+              new SemanticCheckException(
+                  "multikv has no declared output columns. Add an explicit fields clause, for"
+                      + " example: multikv fields <col1> <col2>"))
+          .code(ErrorCode.FIELD_NOT_FOUND)
+          .location("while resolving the output schema for multikv")
+          .context("command", "multikv")
+          .build();
+    }
+
+    // Dispatch on the input field's type. The child is built once here and both branches lower
+    // directly onto that build. A structured (array of objects, or a single object) input is
+    // exploded with mvexpand and each declared column is read with ITEM (typed ANY, since element
+    // types are erased upstream). A text input runs the split pipeline below on the same build.
+    RelNode probe = node.getChild().get(0).accept(this, context);
+    RelDataTypeField probeField = probe.getRowType().getField(node.getInField(), true, false);
+    boolean structuredArray =
+        probeField != null
+            && (SqlTypeUtil.isArray(probeField.getType())
+                || SqlTypeUtil.isMultiset(probeField.getType()));
+    boolean structuredMap = probeField != null && SqlTypeUtil.isMap(probeField.getType());
+    if (structuredArray || structuredMap) {
+      RelBuilder relBuilder = context.relBuilder;
+      if (structuredArray) {
+        // Array of objects: one row per element. A single object (map) needs no explosion.
+        buildExpandRelNode(
+            relBuilder.field(node.getInField()),
+            node.getInField(),
+            node.getInField(),
+            null,
+            context);
+      }
+      if (noFields) {
+        return relBuilder.peek();
+      }
+      List<ExprCoreType> fieldTypes = node.getFieldTypes();
+      List<RexNode> projected = new ArrayList<>();
+      List<String> names = new ArrayList<>();
+      for (int i = 0; i < fields.size(); i++) {
+        Field f = fields.get(i);
+        String col = f.getField().toString();
+        RexNode item =
+            PPLFuncImpTable.INSTANCE.resolve(
+                context.rexBuilder,
+                BuiltinFunctionName.INTERNAL_ITEM,
+                relBuilder.field(node.getInField()),
+                context.rexBuilder.makeLiteral(
+                    col,
+                    context.rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR),
+                    true));
+        ExprCoreType declared = fieldTypes == null ? null : fieldTypes.get(i);
+        if (declared != null) {
+          item =
+              context.rexBuilder.makeCast(
+                  OpenSearchTypeFactory.convertExprTypeToRelDataType(declared), item, true, true);
+        }
+        projected.add(item);
+        names.add(col);
+      }
+      relBuilder.project(projected, names);
+      context.setProjectVisited(true);
+      return relBuilder.peek();
+    }
+    // Reject input types multikv cannot read. The structured branch above handles array/object
+    // fields and text mode splits string values, so a scalar numeric/boolean/date field has no
+    // table text to parse; reject it at plan time instead of silently treating it as text. An
+    // untyped ANY field is allowed, since its runtime value may be text.
+    if (probeField != null) {
+      SqlTypeName inputTypeName = probeField.getType().getSqlTypeName();
+      boolean textLike = inputTypeName == SqlTypeName.VARCHAR || inputTypeName == SqlTypeName.CHAR;
+      boolean untyped = inputTypeName == SqlTypeName.ANY || inputTypeName == SqlTypeName.NULL;
+      if (!textLike && !untyped) {
+        throw new SemanticCheckException(
+            "multikv input field '"
+                + node.getInField()
+                + "' has type "
+                + inputTypeName
+                + "; multikv reads table-formatted text or an array/object field. Cast it to a"
+                + " string first, for example: eval "
+                + node.getInField()
+                + " = cast("
+                + node.getInField()
+                + " as string).");
+      }
+    }
+    // Text input: lower the split pipeline directly onto the probe build above, so the child is
+    // visited exactly once. eval __multikv_record__ = MULTIKV_SPLIT(inField, ...), explode it with
+    // mvexpand (one row per table data row), then read the declared columns with MULTIKV_EXTRACT.
+    RelBuilder relBuilder = context.relBuilder;
+    final String lineField = "__multikv_record__";
+    final int forceHeader = node.getForceHeader() == null ? -1 : node.getForceHeader();
+    final String filterJoined =
+        (node.getFilterTerms() == null || node.getFilterTerms().isEmpty())
+            ? ""
+            : String.join(MultikvParser.FS, node.getFilterTerms());
+
+    RexNode split =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder,
+            BuiltinFunctionName.MULTIKV_SPLIT,
+            relBuilder.field(node.getInField()),
+            relBuilder.literal(forceHeader),
+            relBuilder.literal(node.isNoHeader()),
+            context.rexBuilder.makeLiteral(
+                filterJoined,
+                context.rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR),
+                true));
+    relBuilder.projectPlus(relBuilder.alias(split, lineField));
+
+    // mvexpand the record column: 1 -> N rows. Mirrors visitMvExpand (alias == field name).
+    buildExpandRelNode(relBuilder.field(lineField), lineField, lineField, null, context);
+
+    if (noFields) {
+      // Positional noheader, no named columns: row-explosion only. The helper record column is
+      // retained (downstream typically only counts rows). Naming positional columns is deferred.
+      return relBuilder.peek();
+    }
+
+    List<ExprCoreType> fieldTypes = node.getFieldTypes();
+    List<RexNode> projected = new ArrayList<>();
+    List<String> names = new ArrayList<>();
+    for (int i = 0; i < fields.size(); i++) {
+      String col = fields.get(i).getField().toString();
+      RexNode extract =
+          PPLFuncImpTable.INSTANCE.resolve(
+              context.rexBuilder,
+              BuiltinFunctionName.MULTIKV_EXTRACT,
+              relBuilder.field(lineField),
+              context.rexBuilder.makeLiteral(
+                  col,
+                  context.rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR),
+                  true));
+      ExprCoreType declared = fieldTypes == null ? null : fieldTypes.get(i);
+      if (declared != null) {
+        extract =
+            context.rexBuilder.makeCast(
+                OpenSearchTypeFactory.convertExprTypeToRelDataType(declared), extract, true, true);
+      }
+      projected.add(extract);
+      names.add(col);
+    }
+    relBuilder.project(projected, names);
+    context.setProjectVisited(true);
+    return relBuilder.peek();
+  }
+
   @Override
   public RelNode visitValues(Values values, CalcitePlanContext context) {
     List<List<Literal>> rows = values.getValues();

core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java

@@ -272,6 +272,8 @@ public enum BuiltinFunctionName {
   JSON_ARRAY_LENGTH(FunctionName.of("json_array_length")),
   JSON_EXTRACT(FunctionName.of("json_extract")),
   JSON_EXTRACT_ALL(FunctionName.of("json_extract_all"), true),
+  MULTIKV_SPLIT(FunctionName.of("multikv_split"), true),
+  MULTIKV_EXTRACT(FunctionName.of("multikv_extract"), true),
   JSON_KEYS(FunctionName.of("json_keys")),
   JSON_SET(FunctionName.of("json_set")),
   JSON_DELETE(FunctionName.of("json_delete")),

core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java

@@ -144,6 +144,14 @@ public class PPLBuiltinOperators extends ReflectiveSqlOperatorTable {
   public static final SqlOperator JSON_APPEND = new JsonAppendFunctionImpl().toUDF("JSON_APPEND");
   public static final SqlOperator JSON_EXTEND = new JsonExtendFunctionImpl().toUDF("JSON_EXTEND");
 
+  // multikv internal functions
+  public static final SqlOperator MULTIKV_SPLIT =
+      new org.opensearch.sql.expression.function.multikv.MultikvSplitFunctionImpl()
+          .toUDF("MULTIKV_SPLIT");
+  public static final SqlOperator MULTIKV_EXTRACT =
+      new org.opensearch.sql.expression.function.multikv.MultikvExtractFunctionImpl()
+          .toUDF("MULTIKV_EXTRACT");
+
   // Math functions
   public static final SqlOperator SPAN = new SpanFunction().toUDF("SPAN");
   public static final SqlOperator E = new EulerFunction().toUDF("E");

core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java

@@ -163,6 +163,8 @@
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MONTHNAME;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MONTH_OF_YEAR;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MSTIME;
+import static org.opensearch.sql.expression.function.BuiltinFunctionName.MULTIKV_EXTRACT;
+import static org.opensearch.sql.expression.function.BuiltinFunctionName.MULTIKV_SPLIT;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MULTIMATCH;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MULTIMATCHQUERY;
 import static org.opensearch.sql.expression.function.BuiltinFunctionName.MULTIPLY;
@@ -1335,6 +1337,8 @@ void populate() {
       registerOperator(JSON_APPEND, PPLBuiltinOperators.JSON_APPEND);
       registerOperator(JSON_EXTEND, PPLBuiltinOperators.JSON_EXTEND);
       registerOperator(JSON_EXTRACT_ALL, PPLBuiltinOperators.JSON_EXTRACT_ALL); // internal
+      registerOperator(MULTIKV_SPLIT, PPLBuiltinOperators.MULTIKV_SPLIT); // internal
+      registerOperator(MULTIKV_EXTRACT, PPLBuiltinOperators.MULTIKV_EXTRACT); // internal
 
       // Register operators with a different type checker