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,89 @@
+/*
+ * 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;
+
+/**
+ * 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 Splunk input field for multikv. */
+  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;
+
+  /** 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 = inField;
+    this.fields = fields;
+    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,143 @@ 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. A structured (array of objects) input is exploded
+    // natively with mvexpand and each declared column is read with ITEM; the extracted column is
+    // typed ANY (element types are erased to ANY upstream), not the mapped scalar type. A text
+    // input falls through to the split pipeline below. The child is built once here to read
+    // its schema; the structured branch reuses that build, the text branch discards it.
+    boolean savedProjectVisited = context.isProjectVisited();
+    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<RexNode> projected = new ArrayList<>();
+      List<String> names = new ArrayList<>();
+      for (Field f : fields) {
+        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));
+        projected.add(item);
+        names.add(col);
+      }
+      relBuilder.project(projected, names);
+      context.setProjectVisited(true);
+      return relBuilder.peek();
+    }
+    // Text input: discard the probe build and run the split pipeline on a fresh build.
+    context.relBuilder.build();
+    context.setProjectVisited(savedProjectVisited);
+
+    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());
+
+    UnresolvedPlan plan =
+        AstDSL.eval(
+            node.getChild().get(0),
+            AstDSL.let(
+                AstDSL.field(lineField),
+                AstDSL.function(
+                    "multikv_split",
+                    AstDSL.field(node.getInField()),
+                    AstDSL.intLiteral(forceHeader),
+                    AstDSL.booleanLiteral(node.isNoHeader()),
+                    AstDSL.stringLiteral(filterJoined))));
+
+    plan = new MvExpand(AstDSL.field(lineField), null).attach(plan);
+
+    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 plan.accept(this, context);
+    }
+
+    Let[] lets =
+        fields.stream()
+            .map(
+                f -> {
+                  String col = f.getField().toString();
+                  return AstDSL.let(
+                      AstDSL.field(col),
+                      AstDSL.function(
+                          "multikv_extract", AstDSL.field(lineField), AstDSL.stringLiteral(col)));
+                })
+            .toArray(Let[]::new);
+    plan = AstDSL.eval(plan, lets);
+
+    UnresolvedExpression[] projections = fields.toArray(new UnresolvedExpression[0]);
+    plan = AstDSL.project(plan, projections);
+
+    return plan.accept(this, context);
+  }
+
   @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
 

core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvExtractFunctionImpl.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.expression.function.multikv;
+
+import static org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY;
+
+import java.util.List;
+import org.apache.calcite.adapter.enumerable.NotNullImplementor;
+import org.apache.calcite.adapter.enumerable.NullPolicy;
+import org.apache.calcite.adapter.enumerable.RexImpTable;
+import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Types;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.schema.impl.ScalarFunctionImpl;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.sql.expression.function.ImplementorUDF;
+import org.opensearch.sql.expression.function.UDFOperandMetadata;
+
+/**
+ * Internal UDF backing the {@code multikv} command. Extracts a single named cell value out of one
+ * serialized per-row record produced by {@link MultikvSplitFunctionImpl}.
+ *
+ * <p>Signature: {@code MULTIKV_EXTRACT(recordString, columnName)} returns {@code varchar} (null
+ * when the column is absent from the record).
+ */
+public class MultikvExtractFunctionImpl extends ImplementorUDF {
+
+  public MultikvExtractFunctionImpl() {
+    super(new MultikvExtractImplementor(), NullPolicy.ANY);
+  }
+
+  @Override
+  public SqlReturnTypeInference getReturnTypeInference() {
+    return ReturnTypes.explicit(
+        TYPE_FACTORY.createTypeWithNullability(
+            TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR), true));
+  }
+
+  @Override
+  public UDFOperandMetadata getOperandMetadata() {
+    return null;
+  }
+
+  public static Object eval(Object... args) {
+    if (args.length < 2 || args[0] == null || args[1] == null) {
+      return null;
+    }
+    return MultikvParser.extract((String) args[0], (String) args[1]);
+  }
+
+  public static class MultikvExtractImplementor implements NotNullImplementor {
+    @Override
+    public Expression implement(
+        RexToLixTranslator translator, RexCall call, List<Expression> translatedOperands) {
+      ScalarFunctionImpl function =
+          (ScalarFunctionImpl)
+              ScalarFunctionImpl.create(
+                  Types.lookupMethod(MultikvExtractFunctionImpl.class, "eval", Object[].class));
+      return function.getImplementor().implement(translator, call, RexImpTable.NullAs.NULL);
+    }
+  }
+}