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;
@@ -534,6 +535,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;
@@ -461,4 +462,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/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

@@ -53,6 +53,7 @@
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFamily;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexCorrelVariable;
 import org.apache.calcite.rex.RexInputRef;
@@ -121,6 +122,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;
@@ -3086,6 +3088,191 @@ public RelNode visitExpand(Expand expand, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  /**
+   * mvcombine command visitor to collapse rows that are identical on all fields except the target
+   * field, and combine the target field values into a multivalue (array) field.
+   *
+   * <p>Implementation notes:Groups by all input fields except the target field. Aggregates target
+   * values using {@code COLLECT} (MULTISET). Casts the aggregation result from MULTISET to ARRAY
+   * for a stable multivalue output type. Preserves the original output column order.
+   *
+   * @param node mvcombine command to be visited
+   * @param context CalcitePlanContext containing the RelBuilder, RexBuilder, 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 RexBuilder rexBuilder = context.rexBuilder;
+
+    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);
+
+    // 3) Group by all fields except the target.
+    final List<RexNode> groupExprs =
+        buildGroupExpressionsExcludingTarget(targetIndex, inputFieldNames, relBuilder);
+
+    // 4) Aggregate target values using COLLECT, filtering out NULLs.
+    performCollectAggregation(relBuilder, targetIndex, targetName, groupExprs);
+
+    // 5) Restore original output column order, and cast COLLECT's MULTISET output to ARRAY<T>.
+    restoreColumnOrderWithArrayCast(
+        relBuilder, rexBuilder, input, inputFieldNames, targetIndex, groupExprs);
+
+    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 (!(targetRex instanceof RexInputRef)) {
+      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 (fieldType.getSqlTypeName() == org.apache.calcite.sql.type.SqlTypeName.ARRAY
+        || fieldType.getSqlTypeName() == org.apache.calcite.sql.type.SqlTypeName.MULTISET) {
+      throw new SemanticCheckException(
+          "mvcombine target cannot be an array/multivalue type, but got: " + fieldType);
+    }
+
+    return index;
+  }
+
+  /**
+   * Builds group-by expressions for mvcombine: all input fields except the target field.
+   *
+   * @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) {
+    final List<RexNode> groupExprs = new ArrayList<>(Math.max(0, inputFieldNames.size() - 1));
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      if (i == targetIndex) {
+        continue;
+      }
+      groupExprs.add(relBuilder.field(i));
+    }
+    return groupExprs;
+  }
+
+  /**
+   * Applies mvcombine aggregation:
+   *
+   * <p>GROUP BY all non-target fields, and aggregate target values using {@code COLLECT}. {@code
+   * COLLECT} produces a MULTISET in Calcite, which we later cast to ARRAY for output.
+   *
+   * <p>NULL target values are excluded from the collected multivalue list by applying an aggregate
+   * filter. This matches typical "combine values" semantics and avoids polluting the result with
+   * NULL elements.
+   *
+   * @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 (all fields except target)
+   */
+  private void performCollectAggregation(
+      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(SqlStdOperatorTable.COLLECT, targetRef)
+            .filter(notNullTarget)
+            .as(targetName);
+
+    relBuilder.aggregate(relBuilder.groupKey(groupExprs), aggCall);
+  }
+
+  /**
+   * Restores the original output column order after the aggregate step and converts the collected
+   * target from MULTISET to ARRAY&lt;T&gt;.
+   *
+   * <p>After aggregation, the schema is:
+   *
+   * <pre>
+   *   [groupField0, groupField1, ..., groupFieldN, targetAggMultiset]
+   * </pre>
+   *
+   * <p>This method projects fields back to the original input order, replacing the original target
+   * slot with {@code CAST(targetAggMultiset AS ARRAY&lt;T&gt;)}.
+   *
+   * @param relBuilder RelBuilder positioned on the post-aggregate node
+   * @param rexBuilder RexBuilder for explicit casts
+   * @param input Original input RelNode (used to derive the target element type)
+   * @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 restoreColumnOrderWithArrayCast(
+      RelBuilder relBuilder,
+      RexBuilder rexBuilder,
+      RelNode input,
+      List<String> inputFieldNames,
+      int targetIndex,
+      List<RexNode> groupExprs) {
+
+    // Post-aggregate: group fields come first, and the collected target is appended at the end.
+    final int collectedTargetPos = groupExprs.size();
+
+    final RelDataType targetElemType = input.getRowType().getFieldList().get(targetIndex).getType();
+    final RelDataType targetArrayType =
+        relBuilder.getTypeFactory().createArrayType(targetElemType, -1);
+
+    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++) {
+      projectionNames.add(inputFieldNames.get(i));
+
+      if (i == targetIndex) {
+        // COLLECT returns MULTISET; normalize output to ARRAY<T>.
+        final RexNode multisetRef = relBuilder.field(collectedTargetPos);
+        projections.add(rexBuilder.makeCast(targetArrayType, multisetRef));
+      } 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           |