Feature/mvcombine

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

@@ -80,6 +80,7 @@
 import org.opensearch.sql.ast.tree.Lookup;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.Multisearch;
+import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -535,6 +536,11 @@ public LogicalPlan visitAddColTotals(AddColTotals node, AnalysisContext context)
     throw getOnlyForCalciteException("addcoltotals");
   }
 
+  @Override
+  public LogicalPlan visitMvCombine(MvCombine node, AnalysisContext context) {
+    throw getOnlyForCalciteException("mvcombine");
+  }
+
   /** Build {@link ParseExpression} to context and skip to child nodes. */
   @Override
   public LogicalPlan visitParse(Parse node, AnalysisContext context) {

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

@@ -68,6 +68,7 @@
 import org.opensearch.sql.ast.tree.Lookup;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.Multisearch;
+import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -466,4 +467,8 @@ public T visitAddTotals(AddTotals node, C context) {
   public T visitAddColTotals(AddColTotals node, C context) {
     return visitChildren(node, context);
   }
+
+  public T visitMvCombine(MvCombine node, C context) {
+    return visitChildren(node, context);
+  }
 }

core/src/main/java/org/opensearch/sql/ast/analysis/FieldResolutionVisitor.java

@@ -44,6 +44,7 @@
 import org.opensearch.sql.ast.tree.Join;
 import org.opensearch.sql.ast.tree.Lookup;
 import org.opensearch.sql.ast.tree.Multisearch;
+import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
 import org.opensearch.sql.ast.tree.Project;
@@ -628,6 +629,15 @@ public Node visitExpand(Expand node, FieldResolutionContext context) {
     return node;
   }
 
+  @Override
+  public Node visitMvCombine(MvCombine node, FieldResolutionContext context) {
+    Set<String> mvCombineFields = extractFieldsFromExpression(node.getField());
+    context.pushRequirements(context.getCurrentRequirements().or(mvCombineFields));
+    visitChildren(node, context);
+    context.popRequirements();
+    return node;
+  }
+
   private Set<String> extractFieldsFromAggregation(UnresolvedExpression expr) {
     Set<String> fields = new HashSet<>();
     if (expr instanceof Alias alias) {

core/src/main/java/org/opensearch/sql/ast/dsl/AstDSL.java

@@ -62,6 +62,7 @@
 import org.opensearch.sql.ast.tree.Head;
 import org.opensearch.sql.ast.tree.Limit;
 import org.opensearch.sql.ast.tree.MinSpanBin;
+import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
 import org.opensearch.sql.ast.tree.Project;
@@ -468,6 +469,14 @@ public static List<Argument> defaultDedupArgs() {
         argument("consecutive", booleanLiteral(false)));
   }
 
+  public static MvCombine mvcombine(Field field) {
+    return new MvCombine(field, null);
+  }
+
+  public static MvCombine mvcombine(Field field, String delim) {
+    return new MvCombine(field, delim);
+  }
+
   public static List<Argument> sortOptions() {
     return exprList(argument("desc", booleanLiteral(false)));
   }

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

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+@Getter
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = false)
+public class MvCombine extends UnresolvedPlan {
+
+  private final Field field;
+  private final String delim;
+  @Nullable private UnresolvedPlan child;
+
+  public MvCombine(Field field, @Nullable String delim) {
+    this.field = field;
+    this.delim = (delim == null) ? " " : delim;
+  }
+
+  public MvCombine attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  @Override
+  public List<UnresolvedPlan> getChild() {
+    return child == null ? ImmutableList.of() : ImmutableList.of(child);
+  }
+
+  @Override
+  public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) {
+    return nodeVisitor.visitMvCombine(this, context);
+  }
+}

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

@@ -63,6 +63,7 @@
 import org.apache.calcite.rex.RexVisitorImpl;
 import org.apache.calcite.rex.RexWindowBounds;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlLibraryOperators;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.fun.SqlTrimFunction;
 import org.apache.calcite.sql.type.ArraySqlType;
@@ -124,6 +125,7 @@
 import org.opensearch.sql.ast.tree.Lookup.OutputStrategy;
 import org.opensearch.sql.ast.tree.ML;
 import org.opensearch.sql.ast.tree.Multisearch;
