Mvexpand feature

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

@@ -76,6 +76,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.MvExpand;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -700,6 +701,11 @@ public LogicalPlan visitExpand(Expand expand, AnalysisContext context) {
     throw getOnlyForCalciteException("Expand");
   }
 
+  @Override
+  public LogicalPlan visitMvExpand(MvExpand node, AnalysisContext context) {
+    throw getOnlyForCalciteException("MvExpand");
+  }
+
   /** Build {@link LogicalTrendline} for Trendline command. */
   @Override
   public LogicalPlan visitTrendline(Trendline node, AnalysisContext context) {

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

@@ -64,6 +64,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.MvExpand;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -441,4 +442,8 @@ public T visitAppend(Append node, C context) {
   public T visitMultisearch(Multisearch node, C context) {
     return visitChildren(node, context);
   }
+
+  public T visitMvExpand(MvExpand node, C context) {
+    return visitChildren(node, context);
+  }
 }

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

@@ -61,6 +61,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.MvExpand;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
 import org.opensearch.sql.ast.tree.Project;
@@ -135,6 +136,10 @@ public Expand expand(UnresolvedPlan input, Field field, String alias) {
     return new Expand(field, alias).attach(input);
   }
 
+  public static UnresolvedPlan mvexpand(UnresolvedPlan input, Field field, Integer limit) {
+    return new MvExpand(field, limit);
+  }
+
   public static UnresolvedPlan projectWithArg(
       UnresolvedPlan input, List<Argument> argList, UnresolvedExpression... projectList) {
     return new Project(Arrays.asList(projectList), argList).attach(input);

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

@@ -0,0 +1,55 @@
+/*
+ * 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 an {@code mvexpand <field> [limit N]} operation. */
+@ToString
+@EqualsAndHashCode(callSuper = false)
+public class MvExpand extends UnresolvedPlan {
+
+  private UnresolvedPlan child;
+  @Getter private final Field field;
+  @Getter @Nullable private final Integer limit;
+
+  public MvExpand(Field field, @Nullable Integer limit) {
+    this.field = field;
+    this.limit = limit;
+  }
+
+  @Override
+  public MvExpand attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  public Field getField() {
+    return field;
+  }
+
+  @Nullable
+  public Integer getLimit() {
+    return limit;
+  }
+
+  @Override
+  public List<UnresolvedPlan> getChild() {
+    return this.child == null ? ImmutableList.of() : ImmutableList.of(this.child);
+  }
+
+  @Override
+  public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) {
+    return nodeVisitor.visitMvExpand(this, context);
+  }
+}

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

@@ -113,6 +113,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.MvExpand;
 import org.opensearch.sql.ast.tree.Paginate;
 import org.opensearch.sql.ast.tree.Parse;
 import org.opensearch.sql.ast.tree.Patterns;
@@ -1553,6 +1554,92 @@ private static void buildDedupNotNull(
     context.relBuilder.projectExcept(_row_number_dedup_);
   }
 
+  private void buildMvExpandRelNode(
+      RexInputRef arrayFieldRex, String arrayFieldName, String alias, CalcitePlanContext context) {
+
+    // 1. Capture left node and its schema BEFORE calling build()
+    RelNode leftNode = context.relBuilder.peek();
+    RelDataType leftSchema = leftNode.getRowType();
+
+    // 2. Create correlation variable
+    Holder<RexCorrelVariable> correlVariable = Holder.empty();
+    context.relBuilder.variable(correlVariable::set);
+
+    // 3. Find the array field index in the left schema by name (robust)
+    RelDataTypeField leftField = leftSchema.getField(arrayFieldName, false, false);
+    int arrayFieldIndexInLeft;
+    if (leftField != null) {
+      arrayFieldIndexInLeft = leftField.getIndex();
+    } else {
+      // fallback (best effort)
+      arrayFieldIndexInLeft = arrayFieldRex.getIndex();
+    }
+
+    // 4. Build correlated field access for the right-side projection
+    RexNode correlArrayFieldAccess =
+        context.relBuilder.field(
+            context.rexBuilder.makeCorrel(leftSchema, correlVariable.get().id),
+            arrayFieldIndexInLeft);
+
+    // 5. Build left and right nodes (leftBuilt is the original left, rightNode uncollects the
+    // array)
+    RelNode leftBuilt = context.relBuilder.build();
+    RelNode rightNode =
+        context
+            .relBuilder
+            .push(LogicalValues.createOneRow(context.relBuilder.getCluster()))
+            .project(List.of(correlArrayFieldAccess), List.of(arrayFieldName))
+            .uncollect(List.of(), false)
+            .build();
+
+    // 6. Compute a proper RexInputRef that refers to leftBuilt's rowType at the
+    // arrayFieldIndexInLeft.
+    //    Use rexBuilder.makeInputRef with leftBuilt.getRowType()
+    RexNode requiredColumnRef =
+        context.rexBuilder.makeInputRef(leftBuilt.getRowType(), arrayFieldIndexInLeft);
+
+    // 7. Correlate left and right using the proper required column ref
+    context
+        .relBuilder
+        .push(leftBuilt)
+        .push(rightNode)
+        .correlate(JoinRelType.INNER, correlVariable.get().id, List.of(requiredColumnRef));
+
+    // 8. Remove the original array field from the output by name using the builder's field()
+    //    (this ensures we remove the correct column instance)
+    RexNode toRemove;
+    try {
+      toRemove = context.relBuilder.field(arrayFieldName);
+    } catch (Exception e) {
+      // Fallback in case name lookup fails
+      toRemove = arrayFieldRex;
+    }
+    context.relBuilder.projectExcept(toRemove);
+
+    // 9. Optional rename into alias (same as your prior logic)
+    if (alias != null) {
+      tryToRemoveNestedFields(context);
+      RexInputRef expandedField = context.relBuilder.field(arrayFieldName);
+      List<String> names = new ArrayList<>(context.relBuilder.peek().getRowType().getFieldNames());
+      names.set(expandedField.getIndex(), alias);
+      context.relBuilder.rename(names);
+    }
+  }
+
+  @Override
+  public RelNode visitMvExpand(MvExpand node, CalcitePlanContext context) {
+    visitChildren(node, context);
+    Field arrayField = node.getField();
+    RexInputRef arrayFieldRex = (RexInputRef) rexVisitor.analyze(arrayField, context);
+
+    buildMvExpandRelNode(arrayFieldRex, arrayField.getField().toString(), null, context);
+
+    if (node.getLimit() != null) {
+      context.relBuilder.limit(0, node.getLimit());
+    }
+    return context.relBuilder.peek();
+  }
+
   @Override
   public RelNode visitWindow(Window node, CalcitePlanContext context) {
     visitChildren(node, context);

docs/user/ppl/cmd/mvexpand.rst

@@ -0,0 +1,180 @@
+=============
+mvexpand
+=============
+
+.. rubric:: Table of contents
+
+.. contents::
+   :local:
+   :depth: 2
+
+
+Description
+============
+| The ``mvexpand`` command expands each value in a multivalue (array) field into a separate row, similar to Splunk's `mvexpand` command.
+| For each document, every value in the specified field is returned as a new row. This is especially useful for log analytics and data exploration involving array fields.
+
+| Key features of ``mvexpand``:
+- Expands array fields into multiple rows, one per value.
+- Supports an optional ``limit`` parameter to restrict the number of expanded values per document.
+- Handles empty, null, and non-array fields gracefully.
+- Works as a streaming/distributable command for performance and scalability.
+
+Version
+=======
+3.3.0
+
+Syntax
+======
+mvexpand <field> [limit=<int>]
+
+* **field**: The multivalue (array) field to expand. (Required)
+* **limit**: Maximum number of values per document to expand. (Optional)
+
+Usage
+=====
+Basic expansion::
+
+    os> source=logs | mvexpand tags
+
+Expansion with limit::
+
+    os> source=docs | mvexpand ids limit=3
+
+Limitations
+===========
+- Only one field can be expanded per mvexpand command.
+- For non-array fields, the value is returned as-is.
+- For empty or null arrays, no rows are returned.
+- Large arrays may be subject to resource/memory limits; exceeding them results in an error or warning.
+
+Examples and Edge Cases
+=======================
+
+Example 1: Basic Expansion
+--------------------------
+Expand all values from an array field.
+
+Input document::
+
+    { "tags": ["error", "warning", "info"] }
+
+PPL query::
+
+    os> source=logs | mvexpand tags
+    fetched rows / total rows = 3/3
+    +--------+
+    | tags   |
+    +--------+
+    | error  |
+    | warning|
+    | info   |
+    +--------+
+
+Example 2: Expansion with Limit
+-------------------------------
+Limit the number of expanded values per document.
+
+Input document::
+
+    { "ids": [1, 2, 3, 4, 5] }
+
+PPL query::
+
+    os> source=docs | mvexpand ids limit=3
+    fetched rows / total rows = 3/3
+    +-----+
+    | ids |
+    +-----+
+    |  1  |
+    |  2  |
+    |  3  |
+    +-----+
+
+Example 3: Empty or Null Arrays
+------------------------------
+Handles documents with empty or null array fields.
+
+Input document::
+
+    { "tags": [] }
+
+PPL query::
+
+    os> source=logs | mvexpand tags
+    fetched rows / total rows = 0/0
+    +------+
+    | tags |
+    +------+
+    +------+
+
+Input document::
+
+    { "tags": null }
+
+PPL query::
+
+    os> source=logs | mvexpand tags
+    fetched rows / total rows = 0/0
+    +------+
+    | tags |
+    +------+
+    +------+
+
+Example 4: Non-array Field
+--------------------------
+If the field is a single value (not an array), mvexpand returns the value as-is.
+
+Input document::
+
+    { "tags": "error" }
+
+PPL query::
+
+    os> source=logs | mvexpand tags
+    fetched rows / total rows = 1/1
+    +-------+
+    | tags  |
+    +-------+
+    | error |
+    +-------+
+
+Example 5: Large Arrays and Memory Limits
+----------------------------------------
+If an array exceeds configured memory/resource limits, mvexpand returns an error.
+
+Input document::
+
+    { "ids": [1, 2, ..., 100000] }
+
+PPL query::
+
+    os> source=docs | mvexpand ids
+    Error: Memory/resource limit exceeded while expanding field 'ids'. Please reduce the array size or specify a limit.
+
+Example 6: Multiple Fields (Limitation)
+---------------------------------------
+mvexpand only supports expanding one field per command. To expand multiple fields, use multiple mvexpand commands or document the limitation.
+
+PPL query::
+
+    os> source=docs | mvexpand a | mvexpand b
+
+Example 7: Edge Case - Field Missing
+------------------------------------
+If the field does not exist in a document, no row is produced for that document.
+
+Input document::
+
+    { "other": [1,2] }
+
+PPL query::
+
+    os> source=docs | mvexpand tags
+    fetched rows / total rows = 0/0
+    +------+
+    | tags |
+    +------+
+    +------+
+
+---