[Feature] implement transpose command as in the roadmap #4786

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

@@ -99,6 +99,7 @@
 import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
 import org.opensearch.sql.ast.tree.TableFunction;
+import org.opensearch.sql.ast.tree.Transpose;
 import org.opensearch.sql.ast.tree.Trendline;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
 import org.opensearch.sql.ast.tree.Values;
@@ -704,6 +705,11 @@ public LogicalPlan visitML(ML node, AnalysisContext context) {
     return new LogicalML(child, node.getArguments());
   }
 
+  @Override
+  public LogicalPlan visitTranspose(Transpose node, AnalysisContext context) {
+    throw getOnlyForCalciteException("Transpose");
+  }
+
   @Override
   public LogicalPlan visitBin(Bin node, AnalysisContext context) {
     throw getOnlyForCalciteException("Bin");

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

@@ -86,6 +86,7 @@
 import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
 import org.opensearch.sql.ast.tree.TableFunction;
+import org.opensearch.sql.ast.tree.Transpose;
 import org.opensearch.sql.ast.tree.Trendline;
 import org.opensearch.sql.ast.tree.Values;
 import org.opensearch.sql.ast.tree.Window;
@@ -282,6 +283,10 @@ public T visitReverse(Reverse node, C context) {
     return visitChildren(node, context);
   }
 
+  public T visitTranspose(Transpose node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitChart(Chart node, C context) {
     return visitChildren(node, context);
   }

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

@@ -59,6 +59,7 @@
 import org.opensearch.sql.ast.tree.Sort;
 import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
+import org.opensearch.sql.ast.tree.Transpose;
 import org.opensearch.sql.ast.tree.Trendline;
 import org.opensearch.sql.ast.tree.UnresolvedPlan;
 import org.opensearch.sql.ast.tree.Values;
@@ -554,6 +555,12 @@ public Node visitTrendline(Trendline node, FieldResolutionContext context) {
     return node;
   }
 
+  @Override
+  public Node visitTranspose(Transpose node, FieldResolutionContext context) {
+    visitChildren(node, context);
+    return node;
+  }
+
   @Override
   public Node visitChart(Chart node, FieldResolutionContext context) {
     Set<String> chartFields = extractFieldsFromAggregation(node.getAggregationFunction());

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

@@ -0,0 +1,68 @@
+/*
+ * 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 lombok.*;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.expression.Argument;
+import org.opensearch.sql.common.utils.StringUtils;
+
+/** AST node represent Transpose operation. */
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+public class Transpose extends UnresolvedPlan {
+  private final @NonNull java.util.Map<String, Argument> arguments;
+  private UnresolvedPlan child;
+  private static final int MAX_LIMIT_TRANSPOSE = 10000;
+  private static final int DEFAULT_MAX_ROWS = 5;
+  private static final String DEFAULT_COLUMN_NAME = "column";
+  private final int maxRows;
+  private final String columnName;
+
+  public Transpose(java.util.Map<String, Argument> arguments) {
+
+    this.arguments = arguments;
+    int tempMaxRows = DEFAULT_MAX_ROWS;
+    if (arguments.containsKey("number") && arguments.get("number").getValue() != null) {
+      try {
+        tempMaxRows = Integer.parseInt(arguments.get("number").getValue().toString());
+      } catch (NumberFormatException e) {
+        // log warning and use default
+
+      }
+    }
+    maxRows = tempMaxRows;
+    if (maxRows > MAX_LIMIT_TRANSPOSE) {
+      throw new IllegalArgumentException(
+          StringUtils.format("Maximum limit to transpose is %s", MAX_LIMIT_TRANSPOSE));
+    }
+    if (arguments.containsKey("columnName") && arguments.get("columnName").getValue() != null) {
+      columnName = arguments.get("columnName").getValue().toString();
+    } else {
+      columnName = DEFAULT_COLUMN_NAME;
+    }
+  }
+
+  @Override
+  public Transpose attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  @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.visitTranspose(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;
@@ -62,6 +63,7 @@
 import org.apache.calcite.rex.RexWindowBounds;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.fun.SqlTrimFunction;
 import org.apache.calcite.sql.type.ArraySqlType;
 import org.apache.calcite.sql.type.MapSqlType;
 import org.apache.calcite.sql.type.SqlTypeFamily;
@@ -679,6 +681,76 @@ public RelNode visitReverse(
     return context.relBuilder.peek();
   }
 
+  @Override
+  public RelNode visitTranspose(
+      org.opensearch.sql.ast.tree.Transpose node, CalcitePlanContext context) {
+
+    visitChildren(node, context);
+
+    int maxRows =
+        Optional.ofNullable(node.getMaxRows())
+            .filter(r -> r > 0)
+            .orElseThrow(() -> new IllegalArgumentException("maxRows must be positive"));
+
+    String columnName = node.getColumnName();
+    List<String> fieldNames =
+        context.relBuilder.peek().getRowType().getFieldNames().stream()
+            .filter(fieldName -> !isMetadataField(fieldName))
+            .toList();
+
+    RelBuilder b = context.relBuilder;
+    RexBuilder rx = context.rexBuilder;
+    RelDataType varchar = rx.getTypeFactory().createSqlType(SqlTypeName.VARCHAR);
+
+    // Step 1: ROW_NUMBER
+    b.projectPlus(
+        b.aggregateCall(SqlStdOperatorTable.ROW_NUMBER)
+            .over()
+            .rowsTo(RexWindowBounds.CURRENT_ROW)
+            .as(PlanUtils.ROW_NUMBER_COLUMN_FOR_TRANSPOSE));
+
+    // Step 2: UNPIVOT
+    b.unpivot(
+        false,
+        ImmutableList.of("value"),
+        ImmutableList.of(columnName),
+        fieldNames.stream()
+            .map(
+                f ->
+                    Map.entry(
+                        ImmutableList.of(rx.makeLiteral(f)),
+                        ImmutableList.of((RexNode) rx.makeCast(varchar, b.field(f), true))))
+            .collect(Collectors.toList()));
+
+    // Step 3: Trim spaces from columnName column before pivot
+
+    RexNode trimmedColumnName =
+        context.rexBuilder.makeCall(
+            SqlStdOperatorTable.TRIM,
+            context.rexBuilder.makeFlag(SqlTrimFunction.Flag.BOTH),
+            context.rexBuilder.makeLiteral(" "),
+            b.field(columnName));
+
+    // Step 4: PIVOT
+    b.pivot(
+        b.groupKey(trimmedColumnName),
+        ImmutableList.of(b.max(b.field("value"))),
+        ImmutableList.of(b.field(PlanUtils.ROW_NUMBER_COLUMN_FOR_TRANSPOSE)),
+        IntStream.rangeClosed(1, maxRows)
+            .mapToObj(i -> Map.entry("row " + i, ImmutableList.of((RexNode) b.literal(i))))
+            .collect(Collectors.toList()));
+
+    // Step 4: RENAME
+    List<String> cleanNames = new ArrayList<>();
+    cleanNames.add(columnName);
+    for (int i = 1; i <= maxRows; i++) {
+      cleanNames.add("row " + i);
+    }
+    b.rename(cleanNames);
+
+    return b.peek();
+  }
+
   @Override
   public RelNode visitBin(Bin node, CalcitePlanContext context) {
     visitChildren(node, context);

core/src/main/java/org/opensearch/sql/calcite/utils/PlanUtils.java

@@ -80,6 +80,7 @@ public interface PlanUtils {
   String ROW_NUMBER_COLUMN_FOR_SUBSEARCH = "_row_number_subsearch_";
   String ROW_NUMBER_COLUMN_FOR_STREAMSTATS = "__stream_seq__";
   String ROW_NUMBER_COLUMN_FOR_CHART = "_row_number_chart_";
+  String ROW_NUMBER_COLUMN_FOR_TRANSPOSE = "_row_number_transpose_";
 
   static SpanUnit intervalUnitToSpanUnit(IntervalUnit unit) {
     return switch (unit) {

docs/category.json

@@ -43,6 +43,7 @@
     "user/ppl/cmd/timechart.md",
     "user/ppl/cmd/top.md",
     "user/ppl/cmd/trendline.md",
+    "user/ppl/cmd/transpose.md",
     "user/ppl/cmd/where.md",
     "user/ppl/functions/aggregations.md",
     "user/ppl/functions/collection.md",

docs/user/ppl/cmd/transpose.md

@@ -0,0 +1,92 @@
+# transpose
+
+## Description
+
+The `transpose` command outputs the requested number of rows as columns, effectively transposing each result row into a corresponding column of field values.
+
+## Syntax
+
+transpose [int] [column_name=<string>]
+
+* number-of-rows: optional. The number of rows to transform into columns. Default value is 5. Maximum allowed is 10000.
+* column_name: optional. The name of the first column to use when transposing rows. This column holds the field names.
+
+
+## Example 1: Transpose results
+
+This example shows transposing wihtout any parameters. It transforms 5 rows into columns as default is 5.
+
+```ppl
+source=accounts
+| head 5 
+| fields account_number, firstname,  lastname, balance 
+| transpose
+```
+
+Expected output:
+
+```text
+fetched rows / total rows = 4/4
++----------------+-------+--------+---------+-------+-------+
+| column         | row 1 | row 2  | row 3   | row 4 | row 5 |
+|----------------+-------+--------+---------+-------+-------|
+| account_number | 1     | 6      | 13      | 18    | null  |
+| firstname      | Amber | Hattie | Nanette | Dale  | null  |
+| balance        | 39225 | 5686   | 32838   | 4180  | null  |
+| lastname       | Duke  | Bond   | Bates   | Adams | null  |
++----------------+-------+--------+---------+-------+-------+
+```
+
+## Example 2: Tranpose results up to a provided number of rows.
+
+This example shows transposing wihtout any parameters. It transforms 4 rows into columns as default is 5.
+
+```ppl
+source=accounts
+| head 5 
+| fields  account_number, firstname,  lastname, balance 
+| transpose 4
+```
+
+Expected output:
+
+```text
+fetched rows / total rows = 4/4
++----------------+-------+--------+---------+-------+
+| column         | row 1 | row 2  | row 3   | row 4 |
+|----------------+-------+--------+---------+-------|
+| account_number | 1     | 6      | 13      | 18    |
+| firstname      | Amber | Hattie | Nanette | Dale  |
+| balance        | 39225 | 5686   | 32838   | 4180  |
+| lastname       | Duke  | Bond   | Bates   | Adams |
++----------------+-------+--------+---------+-------+
+```
+
+## Example 2: Tranpose results up to a provided number of rows and first column with specified column name.
+
+This example shows transposing wihtout any parameters. It transforms 4 rows into columns as default is 5.
+
+```ppl
+source=accounts
+| head 5 
+| fields  account_number, firstname,  lastname, balance 
+| transpose 4 column_name='column_names'
+```
+
+Expected output:
+
+```text
+fetched rows / total rows = 4/4
++----------------+-------+--------+---------+-------+
+| column_names   | row 1 | row 2  | row 3   | row 4 |
+|----------------+-------+--------+---------+-------|
+| account_number | 1     | 6      | 13      | 18    |
+| firstname      | Amber | Hattie | Nanette | Dale  |
+| balance        | 39225 | 5686   | 32838   | 4180  |
+| lastname       | Duke  | Bond   | Bates   | Adams |
++----------------+-------+--------+---------+-------+
+```
+
+## Limitations
+
+The `transpose` command transforms up to a number of rows specified and if not enough rows found, it shows those transposed rows as null columns. 

docs/user/ppl/index.md

@@ -78,9 +78,10 @@ source=accounts
 | [describe command](cmd/describe.md) | 2.1 | stable (since 2.1) | Query the metadata of an index. |
 | [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. |
- | [addtotals command](cmd/addtotals.md) | 3.4 | stable (since 3.4) | Adds row and column values and appends a totals column and row. | 
- | [addcoltotals command](cmd/addcoltotals.md) | 3.4 | stable (since 3.4) | Adds column values and appends a totals 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. |
+
   - [Syntax](cmd/syntax.md) - PPL query structure and command syntax formatting  
 * **Functions**  
   - [Aggregation Functions](functions/aggregations.md)  

integ-test/src/test/java/org/opensearch/sql/calcite/CalciteNoPushdownIT.java

@@ -104,6 +104,7 @@
   CalciteTextFunctionIT.class,
   CalciteTopCommandIT.class,
   CalciteTrendlineCommandIT.class,
+  CalciteTransposeCommandIT.class,
   CalciteVisualizationFormatIT.class,
   CalciteWhereCommandIT.class,
   CalcitePPLTpchIT.class

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -2029,6 +2029,18 @@ public void testaddColTotalsExplain() throws IOException {
                 + "|  addcoltotals balance age label='GrandTotal'"));
   }
 
+  @Test
+  public void testTransposeExplain() throws IOException {
+    enabledOnlyWhenPushdownIsEnabled();
+    String expected = loadExpectedPlan("explain_transpose.yaml");
+    assertYamlEqualsIgnoreId(
+        expected,
+        explainQueryYaml(
+            "source=opensearch-sql_test_index_account"
+                + "| head 5 "
+                + "|  transpose 4 column_name='column_names'"));
+  }
+
   public void testComplexDedup() throws IOException {
     enabledOnlyWhenPushdownIsEnabled();
     String expected = loadExpectedPlan("explain_dedup_complex1.yaml");