+import org.opensearch.sql.ast.tree.MvCombine;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -3168,6 +3170,204 @@ public RelNode visitExpand(Expand expand, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  /**
+   * mvcombine command visitor to collapse rows that are identical on all non-target, non-metadata
+   * fields, and combine the target field values into a multivalue (array) field.
+   *
+   * <p>Implementation notes: Groups by all non-target, non-metadata fields. Aggregates target
+   * values using {@code ARRAY_AGG}, producing a multivalue {@code ARRAY<T>} directly. Preserves the
+   * original output column order.
+   *
+   * @param node mvcombine command to be visited
+   * @param context CalcitePlanContext containing the RelBuilder and resolution context
+   * @return RelNode representing collapsed records with the target combined into a multivalue array
+   * @throws SemanticCheckException if the mvcombine target is not a direct field reference
+   */
+  @Override
+  public RelNode visitMvCombine(MvCombine node, CalcitePlanContext context) {
+    // 1) Lower the child plan first so the RelBuilder has the input schema on the stack.
+    visitChildren(node, context);
+
+    final RelBuilder relBuilder = context.relBuilder;
+
+    final RelNode input = relBuilder.peek();
+    final List<String> inputFieldNames = input.getRowType().getFieldNames();
+
+    // 2) Resolve the mvcombine target to an input column index (must be a direct field reference).
+    final Field targetField = node.getField();
+    final int targetIndex = resolveTargetIndex(targetField, context);
+    final String targetName = inputFieldNames.get(targetIndex);
+    final boolean includeMetaFields = context.isProjectVisited();
+
+    // 3) Group by all non-target, non-metadata fields.
+    final List<RexNode> groupExprs =
+        buildGroupExpressionsExcludingTarget(
+            targetIndex, inputFieldNames, relBuilder, includeMetaFields);
+
+    // If all remaining fields are metadata or the target itself, mvcombine degenerates to a global
+    // combine. This is intentional: result collapses to a single row with aggregated target values.
+
+    // 4) Aggregate target values using ARRAY_AGG, filtering out NULLs.
+    performArrayAggAggregation(relBuilder, targetIndex, targetName, groupExprs);
+
+    // 5) Restore original output column order (ARRAY_AGG already returns ARRAY<T>).
+    restoreColumnOrderAfterArrayAgg(
+        relBuilder, inputFieldNames, targetIndex, groupExprs, includeMetaFields);
+
+    return relBuilder.peek();
+  }
+
+  /**
+   * Resolves the mvcombine target expression to an input field index.
+   *
+   * <p>mvcombine requires the target to be a direct field reference (RexInputRef). This keeps the
+   * command semantics predictable and avoids accidental grouping on computed expressions.
+   *
+   * <p>The target must also be a scalar-ish field. mvcombine outputs ARRAY&lt;T&gt;, so the input
+   * target cannot already be an ARRAY or MULTISET.
+   *
+   * @param targetField Target field expression from the AST
+   * @param context Planning context
+   * @return 0-based input field index for the target
+   * @throws SemanticCheckException if the target is not a direct field reference or has an array
+   *     type
+   */
+  private int resolveTargetIndex(Field targetField, CalcitePlanContext context) {
+    final RexNode targetRex = rexVisitor.analyze(targetField, context);
+    if (!isInputRef(targetRex)) {
+      throw new SemanticCheckException(
+          "mvcombine target must be a direct field reference, but got: " + targetField);
+    }
+
+    final int index = ((RexInputRef) targetRex).getIndex();
+
+    final RelDataType fieldType =
+        context.relBuilder.peek().getRowType().getFieldList().get(index).getType();
+
+    if (SqlTypeUtil.isArray(fieldType) || SqlTypeUtil.isMultiset(fieldType)) {
+      throw new SemanticCheckException(
+          "mvcombine target cannot be an array/multivalue type, but got: " + fieldType);
+    }
+
+    return index;
+  }
+
+  /**
+   * Builds group-by expressions for mvcombine: all non-target, non-metadata input fields.
+   *
+   * @param targetIndex Input index of the mvcombine target field
+   * @param inputFieldNames Input schema field names (for sizing/ordering)
+   * @param relBuilder RelBuilder positioned on the input
+   * @return Group-by expressions in input order excluding the target
+   */
+  private List<RexNode> buildGroupExpressionsExcludingTarget(
+      int targetIndex,
+      List<String> inputFieldNames,
+      RelBuilder relBuilder,
+      boolean includeMetaFields) {
+    final List<RexNode> groupExprs = new ArrayList<>(Math.max(0, inputFieldNames.size() - 1));
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      if (i == targetIndex) {
+        continue;
+      }
+      final String fieldName = inputFieldNames.get(i);
+
+      if (isMetadataField(fieldName) && !includeMetaFields) continue;
+
+      groupExprs.add(relBuilder.field(i));
+    }
+    return groupExprs;
+  }
+
+  /**
+   * Applies mvcombine aggregation:
+   *
+   * <p>GROUP BY all non-target fields (excluding metadata fields), and aggregate target values
+   * using {@code ARRAY_AGG}. {@code ARRAY_AGG} produces an {@code ARRAY<T>} in Calcite, which we
+   * keep as-is for output.
+   *
+   * <p>NULL target values are excluded from the aggregated array by applying an aggregate filter.
+   *
+   * @param relBuilder RelBuilder positioned on the input
+   * @param targetIndex Target field input index
+   * @param targetName Target field output name (preserved)
+   * @param groupExprs Group-by expressions (non-target, non-metadata fields)
+   */
+  private void performArrayAggAggregation(
+      RelBuilder relBuilder, int targetIndex, String targetName, List<RexNode> groupExprs) {
+
+    final RexNode targetRef = relBuilder.field(targetIndex);
+    final RexNode notNullTarget = relBuilder.isNotNull(targetRef);
+
+    final RelBuilder.AggCall aggCall =
+        relBuilder
+            .aggregateCall(SqlLibraryOperators.ARRAY_AGG, targetRef)
+            .filter(notNullTarget)
+            .as(targetName);
+
+    relBuilder.aggregate(relBuilder.groupKey(groupExprs), aggCall);
+  }
+
+  /**
+   * Restores the original output column order after the aggregate step.
+   *
+   * <p>After aggregation, the schema is:
+   *
+   * <pre>
+   *   [groupField0, groupField1, ..., groupFieldN, targetAggArray]
+   * </pre>
+   *
+   * <p>This method projects fields back to the original input order, replacing the original target
+   * slot with the aggregated target value. Metadata fields that were excluded from the group key
+   * are preserved in the output schema but projected as {@code NULL}, since they do not exist in
+   * the post-aggregate row.
+   *
+   * <p>Since {@code ARRAY_AGG} already returns {@code ARRAY<T>}, no cast is needed.
+   *
+   * @param relBuilder RelBuilder positioned on the post-aggregate node
+   * @param inputFieldNames Original input field names (also output field names)
+   * @param targetIndex Target field index in the original input
+   * @param groupExprs Group-by expressions used during aggregation
+   */
+  private void restoreColumnOrderAfterArrayAgg(
+      RelBuilder relBuilder,
+      List<String> inputFieldNames,
+      int targetIndex,
+      List<RexNode> groupExprs,
+      boolean includeMetaFields) {
+
+    // Post-aggregate: group fields come first, and the aggregated target is appended at the end.
+    final int aggregatedTargetPos = groupExprs.size();
+
+    final List<RexNode> projections = new ArrayList<>(inputFieldNames.size());
+    final List<String> projectionNames = new ArrayList<>(inputFieldNames.size());
+
+    int groupPos = 0;
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      final String fieldName = inputFieldNames.get(i);
+      projectionNames.add(fieldName);
+
+      if (i == targetIndex) {
+        // ARRAY_AGG already returns ARRAY<T>
+        projections.add(relBuilder.field(aggregatedTargetPos));
+      } else if (isMetadataField(fieldName)) {
+        // Metadata fields are intentionally not grouped by mvcombine.
+        // Preserve schema correctness by projecting a TYPED NULL
+        // (prevents "undefined type" for fields like _id when explicitly selected).
+        projections.add(
+            relBuilder
+                .getRexBuilder()
+                .makeNullLiteral(relBuilder.peek().getRowType().getFieldList().get(i).getType()));
+      } else {
+        projections.add(relBuilder.field(groupPos));
+        groupPos++;
+      }
+    }
+
+    // Force projection to avoid Calcite "identity" short-circuit when only names/types change.
+    relBuilder.project(projections, projectionNames, /* force= */ true);
+  }
+
   @Override
   public RelNode visitValues(Values values, CalcitePlanContext context) {
     if (values.getValues() == null || values.getValues().isEmpty()) {

docs/category.json

@@ -36,6 +36,7 @@
     "user/ppl/cmd/sort.md",
     "user/ppl/cmd/spath.md",
     "user/ppl/cmd/stats.md",
+    "user/ppl/cmd/mvcombine.md",
     "user/ppl/cmd/streamstats.md",
     "user/ppl/cmd/subquery.md",
     "user/ppl/cmd/syntax.md",

docs/user/dql/metadata.rst

@@ -35,7 +35,7 @@ Example 1: Show All Indices Information
 SQL query::
 
     os> SHOW TABLES LIKE '%'
-    fetched rows / total rows = 23/23
+    fetched rows / total rows = 24/24
     +----------------+-------------+-------------------+------------+---------+----------+------------+-----------+---------------------------+----------------+
     | TABLE_CAT      | TABLE_SCHEM | TABLE_NAME        | TABLE_TYPE | REMARKS | TYPE_CAT | TYPE_SCHEM | TYPE_NAME | SELF_REFERENCING_COL_NAME | REF_GENERATION |
     |----------------+-------------+-------------------+------------+---------+----------+------------+-----------+---------------------------+----------------|
@@ -48,6 +48,7 @@ SQL query::
     | docTestCluster | null        | events_many_hosts | BASE TABLE | null    | null     | null       | null      | null                      | null           |
     | docTestCluster | null        | events_null       | BASE TABLE | null    | null     | null       | null      | null                      | null           |
     | docTestCluster | null        | json_test         | BASE TABLE | null    | null     | null       | null      | null                      | null           |
+    | docTestCluster | null        | mvcombine_data         | BASE TABLE | null    | null     | null       | null      | null                      | null           |
     | docTestCluster | null        | nested            | BASE TABLE | null    | null     | null       | null      | null                      | null           |
     | docTestCluster | null        | nyc_taxi          | BASE TABLE | null    | null     | null       | null      | null                      | null           |
     | docTestCluster | null        | occupation        | BASE TABLE | null    | null     | null       | null      | null                      | null           |