Rename fields to intended ones after aggregation

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

Author: yuancu

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

@@ -997,13 +997,49 @@ private Pair<List<RexNode>, List<AggCall>> aggregateWithTrimming(
     // because that Mapping only works for RexNode, but we need both AggCall and RexNode list.
     Pair<List<RexNode>, List<AggCall>> reResolved =
         resolveAttributesForAggregation(groupExprList, aggExprList, context);
-
+    List<String> names = getGroupKeyNamesAfterAggregation(reResolved.getLeft());
     context.relBuilder.aggregate(
         context.relBuilder.groupKey(reResolved.getLeft()), reResolved.getRight());
+    // During aggregation, Calcite projects both input dependencies and output group-by fields.
+    // When names conflict, Calcite adds numeric suffixes (e.g., "value0").
+    // Apply explicit renaming to restore the intended aliases.
+    context.relBuilder.rename(names);
 
     return Pair.of(reResolved.getLeft(), reResolved.getRight());
   }
 
+  private List<String> getGroupKeyNamesAfterAggregation(List<RexNode> nodes) {
+    List<RexNode> reordered = new ArrayList<>();
+    List<RexNode> left = new ArrayList<>();
+    for (RexNode n : nodes) {
+      if (isInputRef(n)) {
+        reordered.add(n);
+      } else {
+        left.add(n);
+      }
+    }
+    reordered.addAll(left);
+    return reordered.stream()
+        .map(this::extractAliasLiteral)
+        .flatMap(Optional::stream)
+        .map(RexLiteral::stringValue)
+        .toList();
+  }
+
+  /**
+   * Immitates registerExpression of {@link RelBuilder.Registrar} to derive the output order of
+   * group-by keys after aggregation
+   */
+  private boolean isInputRef(RexNode node) {
+    return switch (node.getKind()) {
+      case AS, DESCENDING, NULLS_FIRST, NULLS_LAST -> {
+        final List<RexNode> operands = ((RexCall) node).operands;
+        yield isInputRef(operands.getFirst());
+      }
+      default -> node instanceof RexInputRef;
+    };
+  }
+
   /**
    * Resolve attributes for aggregation.
    *
@@ -1102,7 +1138,7 @@ public RelNode visitAggregation(Aggregation node, CalcitePlanContext context) {
         aggregationAttributes.getLeft().stream()
             .map(this::extractAliasLiteral)
             .flatMap(Optional::stream)
-            .map(ref -> ((RexLiteral) ref).getValueAs(String.class))
+            .map(ref -> ref.getValueAs(String.class))
             .map(context.relBuilder::field)
             .map(f -> (RexNode) f)
             .toList();

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4580.yml

@@ -0,0 +1,81 @@
+setup:
+  - do:
+      indices.create:
+        index: time_test
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : true
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{"index": {"_index": "time_test"}}'
+          - '{"category":"A","value":1000,"@timestamp":"2024-01-01T00:00:00Z"}'
+          - '{"index": {"_index": "time_test"}}'
+          - '{"category":"B","value":2000,"@timestamp":"2024-01-01T00:05:00Z"}'
+          - '{"index": {"_index": "time_test"}}'
+          - '{"category":"A","value":1500,"@timestamp":"2024-01-01T00:10:00Z"}'
+          - '{"index": {"_index": "time_test"}}'
+          - '{"category":"C","value":3000,"@timestamp":"2024-01-01T00:20:00Z"}'
+
+---
+teardown:
+  - do:
+      query.settings:
+        body:
+          transient:
+            plugins.calcite.enabled : false
+
+---
+"Test span aggregation with field name collision - basic case":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=time_test | stats count() by span(value, 1000) as value
+
+  - match: { total: 3 }
+  - match: { schema: [{"name": "count()", "type": "bigint"}, {"name": "value", "type": "bigint"}] }
+  - match: { datarows: [[2, 1000], [1, 2000], [1, 3000]] }
+
+---
+"Test span aggregation with field name collision - multiple aggregations":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=time_test | stats count(), avg(value) by span(value, 1000) as value
+
+  - match: { total: 3 }
+  - match: { schema: [{"name": "count()", "type": "bigint"}, {"name": "avg(value)", "type": "double"}, {"name": "value", "type": "bigint"}] }
+  - match: { datarows: [[2, 1250.0, 1000], [1, 2000.0, 2000], [1, 3000.0, 3000]] }
+
+---
+"Test span aggregation without name collision - multiple group-by":
+  - skip:
+      features:
+        - headers
+        - allowed_warnings
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=time_test | stats count() by span(@timestamp, 10min) as @timestamp, category, value
+
+  - match: { total: 4 }
+  - match: { schema: [{"name": "count()", "type": "bigint"}, {"name": "@timestamp", "type": "timestamp"}, {"name": "category", "type": "string"}, {"name": "value", "type": "bigint"}] }
+  - match: { datarows: [[1, "2024-01-01 00:00:00", "A", 1000], [1, "2024-01-01 00:10:00", "A", 1500], [1, "2024-01-01 00:00:00", "B", 2000], [1, "2024-01-01 00:20:00", "C", 3000]] }