diff --git a/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java b/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java
index 701d1545b7..67ae51e468 100644
--- a/core/src/main/java/org/opensearch/sql/analysis/Analyzer.java
+++ b/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;
@@ -568,6 +569,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");
diff --git a/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java b/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java
index acb6e10566..12591af1dd 100644
--- a/core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java
+++ b/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;
@@ -517,6 +518,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);
}
diff --git a/core/src/main/java/org/opensearch/sql/ast/tree/Multikv.java b/core/src/main/java/org/opensearch/sql/ast/tree/Multikv.java
new file mode 100644
index 0000000000..a8d7a76c68
--- /dev/null
+++ b/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.
+ *
+ * {@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.
+ *
+ *
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 fields;
+
+ /** Filter terms; a table row is kept only if it contains at least one term. Null when absent. */
+ @Nullable private final List 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 fields,
+ @Nullable List 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 getChild() {
+ return this.child == null ? ImmutableList.of() : ImmutableList.of(this.child);
+ }
+
+ @Override
+ public T accept(AbstractNodeVisitor nodeVisitor, C context) {
+ return nodeVisitor.visitMultikv(this, context);
+ }
+}
diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
index 94c40e5adb..a23c0af721 100644
--- a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
+++ b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
@@ -142,6 +142,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;
@@ -197,6 +198,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;
@@ -4469,6 +4471,146 @@ public RelNode visitMvExpand(MvExpand mvExpand, CalcitePlanContext context) {
return relBuilder.peek();
}
+ /**
+ * multikv (fixed-schema): rewrite to an equivalent pipeline using existing operators.
+ *
+ *
+ * eval __multikv_record__ = MULTIKV_SPLIT(inField, forceHeader, noHeader, filter) // array<varchar>
+ * | mvexpand __multikv_record__ // 1 -> N rows
+ * | eval <col> = MULTIKV_EXTRACT(__multikv_record__, '<col>') for each declared field
+ * | fields <col1>, <col2>, ... // declared output schema
+ *
+ *
+ * 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.
+ *
+ * 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}. This preserves value types
+ * and shares the {@code fields} contract.
+ */
+ @Override
+ public RelNode visitMultikv(Multikv node, CalcitePlanContext context) {
+ List fields = node.getFields();
+ boolean noFields = (fields == null || fields.isEmpty());
+
+ // Fixed-schema: named output columns must be declared via the fields clause. The only
+ // no-fields form that is supported is positional noheader, which produces row-explosion with
+ // no named columns (e.g. `... | multikv noheader=true | stats count`). Any other no-fields
+ // form (bare auto-header, or forceheader without fields) cannot yield a plan-time schema and
+ // is rejected here with guidance, rather than failing at execution.
+ if (noFields && !node.isNoHeader()) {
+ throw ErrorReport.wrap(
+ new SemanticCheckException(
+ "multikv has no declared output columns. Add an explicit fields clause, for"
+ + " example: multikv fields "))
+ .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, preserving element types.
+ // 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 projected = new ArrayList<>();
+ List 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());
+
+ // 1. eval __multikv_record__ = MULTIKV_SPLIT(inField, forceHeader, noHeader, filter)
+ 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))));
+
+ // 2. mvexpand __multikv_record__ (one row per table data row)
+ 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);
+ }
+
+ // 3. eval each declared column = MULTIKV_EXTRACT(__multikv_record__, '')
+ 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);
+
+ // 4. project the declared columns only (drops original event fields and the helper record)
+ 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> rows = values.getValues();
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java b/core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java
index e30d723ccf..ed023f49b0 100644
--- a/core/src/main/java/org/opensearch/sql/expression/function/BuiltinFunctionName.java
+++ b/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")),
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java b/core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java
index d64f04bb9a..83400e5f52 100644
--- a/core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java
+++ b/core/src/main/java/org/opensearch/sql/expression/function/PPLBuiltinOperators.java
@@ -141,6 +141,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");
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java b/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java
index 151c4a9665..010bd0c2b6 100644
--- a/core/src/main/java/org/opensearch/sql/expression/function/PPLFuncImpTable.java
+++ b/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
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvExtractFunctionImpl.java b/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvExtractFunctionImpl.java
new file mode 100644
index 0000000000..f319e97fe2
--- /dev/null
+++ b/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}.
+ *
+ * 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 translatedOperands) {
+ ScalarFunctionImpl function =
+ (ScalarFunctionImpl)
+ ScalarFunctionImpl.create(
+ Types.lookupMethod(MultikvExtractFunctionImpl.class, "eval", Object[].class));
+ return function.getImplementor().implement(translator, call, RexImpTable.NullAs.NULL);
+ }
+ }
+}
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java b/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java
new file mode 100644
index 0000000000..329467d7db
--- /dev/null
+++ b/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvParser.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.expression.function.multikv;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Pattern;
+
+/**
+ * Shared parsing logic for the {@code multikv} command UDFs.
+ *
+ * {@link #parse} turns the table text of a single event into a list of serialized records, one
+ * per table data row. Each record encodes the row's cells as {@code namevalue} pairs joined by
+ * {@code }, where the column name comes from the detected header (or positional {@code
+ * Column_N} naming under {@code noheader}). {@link #extract} looks a single column value up out of
+ * one serialized record.
+ *
+ * Columns are split on runs of whitespace. Richer aligned-offset handling is a later
+ * enhancement.
+ */
+public final class MultikvParser {
+
+ /** Field separator between cells inside a serialized record. */
+ public static final String FS = "\u001f";
+
+ /** Key/value separator between a column name and its value. */
+ public static final String KV = "\u0002";
+
+ private static final Pattern WS = Pattern.compile("\\s+");
+ private static final Pattern LINE = Pattern.compile("\\R");
+ private static final Pattern FS_SPLIT = Pattern.compile(Pattern.quote(FS));
+
+ private MultikvParser() {}
+
+ /**
+ * Parse the table text into serialized per-row records.
+ *
+ * @param raw the table text (typically the {@code _raw} field)
+ * @param forceHeader 1-based header line to force, or a value <= 0 for auto-detection
+ * @param noHeader when true, columns are named positionally ({@code Column_1}, ...)
+ * @param filterTerms keep only rows containing at least one term; empty means keep all
+ */
+ public static List parse(
+ String raw, int forceHeader, boolean noHeader, List filterTerms) {
+ if (raw == null) {
+ return Collections.emptyList();
+ }
+ List lines = new ArrayList<>();
+ for (String l : LINE.split(raw)) {
+ if (!l.trim().isEmpty()) {
+ lines.add(l);
+ }
+ }
+ if (lines.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List header;
+ int dataStart;
+ if (noHeader) {
+ header = null;
+ dataStart = 0;
+ } else if (forceHeader > 0) {
+ int hIdx = Math.min(forceHeader - 1, lines.size() - 1);
+ header = splitCols(lines.get(hIdx));
+ dataStart = hIdx + 1;
+ } else {
+ header = splitCols(lines.get(0));
+ dataStart = 1;
+ }
+
+ boolean hasFilter = filterTerms != null && !filterTerms.isEmpty();
+ List out = new ArrayList<>();
+ for (int i = dataStart; i < lines.size(); i++) {
+ String line = lines.get(i);
+ if (hasFilter && !matchesFilter(line, filterTerms)) {
+ continue;
+ }
+ out.add(serialize(header, splitCols(line)));
+ }
+ return out;
+ }
+
+ /** Extract the value for {@code col} out of one serialized record, or null when absent. */
+ public static String extract(String record, String col) {
+ if (record == null || col == null) {
+ return null;
+ }
+ for (String pair : FS_SPLIT.split(record, -1)) {
+ int idx = pair.indexOf(KV);
+ if (idx < 0) {
+ continue;
+ }
+ if (pair.substring(0, idx).equals(col)) {
+ return pair.substring(idx + 1);
+ }
+ }
+ return null;
+ }
+
+ private static boolean matchesFilter(String line, List filterTerms) {
+ for (String t : filterTerms) {
+ if (t != null && !t.isEmpty() && line.contains(t)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static List splitCols(String line) {
+ String t = line.trim();
+ if (t.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return Arrays.asList(WS.split(t));
+ }
+
+ private static String serialize(List header, List cells) {
+ int n = (header != null) ? Math.max(header.size(), cells.size()) : cells.size();
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < n; i++) {
+ String name = (header != null && i < header.size()) ? header.get(i) : ("Column_" + (i + 1));
+ String val = (i < cells.size()) ? cells.get(i) : "";
+ if (sb.length() > 0) {
+ sb.append(FS);
+ }
+ sb.append(name).append(KV).append(val);
+ }
+ return sb.toString();
+ }
+}
diff --git a/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvSplitFunctionImpl.java b/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvSplitFunctionImpl.java
new file mode 100644
index 0000000000..f8d6736368
--- /dev/null
+++ b/core/src/main/java/org/opensearch/sql/expression/function/multikv/MultikvSplitFunctionImpl.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.expression.function.multikv;
+
+import static org.apache.calcite.sql.type.SqlTypeUtil.createArrayType;
+
+import java.util.Arrays;
+import java.util.Collections;
+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.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.schema.impl.ScalarFunctionImpl;
+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. Parses the table text of one event into an
+ * array of serialized per-row records (see {@link MultikvParser}). The array is then exploded into
+ * one row per table data row (via the mvexpand uncollect/correlate primitive), after which {@link
+ * MultikvExtractFunctionImpl} pulls named cell values out of each record.
+ *
+ * Signature: {@code MULTIKV_SPLIT(rawText, forceHeaderInt, noHeaderBool, filterJoined)} where
+ * {@code forceHeaderInt} is the 1-based forced header line or a value <= 0 for auto, and {@code
+ * filterJoined} is the filter terms joined by {@link MultikvParser#FS} (empty for none). Returns
+ * {@code array}.
+ */
+public class MultikvSplitFunctionImpl extends ImplementorUDF {
+
+ public MultikvSplitFunctionImpl() {
+ super(new MultikvSplitImplementor(), NullPolicy.ANY);
+ }
+
+ @Override
+ public SqlReturnTypeInference getReturnTypeInference() {
+ return sqlOperatorBinding -> {
+ RelDataTypeFactory typeFactory = sqlOperatorBinding.getTypeFactory();
+ return createArrayType(
+ typeFactory,
+ typeFactory.createTypeWithNullability(
+ typeFactory.createSqlType(SqlTypeName.VARCHAR), true),
+ true);
+ };
+ }
+
+ @Override
+ public UDFOperandMetadata getOperandMetadata() {
+ return null;
+ }
+
+ public static Object eval(Object... args) {
+ if (args.length < 1 || args[0] == null) {
+ return Collections.emptyList();
+ }
+ String raw = (String) args[0];
+ int forceHeader = (args.length > 1 && args[1] != null) ? ((Number) args[1]).intValue() : -1;
+ boolean noHeader = args.length > 2 && Boolean.TRUE.equals(args[2]);
+ String filterJoined = (args.length > 3 && args[3] != null) ? (String) args[3] : "";
+ List filterTerms =
+ filterJoined.isEmpty()
+ ? Collections.emptyList()
+ : Arrays.asList(
+ filterJoined.split(java.util.regex.Pattern.quote(MultikvParser.FS), -1));
+ return MultikvParser.parse(raw, forceHeader, noHeader, filterTerms);
+ }
+
+ public static class MultikvSplitImplementor implements NotNullImplementor {
+ @Override
+ public Expression implement(
+ RexToLixTranslator translator, RexCall call, List translatedOperands) {
+ ScalarFunctionImpl function =
+ (ScalarFunctionImpl)
+ ScalarFunctionImpl.create(
+ Types.lookupMethod(MultikvSplitFunctionImpl.class, "eval", Object[].class));
+ return function.getImplementor().implement(translator, call, RexImpTable.NullAs.NULL);
+ }
+ }
+}
diff --git a/docs/user/ppl/cmd/multikv.md b/docs/user/ppl/cmd/multikv.md
new file mode 100644
index 0000000000..e04b82b860
--- /dev/null
+++ b/docs/user/ppl/cmd/multikv.md
@@ -0,0 +1,95 @@
+
+# multikv
+
+The `multikv` command extracts field values from an input field and emits one row per source row. The input field can be table-formatted text (for example the aligned output of `ps`, `top`, `netstat`, or `df`), where a header row names the columns and each following line becomes its own output row. The input field can also be an object or an array of objects: a single object yields one row, an array yields one row per element, and each declared column is read from the object. Either way it is a one-to-many (row-multiplying) command.
+
+> **Note**: `multikv` is a streaming (mid-pipeline) command that runs on the coordinating node. It requires the Calcite engine (`plugins.calcite.enabled=true`). It reads from the input field named by `field=`, defaulting to `_raw`. When that field is text it is split into columns; when it is an array of objects each declared column is read from the element (`field= fields ...`). OpenSearch has no implicit `_raw` field, so for the text form either pass `field=` or place the text in `_raw` first with `eval _raw=`. Version 1 is fixed-schema: the output column names must be determinable at plan time via the `fields` clause.
+
+## Syntax
+
+The `multikv` command has the following syntax:
+
+```syntax
+multikv [field=] [fields ...] [forceheader=] [noheader=]
+```
+
+## Parameters
+
+| Parameter | Required/Optional | Description |
+| --- | --- | --- |
+| `field=` | Optional | The input field that holds the table text. Defaults to `_raw`. Use this to read the text directly from a field such as `message` without a preceding `eval _raw=`. |
+| `fields ...` | Optional | Declares the output columns to extract by name. This is the only form that yields named columns. Each extracted column is typed as `string`. |
+| `forceheader` | Optional | The 1-based line number to use as the header, which skips banner lines above it. Must be a positive integer. Used together with `fields`. |
+| `noheader` | Optional | When `true`, there is no header row; the command performs row explosion only and does not produce named columns (for example to count data lines). Default is `false`. |
+
+### Column typing
+
+Every extracted column is typed as `string`, because the values come from splitting text and there is no index mapping to infer a type from. The typed engine coerces these string columns per operation, so numeric use works without an explicit cast (for example `stats avg(pctIdle)` computes numerically, and `where pctIdle > 100` compares numerically). To pin a hard type, cast downstream, for example `eval n = cast(pctIdle as double)`.
+
+## Example 1: Extract a single column
+
+The following query reads the table text from the `raw` field with `field=` and extracts the `pctIdle` column:
+
+```ppl
+source=metrics
+| multikv field=raw fields pctIdle
+| fields pctIdle
+```
+
+The query returns one row per table data row, with a single `pctIdle` (string) column.
+
+## Example 2: Extract multiple columns
+
+```ppl
+source=metrics
+| eval _raw = raw
+| multikv fields CPU pctIdle
+```
+
+The query returns one row per table data row, with `CPU` (string) and `pctIdle` (string) columns.
+
+## Example 3: Skip a banner line with forceheader
+
+When the first line is a banner and the real header is on line 2:
+
+```ppl
+source=report
+| eval _raw = raw
+| multikv fields endpoint forceheader=2
+```
+
+The query uses line 2 as the header and returns the `endpoint` (string) column, one row per data line.
+
+## Example 4: Row explosion with noheader
+
+When there is no header row and only the number of data lines matters:
+
+```ppl
+source=lines
+| eval _raw = raw
+| multikv noheader=true
+| stats count
+```
+
+The query explodes the table text into one row per data line and counts them.
+
+## Example 5: Structured input (array of objects)
+
+When `field=` points at an array of objects, `multikv` emits one row per element and reads each declared column from the element, preserving the element value:
+
+```ppl
+source=hosts
+| multikv field=procs fields pid cpu
+```
+
+For a document with `procs = [{"pid":1,"cpu":0.5},{"pid":42,"cpu":9.1}]`, the query returns two rows: `(1, 0.5)` and `(42, 9.1)`. When `field=` points at a single object rather than an array, one row is returned. Nested container values are returned as-is; extract deeper fields downstream with `spath` or another `multikv field=`.
+
+## Limitations
+
+Version 1 is fixed-schema, so a bare `multikv` with no `fields` clause and no `noheader=true` cannot resolve its output column names at plan time and is rejected with guidance:
+
+```ppl
+source=metrics | eval _raw = raw | multikv | fields pctIdle
+```
+
+Add an explicit `fields` clause to fix it, for example `multikv fields pctIdle`. Runtime header auto-detection is planned for a later version. The `filter` and `rmorig` options are not yet supported.
diff --git a/docs/user/ppl/index.md b/docs/user/ppl/index.md
index 1afb162963..8a6fadcefa 100644
--- a/docs/user/ppl/index.md
+++ b/docs/user/ppl/index.md
@@ -81,6 +81,7 @@ source=accounts
| [explain command](cmd/explain.md) | 3.1 | stable (since 3.1) | Explain the plan of query. |
| [show datasources command](cmd/showdatasources.md) | 2.4 | stable (since 2.4) | Query datasources configured in the PPL engine. |
| [makeresults command](cmd/makeresults.md) | 3.8 | experimental (since 3.8) | Generate in-memory rows for testing and seeding, optionally from inline CSV/JSON data. |
+| [multikv command](cmd/multikv.md) | 3.8 | experimental (since 3.8) | Extract fields from table-formatted text in a field, emitting one row per table data row. |
| [addtotals command](cmd/addtotals.md) | 3.5 | stable (since 3.5) | Adds row and column values and appends a totals column and row. |
| [addcoltotals command](cmd/addcoltotals.md) | 3.5 | stable (since 3.5) | Adds column values and appends a totals row. |
| [transpose command](cmd/transpose.md) | 3.5 | stable (since 3.5) | Transpose rows to columns. |
diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLMultikvCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLMultikvCommandIT.java
new file mode 100644
index 0000000000..f8e54c0c74
--- /dev/null
+++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLMultikvCommandIT.java
@@ -0,0 +1,155 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite.remote;
+
+import static org.opensearch.sql.util.MatcherUtils.rows;
+import static org.opensearch.sql.util.MatcherUtils.schema;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifySchema;
+
+import java.io.IOException;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.ResponseException;
+import org.opensearch.sql.legacy.TestUtils;
+import org.opensearch.sql.ppl.PPLIntegTestCase;
+
+/** Integration tests for the {@code multikv} command (fixed-schema). */
+public class CalcitePPLMultikvCommandIT extends PPLIntegTestCase {
+
+ @Override
+ public void init() throws Exception {
+ super.init();
+ enableCalcite();
+
+ // Auto-header table: header on line 1, two data rows.
+ if (!TestUtils.isIndexExist(client(), "test_multikv")) {
+ TestUtils.createIndexByRestClient(client(), "test_multikv", null);
+ Request doc = new Request("PUT", "/test_multikv/_doc/1?refresh=true");
+ // raw = "CPU pctUser pctIdle\nall 5 90\n0 3 92"
+ doc.setJsonEntity("{\"raw\": \"CPU pctUser pctIdle\\nall 5 90\\n0 3 92\"}");
+ client().performRequest(doc);
+ }
+
+ // forceheader: line 1 is a junk banner, the real header is line 2.
+ if (!TestUtils.isIndexExist(client(), "test_multikv_force")) {
+ TestUtils.createIndexByRestClient(client(), "test_multikv_force", null);
+ Request doc = new Request("PUT", "/test_multikv_force/_doc/1?refresh=true");
+ // raw = "== report ==\nendpoint count\nfoo 1\nbar 2"
+ doc.setJsonEntity("{\"raw\": \"== report ==\\nendpoint count\\nfoo 1\\nbar 2\"}");
+ client().performRequest(doc);
+ }
+
+ // noheader: 3 data lines, no header row.
+ if (!TestUtils.isIndexExist(client(), "test_multikv_noheader")) {
+ TestUtils.createIndexByRestClient(client(), "test_multikv_noheader", null);
+ Request doc = new Request("PUT", "/test_multikv_noheader/_doc/1?refresh=true");
+ doc.setJsonEntity("{\"raw\": \"a 1\\nb 2\\nc 3\"}");
+ client().performRequest(doc);
+ }
+
+ // structured: procs is an array of objects (nested), two elements.
+ if (!TestUtils.isIndexExist(client(), "test_multikv_struct")) {
+ String mapping =
+ "{\"mappings\":{\"properties\":{\"procs\":{\"type\":\"nested\","
+ + "\"properties\":{\"pid\":{\"type\":\"long\"},\"cpu\":{\"type\":\"double\"}}}}}}";
+ TestUtils.createIndexByRestClient(client(), "test_multikv_struct", mapping);
+ Request doc = new Request("PUT", "/test_multikv_struct/_doc/1?refresh=true");
+ doc.setJsonEntity("{\"procs\":[{\"pid\":1,\"cpu\":0.5},{\"pid\":42,\"cpu\":9.1}]}");
+ client().performRequest(doc);
+ }
+
+ // single object: meta is one object (map), not an array.
+ if (!TestUtils.isIndexExist(client(), "test_multikv_obj")) {
+ String mapping =
+ "{\"mappings\":{\"properties\":{\"meta\":{\"properties\":"
+ + "{\"owner\":{\"type\":\"keyword\"},\"role\":{\"type\":\"keyword\"}}}}}}";
+ TestUtils.createIndexByRestClient(client(), "test_multikv_obj", mapping);
+ Request doc = new Request("PUT", "/test_multikv_obj/_doc/1?refresh=true");
+ doc.setJsonEntity("{\"meta\":{\"owner\":\"root\",\"role\":\"admin\"}}");
+ client().performRequest(doc);
+ }
+ }
+
+ @Test
+ public void testMultikvFields() throws IOException {
+ // Declared single column: auto-detected header maps pctIdle -> its column.
+ JSONObject result =
+ executeQuery(
+ "source=test_multikv | eval _raw = raw | multikv fields pctIdle | fields pctIdle");
+ verifySchema(result, schema("pctIdle", "string"));
+ verifyDataRows(result, rows("90"), rows("92"));
+ }
+
+ @Test
+ public void testMultikvFieldsMultipleColumns() throws IOException {
+ JSONObject result =
+ executeQuery("source=test_multikv | eval _raw = raw | multikv fields CPU pctIdle");
+ verifySchema(result, schema("CPU", "string"), schema("pctIdle", "string"));
+ verifyDataRows(result, rows("all", "90"), rows("0", "92"));
+ }
+
+ @Test
+ public void testMultikvForceHeader() throws IOException {
+ // forceheader=2 skips the banner and uses line 2 ("endpoint count") as the header.
+ JSONObject result =
+ executeQuery(
+ "source=test_multikv_force | eval _raw = raw | multikv fields endpoint forceheader=2 |"
+ + " fields endpoint");
+ verifySchema(result, schema("endpoint", "string"));
+ verifyDataRows(result, rows("foo"), rows("bar"));
+ }
+
+ @Test
+ public void testMultikvNoHeaderCount() throws IOException {
+ // Positional noheader, downstream only counts rows (3 data lines -> count 3).
+ JSONObject result =
+ executeQuery(
+ "source=test_multikv_noheader | eval _raw = raw | multikv noheader=true | stats count");
+ verifyDataRows(result, rows(3));
+ }
+
+ @Test
+ public void testMultikvFieldEquals() throws IOException {
+ // field= reads the table text directly from a field, no eval _raw needed.
+ JSONObject result =
+ executeQuery("source=test_multikv | multikv field=raw fields pctIdle | fields pctIdle");
+ verifySchema(result, schema("pctIdle", "string"));
+ verifyDataRows(result, rows("90"), rows("92"));
+ }
+
+ @Test
+ public void testMultikvStructuredArrayOfObjects() throws IOException {
+ // Structured input: procs is an array of objects. multikv explodes it into one row per
+ // element and reads each declared column with ITEM.
+ JSONObject result =
+ executeQuery(
+ "source=test_multikv_struct | multikv field=procs fields pid cpu"
+ + " | eval p = cast(pid as int) | fields p | sort p");
+ verifyDataRows(result, rows(1), rows(42));
+ }
+
+ @Test
+ public void testMultikvSingleObject() throws IOException {
+ // Single object (map): no row multiplication, each declared column read with ITEM -> 1 row.
+ JSONObject result =
+ executeQuery("source=test_multikv_obj | multikv field=meta fields owner role");
+ verifyDataRows(result, rows("root", "admin"));
+ }
+
+ @Test
+ public void testBareMultikvRejectedWithGuidance() throws IOException {
+ // Bare auto-header multikv referencing a column downstream must be rejected at plan time
+ // with guidance to declare a fields clause.
+ Throwable t =
+ org.junit.Assert.assertThrows(
+ ResponseException.class,
+ () -> executeQuery("source=test_multikv | eval _raw = raw | multikv | fields pctIdle"));
+ org.junit.Assert.assertTrue(
+ t.getMessage(), t.getMessage().toLowerCase().contains("fields clause"));
+ }
+}
diff --git a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java
index 6b5ac0d430..127adc4230 100644
--- a/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java
+++ b/integ-test/src/test/java/org/opensearch/sql/ppl/NewAddedCommandsIT.java
@@ -18,6 +18,7 @@
import org.json.JSONArray;
import org.json.JSONObject;
import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
import org.opensearch.client.ResponseException;
import org.opensearch.sql.util.TestUtils;
@@ -339,6 +340,34 @@ public void testMakeResults() throws IOException {
}
}
+ @Test
+ public void testMultikv() throws IOException {
+ // v3-marker (registered on Calcite, rejected on V2). The table text (with newlines) lives in
+ // indexed data, not the query string, so the request payload stays valid JSON: a header line
+ // plus two data rows yields two extracted rows.
+ String index = "test_new_added_multikv";
+ if (!TestUtils.isIndexExist(client(), index)) {
+ TestUtils.createIndexByRestClient(client(), index, null);
+ Request doc = new Request("PUT", "/" + index + "/_doc/1?refresh=true");
+ doc.setJsonEntity("{\"raw\": \"CPU pctIdle\\nall 90\\n0 92\"}");
+ client().performRequest(doc);
+ }
+ JSONObject result;
+ try {
+ result =
+ executeQuery("source=" + index + " | multikv field=raw fields pctIdle | fields pctIdle");
+ } catch (ResponseException e) {
+ result = new JSONObject(TestUtils.getResponseBody(e.getResponse()));
+ }
+
+ if (isCalciteEnabled()) {
+ assertThat(result.getJSONArray("datarows").length(), equalTo(2));
+ } else {
+ // multikv is Calcite-only, so V2 returns an error.
+ assertThat(result.has("error"), equalTo(true));
+ }
+ }
+
@Test
public void testMvExpandCommandBasicExpansion() throws IOException {
JSONObject result;
diff --git a/ppl/src/main/antlr/OpenSearchPPLLexer.g4 b/ppl/src/main/antlr/OpenSearchPPLLexer.g4
index b26751ad61..573e7177ad 100644
--- a/ppl/src/main/antlr/OpenSearchPPLLexer.g4
+++ b/ppl/src/main/antlr/OpenSearchPPLLexer.g4
@@ -76,6 +76,10 @@ ROW: 'ROW';
COL: 'COL';
EXPAND: 'EXPAND';
MVEXPAND: 'MVEXPAND';
+MULTIKV: 'MULTIKV';
+FORCEHEADER: 'FORCEHEADER';
+NOHEADER: 'NOHEADER';
+RMORIG: 'RMORIG';
SIMPLE_PATTERN: 'SIMPLE_PATTERN';
BRAIN: 'BRAIN';
VARIABLE_COUNT_THRESHOLD: 'VARIABLE_COUNT_THRESHOLD';
diff --git a/ppl/src/main/antlr/OpenSearchPPLParser.g4 b/ppl/src/main/antlr/OpenSearchPPLParser.g4
index eeaed6daf5..37895f4cfb 100644
--- a/ppl/src/main/antlr/OpenSearchPPLParser.g4
+++ b/ppl/src/main/antlr/OpenSearchPPLParser.g4
@@ -86,6 +86,7 @@ commands
| appendCommand
| expandCommand
| mvexpandCommand
+ | multikvCommand
| flattenCommand
| reverseCommand
| regexCommand
@@ -135,6 +136,7 @@ commandName
| CONVERT
| EXPAND
| MVEXPAND
+ | MULTIKV
| FLATTEN
| TRENDLINE
| TIMECHART
@@ -652,6 +654,21 @@ mvexpandCommand
: MVEXPAND fieldExpression (LIMIT EQUAL INTEGER_LITERAL)?
;
+// multikv: extract field values from table-formatted text in a field (default _raw),
+// emitting one row per table data row. v1 fixed-schema: output columns must be
+// determinable at plan time via the fields list, a literal forceheader, or noheader.
+multikvCommand
+ : MULTIKV multikvParameter*
+ ;
+
+multikvParameter
+ : FIELDS fields = fieldList
+ | FIELD EQUAL inField = qualifiedName
+ | FORCEHEADER EQUAL forceHeader = integerLiteral
+ | NOHEADER EQUAL noHeader = booleanLiteral
+ | RMORIG EQUAL rmOrig = booleanLiteral
+ ;
+
flattenCommand
: FLATTEN fieldExpression (AS aliases = identifierSeq)?
;
diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java
index e87264909c..efe35f8baa 100644
--- a/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java
+++ b/ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java
@@ -99,6 +99,7 @@
import org.opensearch.sql.ast.tree.ML;
import org.opensearch.sql.ast.tree.MakeResults;
import org.opensearch.sql.ast.tree.MinSpanBin;
+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;
@@ -1076,6 +1077,38 @@ public UnresolvedPlan visitMvexpandCommand(OpenSearchPPLParser.MvexpandCommandCo
return new MvExpand(field, limit);
}
+ @Override
+ public UnresolvedPlan visitMultikvCommand(OpenSearchPPLParser.MultikvCommandContext ctx) {
+ List fields = null;
+ String inField = Multikv.DEFAULT_INPUT_FIELD;
+ Integer forceHeader = null;
+ boolean noHeader = false;
+ boolean rmOrig = true; // Splunk default
+
+ for (OpenSearchPPLParser.MultikvParameterContext p : ctx.multikvParameter()) {
+ if (p.fields != null) {
+ fields =
+ p.fields.fieldExpression().stream()
+ .map(f -> (Field) expressionBuilder.visit(f))
+ .collect(Collectors.toList());
+ } else if (p.inField != null) {
+ inField = p.inField.getText();
+ } else if (p.forceHeader != null) {
+ forceHeader = Integer.parseInt(p.forceHeader.getText());
+ if (forceHeader <= 0) {
+ throw new IllegalArgumentException(
+ "multikv forceheader must be a positive line number, got: " + forceHeader);
+ }
+ } else if (p.noHeader != null) {
+ noHeader = Boolean.parseBoolean(p.noHeader.getText());
+ } else if (p.rmOrig != null) {
+ rmOrig = Boolean.parseBoolean(p.rmOrig.getText());
+ }
+ }
+
+ return new Multikv(inField, fields, null, forceHeader, noHeader, rmOrig);
+ }
+
@Override
public UnresolvedPlan visitGrokCommand(OpenSearchPPLParser.GrokCommandContext ctx) {
UnresolvedExpression sourceField = internalVisitExpression(ctx.source_field);
diff --git a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java
index 11c47d137e..4629197886 100644
--- a/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java
+++ b/ppl/src/main/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizer.java
@@ -84,6 +84,7 @@
import org.opensearch.sql.ast.tree.Join;
import org.opensearch.sql.ast.tree.Lookup;
import org.opensearch.sql.ast.tree.MinSpanBin;
+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;
@@ -611,6 +612,25 @@ public String visitForeach(Foreach node, String context) {
return command.append(" [ eval ").append(evalClauses).append(" ]").toString();
}
+ @Override
+ public String visitMultikv(Multikv node, String context) {
+ String child = node.getChild().get(0).accept(this, context);
+ StringBuilder sb = new StringBuilder(child).append(" | multikv");
+ if (node.getFields() != null && !node.getFields().isEmpty()) {
+ sb.append(" fields ").append(MASK_COLUMN);
+ }
+ if (node.getFilterTerms() != null && !node.getFilterTerms().isEmpty()) {
+ sb.append(" filter ").append(MASK_LITERAL);
+ }
+ if (node.getForceHeader() != null) {
+ sb.append(" forceheader=").append(MASK_LITERAL);
+ }
+ if (node.isNoHeader()) {
+ sb.append(" noheader=true");
+ }
+ return sb.toString();
+ }
+
/** Build {@link LogicalSort}. */
@Override
public String visitSort(Sort node, String context) {
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMultikvTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMultikvTest.java
new file mode 100644
index 0000000000..abe55d6382
--- /dev/null
+++ b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLMultikvTest.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ppl.calcite;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.test.CalciteAssert;
+import org.junit.Test;
+
+/**
+ * Logical-plan tests for the {@code multikv} command (fixed-schema).
+ *
+ * multikv lowers to a three-layer rewrite: an {@code eval} that calls {@code MULTIKV_SPLIT} into
+ * a helper record column, an mvexpand (Uncollect over a Correlate) that explodes one row per table
+ * data row, and a per-declared-column {@code MULTIKV_EXTRACT} eval followed by a narrowing project.
+ * These assertions pin the presence and absence of those operators rather than the full Rex
+ * rendering, which is exercised end-to-end by CalcitePPLMultikvCommandIT.
+ */
+public class CalcitePPLMultikvTest extends CalcitePPLAbstractTest {
+ public CalcitePPLMultikvTest() {
+ super(CalciteAssert.SchemaSpec.SCOTT_WITH_TEMPORAL);
+ }
+
+ private String plan(String ppl) {
+ RelNode root = getRelNode(ppl);
+ return RelOptUtil.toString(root);
+ }
+
+ private void expectError(String ppl, String messageFragment) {
+ try {
+ getRelNode(ppl);
+ fail("expected an error for: " + ppl);
+ } catch (Exception e) {
+ String msg = String.valueOf(e.getMessage());
+ assertTrue(
+ "expected message containing '" + messageFragment + "' but got: " + msg,
+ msg.contains(messageFragment));
+ }
+ }
+
+ @Test
+ public void testMultikvFieldsSingleColumn() {
+ String p = plan("source=EMP | eval _raw = ENAME | multikv fields pctIdle | fields pctIdle");
+ // parse layer
+ assertTrue(p, p.contains("MULTIKV_SPLIT"));
+ // explode layer (mvexpand = Uncollect over Correlate)
+ assertTrue(p, p.contains("Uncollect") || p.contains("Correlate"));
+ // extract layer for the declared column
+ assertTrue(p, p.contains("MULTIKV_EXTRACT"));
+ assertTrue(p, p.contains("pctIdle"));
+ }
+
+ @Test
+ public void testMultikvFieldsMultipleColumns() {
+ String p = plan("source=EMP | eval _raw = ENAME | multikv fields CPU pctIdle");
+ // one extract per declared column
+ int idx = p.indexOf("MULTIKV_EXTRACT");
+ assertTrue(p, idx >= 0 && p.indexOf("MULTIKV_EXTRACT", idx + 1) >= 0);
+ assertTrue(p, p.contains("CPU"));
+ assertTrue(p, p.contains("pctIdle"));
+ }
+
+ @Test
+ public void testMultikvForceHeader() {
+ // forceheader=2 is threaded into MULTIKV_SPLIT as the header-line argument.
+ String p = plan("source=EMP | eval _raw = ENAME | multikv fields endpoint forceheader=2");
+ assertTrue(p, p.contains("MULTIKV_SPLIT"));
+ assertTrue(p, p.contains("MULTIKV_EXTRACT"));
+ assertTrue(p, p.contains("endpoint"));
+ }
+
+ @Test
+ public void testMultikvNoHeaderIsRowExplosionOnly() {
+ // Positional noheader with no fields: split + explode, but no named-column extract/project.
+ String p = plan("source=EMP | eval _raw = ENAME | multikv noheader=true");
+ assertTrue(p, p.contains("MULTIKV_SPLIT"));
+ assertTrue(p, p.contains("Uncollect") || p.contains("Correlate"));
+ assertFalse(p, p.contains("MULTIKV_EXTRACT"));
+ }
+
+ @Test
+ public void testBareMultikvRejectedWithGuidance() {
+ // Bare auto-header multikv (no fields, no noheader) cannot resolve a plan-time schema and is
+ // rejected at the field-resolution phase with actionable guidance.
+ expectError("source=EMP | eval _raw = ENAME | multikv", "fields clause");
+ }
+}
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java
index d5f45a96f8..07afb12fb9 100644
--- a/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java
+++ b/ppl/src/test/java/org/opensearch/sql/ppl/parser/AstBuilderTest.java
@@ -82,6 +82,7 @@
import org.opensearch.sql.ast.tree.Kmeans;
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.RareTopN.CommandType;
import org.opensearch.sql.common.antlr.SyntaxCheckException;
import org.opensearch.sql.common.setting.Settings.Key;
@@ -1114,6 +1115,25 @@ public void testMakeResultsCommand() {
assertEqual("makeresults count=5", new MakeResults(5));
}
+ @Test
+ public void testMultikvCommand() {
+ assertEqual(
+ "source=t | multikv fields CPU pctIdle",
+ new Multikv("_raw", Arrays.asList(field("CPU"), field("pctIdle")), null, null, false, true)
+ .attach(relation("t")));
+ assertEqual(
+ "source=t | multikv fields endpoint forceheader=2",
+ new Multikv("_raw", Arrays.asList(field("endpoint")), null, 2, false, true)
+ .attach(relation("t")));
+ assertEqual(
+ "source=t | multikv noheader=true",
+ new Multikv("_raw", null, null, null, true, true).attach(relation("t")));
+ assertEqual(
+ "source=t | multikv field=message fields CPU",
+ new Multikv("message", Arrays.asList(field("CPU")), null, null, false, true)
+ .attach(relation("t")));
+ }
+
@Test
public void testDescribeMatchAllCrossClusterSearchCommand() {
assertEqual("describe *:t", describe(mappingTable("*:t")));
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java
index eba4f57112..63cc8f6c87 100644
--- a/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java
+++ b/ppl/src/test/java/org/opensearch/sql/ppl/utils/PPLQueryDataAnonymizerTest.java
@@ -41,6 +41,25 @@ public void testMakeResultsCommand() {
assertEquals("makeresults", anonymize("makeresults count=5"));
}
+ @Test
+ public void testMultikvFieldsCommand() {
+ assertEquals(
+ "source=table | multikv fields identifier", anonymize("source=t | multikv fields pctIdle"));
+ }
+
+ @Test
+ public void testMultikvForceHeaderCommand() {
+ assertEquals(
+ "source=table | multikv fields identifier forceheader=***",
+ anonymize("source=t | multikv fields endpoint forceheader=2"));
+ }
+
+ @Test
+ public void testMultikvNoHeaderCommand() {
+ assertEquals(
+ "source=table | multikv noheader=true", anonymize("source=t | multikv noheader=true"));
+ }
+
@Test
public void testTableFunctionCommand() {
assertEquals(