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;
@@ -533,6 +534,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,18 @@ public static List<Argument> defaultDedupArgs() {
         argument("consecutive", booleanLiteral(false)));
   }
 
+  public static MvCombine mvcombine(Field field) {
+    return new MvCombine(field, null, false);
+  }
+
+  public static MvCombine mvcombine(Field field, String delim) {
+    return new MvCombine(field, delim, false);
+  }
+
+  public static MvCombine mvcombine(Field field, String delim, boolean nomv) {
+    return new MvCombine(field, delim, nomv);
+  }
+
   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,47 @@
+/*
+ * 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;
+  private final boolean nomv;
+
+  public MvCombine(Field field, @Nullable String delim, boolean nomv) {
+    this.field = field;
+    this.delim = (delim == null) ? " " : delim;
+    this.nomv = nomv;
+  }
+
+  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;
@@ -122,6 +123,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;
@@ -3116,6 +3118,175 @@ 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:
+   *
+   * <p>Groups by all input fields except the target field. Aggregates target using {@code COLLECT}
+   * (MULTISET) and casts MULTISET -> ARRAY. Preserves original output column order. If {@code
+   * nomv=true}, renders target array as a scalar string via {@code ARRAY_TO_STRING} using {@code
+   * delim}. {@code delim} only affects output when {@code nomv=true}.
+   *
+   * @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
+   *     (or a delimited string when {@code nomv=true})
+   * @throws SemanticCheckException if the mvcombine target is not a direct field reference
+   */
+  @Override
+  public RelNode visitMvCombine(MvCombine node, CalcitePlanContext context) {
+    // 1) Lower child first
+    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) Accept delim (default is single space).
+    // NOTE: delim only affects single-value rendering when nomv=true.
+    final String delim = node.getDelim();
+
+    // 3) Resolve target field to an input ref (must be a direct field reference)
+    final Field targetField = node.getField();
+    final int targetIndex = resolveTargetIndex(targetField, context);
+    final String targetName = inputFieldNames.get(targetIndex);
+
+    // 4) Group key = all fields except target
+    final List<RexNode> groupExprs =
+        buildGroupExpressionsExcludingTarget(targetIndex, inputFieldNames, relBuilder);
+
+    // 5) Aggregate target values: COLLECT => MULTISET
+    performCollectAggregation(relBuilder, targetIndex, targetName, groupExprs);
+
+    // 6) Restore original output column order AND cast MULTISET -> ARRAY using RexBuilder
+    restoreColumnOrderWithArrayCast(
+        relBuilder, rexBuilder, input, inputFieldNames, targetIndex, groupExprs);
+
+    // 7) nomv=true converts multivalue output to a single delimited string.
+    if (node.isNomv()) {
+      applyNomvConversion(relBuilder, rexBuilder, inputFieldNames, targetIndex, delim);
+    }
+
+    return relBuilder.peek();
+  }
+
+  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();
+
+    // Validate the target field is scalar-ish. mvcombine outputs ARRAY<T>, so target must not
+    // already be ARRAY.
+    final RelDataType fieldType =
+        context.relBuilder.peek().getRowType().getFieldList().get(index).getType();
+
+    // Prefer Calcite type inspection instead of relying on rel type name string checks.
+    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;
+  }
+
+  private List<RexNode> buildGroupExpressionsExcludingTarget(
+      int targetIndex, List<String> inputFieldNames, RelBuilder relBuilder) {
+    final List<RexNode> groupExprs = new ArrayList<>();
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      if (i == targetIndex) continue;
+      groupExprs.add(relBuilder.field(i));
+    }
+    return groupExprs;
+  }
+
+  private void performCollectAggregation(
+      RelBuilder relBuilder, int targetIndex, String targetName, List<RexNode> groupExprs) {
+    final RelBuilder.AggCall aggCall =
+        relBuilder
+            .aggregateCall(SqlStdOperatorTable.COLLECT, relBuilder.field(targetIndex))
+            .as(targetName);
+
+    relBuilder.aggregate(relBuilder.groupKey(groupExprs), aggCall);
+  }
+
+  private void restoreColumnOrderWithArrayCast(
+      RelBuilder relBuilder,
+      RexBuilder rexBuilder,
+      RelNode input,
+      List<String> inputFieldNames,
+      int targetIndex,
+      List<RexNode> groupExprs) {
+
+    // After aggregate => [group fields..., targetAgg(multiset)]
+    final int aggPos = groupExprs.size();
+
+    final RelDataType elemType = input.getRowType().getFieldList().get(targetIndex).getType();
+    final RelDataType arrayType = relBuilder.getTypeFactory().createArrayType(elemType, -1);
+
+    final List<RexNode> projections = new ArrayList<>(inputFieldNames.size());
+    final List<String> projNames = new ArrayList<>(inputFieldNames.size());
+
+    int groupPos = 0;
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      projNames.add(inputFieldNames.get(i));
+      if (i == targetIndex) {
+        final RexNode multisetRef = relBuilder.field(aggPos); // MULTISET
+        projections.add(rexBuilder.makeCast(arrayType, multisetRef)); // ARRAY
+      } else {
+        projections.add(relBuilder.field(groupPos));
+        groupPos++;
+      }
+    }
+
+    relBuilder.project(projections, projNames, /* force= */ true);
+  }
+
+  private void applyNomvConversion(
+      RelBuilder relBuilder,
+      RexBuilder rexBuilder,
+      List<String> inputFieldNames,
+      int targetIndex,
+      String delim) {
+
+    final List<RexNode> nomvProjections = new ArrayList<>(inputFieldNames.size());
+    final List<String> nomvNames = new ArrayList<>(inputFieldNames.size());
+
+    // Build ARRAY<VARCHAR> type once
+    final RelDataType stringType =
+        relBuilder.getTypeFactory().createSqlType(org.apache.calcite.sql.type.SqlTypeName.VARCHAR);
+    final RelDataType stringArrayType = relBuilder.getTypeFactory().createArrayType(stringType, -1);
+
+    for (int i = 0; i < inputFieldNames.size(); i++) {
+      final String name = inputFieldNames.get(i);
+      nomvNames.add(name);
+
+      if (i == targetIndex) {
+        // At this point relBuilder.field(i) is ARRAY<elemType>. Cast to ARRAY<VARCHAR> so
+        // ARRAY_TO_STRING works.
+        final RexNode stringArray = rexBuilder.makeCast(stringArrayType, relBuilder.field(i));
+
+        nomvProjections.add(
+            relBuilder.call(
+                org.apache.calcite.sql.fun.SqlLibraryOperators.ARRAY_TO_STRING,
+                stringArray,
+                relBuilder.literal(delim)));
+      } else {
+        nomvProjections.add(relBuilder.field(i));
+      }
+    }
+
+    relBuilder.project(nomvProjections, nomvNames, /* 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           |