Fix min/earliest order & fix non-accumulative agg for chart

Signed-off-by: Yuanchun Shen <yuanchu@amazon.com>

Author: yuancu

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

@@ -2013,37 +2013,35 @@ private String getAggFieldAlias(UnresolvedExpression aggregateFunction) {
   public RelNode visitChart(Chart node, CalcitePlanContext context) {
     visitChildren(node, context);
     ArgumentMap argMap = ArgumentMap.of(node.getArguments());
-    List<UnresolvedExpression> groupExprList = new ArrayList<>();
-    UnresolvedExpression span;
-    if (node.getColumnSplit() instanceof Span && node.getRowSplit() instanceof Span) {
-      throw new UnsupportedOperationException("It is not supported to have two span splits");
-    } else if (node.getRowSplit() instanceof Span) {
-      if (node.getColumnSplit() != null) {
-        groupExprList.add(node.getColumnSplit());
-      }
-      span = node.getRowSplit();
-    } else if (node.getColumnSplit() instanceof Span) {
-      if (node.getRowSplit() != null) {
-        groupExprList.add(node.getRowSplit());
-      }
-      span = node.getColumnSplit();
-    } else {
-      groupExprList.addAll(
-          Stream.of(node.getRowSplit(), node.getColumnSplit()).filter(Objects::nonNull).toList());
-      span = null;
-    }
+    List<UnresolvedExpression> groupExprList =
+        Stream.of(node.getRowSplit(), node.getColumnSplit()).filter(Objects::nonNull).toList();
     Boolean useNull = (Boolean) argMap.getOrDefault("usenull", Chart.DEFAULT_USE_NULL).getValue();
     Aggregation aggregation =
         new Aggregation(
             node.getAggregationFunctions(),
             List.of(),
             groupExprList,
-            span,
+            null,
             List.of(new Argument(Argument.BUCKET_NULLABLE, AstDSL.booleanLiteral(useNull))));
-    visitAggregation(aggregation, context);
+    RelNode aggregated = visitAggregation(aggregation, context);
+
+    // If row or column split does not present or limit equals 0, this is the same as `stats agg
+    // [group by col]`
+    Integer limit = (Integer) argMap.getOrDefault("limit", Chart.DEFAULT_LIMIT).getValue();
+    if (node.getRowSplit() == null || node.getColumnSplit() == null || Objects.equals(limit, 0)) {
+      return aggregated;
+    }
+
+    String aggFunctionName = getAggFunctionName(node.getAggregationFunctions().getFirst());
+    Optional<BuiltinFunctionName> aggFuncNameOptional = BuiltinFunctionName.of(aggFunctionName);
+    if (aggFuncNameOptional.isEmpty()) {
+      throw new IllegalArgumentException(
+          StringUtils.format("Unrecognized aggregation function: %s", aggFunctionName));
+    }
+    BuiltinFunctionName aggFunction = aggFuncNameOptional.get();
 
     // Convert the column split to string if necessary: column split was supposed to be pivoted to
-    // column names. This guarantees that its type being compatible with useother and usenull
+    // column names. This guarantees that its type compatibility with useother and usenull
     RelBuilder relBuilder = context.relBuilder;
     RexNode colSplit = relBuilder.field(2);
     String columSplitName = relBuilder.peek().getRowType().getFieldNames().getLast();
@@ -2055,14 +2053,7 @@ public RelNode visitChart(Chart node, CalcitePlanContext context) {
               columSplitName);
     }
     relBuilder.project(relBuilder.field(0), relBuilder.field(1), colSplit);
-    RelNode aggregated = relBuilder.peek();
-
-    // If row or column split does not present or limit equals 0, this is the same as `stats agg
-    // [group by col]`
-    Integer limit = (Integer) argMap.getOrDefault("limit", Chart.DEFAULT_LIMIT).getValue();
-    if (node.getRowSplit() == null || node.getColumnSplit() == null || Objects.equals(limit, 0)) {
-      return aggregated;
-    }
+    aggregated = relBuilder.peek();
 
     Boolean top = (Boolean) argMap.getOrDefault("top", Chart.DEFAULT_TOP).getValue();
     Boolean useOther =
@@ -2075,11 +2066,16 @@ public RelNode visitChart(Chart node, CalcitePlanContext context) {
     // 1: column split; 0: agg
     relBuilder.aggregate(
         relBuilder.groupKey(relBuilder.field(1)),
-        relBuilder.sum(relBuilder.field(0)).as("__grand_total__")); // results: group key, agg calls
+        buildAggCall(context.relBuilder, aggFunction, relBuilder.field(0))
+            .as("__grand_total__")); // results: group key, agg calls
     RexNode grandTotal = relBuilder.field("__grand_total__");
-    if (top) {
+    // Apply sorting: for MIN/EARLIEST, reverse the top/bottom logic
+    boolean smallestFirst =
+        aggFunction == BuiltinFunctionName.MIN || aggFunction == BuiltinFunctionName.EARLIEST;
+    if (top != smallestFirst) {
       grandTotal = relBuilder.desc(grandTotal);
     }
+
     // Always set it to null last so that it does not interfere with top / bottom calculation
     grandTotal = relBuilder.nullsLast(grandTotal);
     RexNode rowNum =
@@ -2138,7 +2134,7 @@ public RelNode visitChart(Chart node, CalcitePlanContext context) {
         relBuilder.alias(columnSplitExpr, columSplitName));
     relBuilder.aggregate(
         relBuilder.groupKey(relBuilder.field(1), relBuilder.field(2)),
-        relBuilder.sum(relBuilder.field(0)).as(aggFieldName));
+        buildAggCall(context.relBuilder, aggFunction, relBuilder.field(0)).as(aggFieldName));
     return relBuilder.peek();
   }
 

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.calcite.remote;
+
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.opensearch.sql.ppl.PPLIntegTestCase;
+
+import java.io.IOException;
+
+import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
+import static org.opensearch.sql.util.MatcherUtils.assertJsonEquals;
+import static org.opensearch.sql.util.MatcherUtils.rows;
+import static org.opensearch.sql.util.MatcherUtils.schema;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifySchema;
+
+public class CalciteChartCommandIT extends PPLIntegTestCase {
+    @Override
+    protected void init() throws Exception {
+        super.init();
+        enableCalcite();
+        loadIndex(Index.BANK);
+        loadIndex(Index.BANK_WITH_NULL_VALUES);
+        loadIndex(Index.OTELLOGS);
+    }
+
+    @Test
+    public void testChartWithSingleGroupKey() throws IOException {
+        JSONObject result1 = executeQuery(String.format("source=%s | chart avg(balance) by gender", TEST_INDEX_BANK));
+        verifySchema(
+                result1,
+                schema("avg(balance)", "double"),
+                schema("gender", "string"));
+        verifyDataRows(result1, rows(40488, "F"), rows(16377.25, "M"));
+        JSONObject result2 = executeQuery(String.format("source=%s | chart avg(balance) over gender", TEST_INDEX_BANK));
+        assertJsonEquals(result1.toString(), result2.toString());
+    }
+
+    @Test
+    public void testChartWithMultipleGroupKeys() throws IOException {
+        JSONObject result1 = executeQuery(String.format("source=%s | chart avg(balance) by gender, age", TEST_INDEX_BANK));
+        verifySchema(
+                result1,
+                schema("avg(balance)", "double"),
+                schema("gender", "string"),
+                schema("age", "string"));
+        verifyDataRows(result1, rows(40488, "F", "36"), rows(16377.25, "M", 36));
+        JSONObject result2 = executeQuery(String.format("source=%s | chart avg(balance) over gender, age", TEST_INDEX_BANK));
+        assertJsonEquals(result1.toString(), result2.toString());
+    }
+
+    // TODOs:
+    // Param nullstr: source=opensearch-sql_test_index_bank_with_null_values | eval age = cast(age as string) |  chart nullstr='nil' max(account_number) over gender by age
+    // Param usenull: source=opensearch-sql_test_index_bank_with_null_values | eval age = cast(age as string) |  chart usenull=false nullstr='nil' max(account_number) over gender by age
+    // Param limit = 0: source=bank | chart limit=0 avg(balance) over state by gender
+    // SPAN:
+
+}