fix(calcite): Default SQL aggregate window frame to running window

Aggregate window functions with ORDER BY but no explicit frame defaulted
to the whole partition, so COUNT(DISTINCT x) OVER(ORDER BY y) returned the
total for every row instead of a running value. Default SQL aggregates to
RANGE UNBOUNDED PRECEDING .. CURRENT ROW, matching the SQL standard and the
non-Calcite engine's peer semantics (ranking functions ignore frames). PPL
windows carry no ORDER BY, so are unaffected.

Signed-off-by: Chen Dai <daichen@amazon.com>

Author: dai-chen

api/src/test/java/org/opensearch/sql/api/UnifiedQueryPlannerSqlV2Test.java

@@ -424,6 +424,33 @@ GROUP BY department HAVING MAX(age) > 30 AND MIN(age) < 50
             """);
   }
 
+  @Test
+  public void testCountDistinctWindowWithOrderBy() {
+    // No frame printed: RANGE .. CURRENT ROW is Calcite's default for ORDER BY.
+    givenQuery(
+            """
+            SELECT department, COUNT(DISTINCT name) OVER(ORDER BY department) FROM catalog.employees
+            """)
+        .assertPlan(
+            """
+            LogicalProject(department=[$3], COUNT(DISTINCT name) OVER(ORDER BY department)=[COUNT(DISTINCT $1) OVER (ORDER BY $3 NULLS FIRST)])
+              LogicalTableScan(table=[[catalog, employees]])
+            """);
+  }
+
+  @Test
+  public void testSumWindowWithPartitionAndOrderBy() {
+    givenQuery(
+            """
+            SELECT name, SUM(age) OVER(PARTITION BY department ORDER BY age) FROM catalog.employees
+            """)
+        .assertPlan(
+            """
+            LogicalProject(name=[$1], SUM(age) OVER(PARTITION BY department ORDER BY age)=[SUM($2) OVER (PARTITION BY $3 ORDER BY $2 NULLS FIRST)])
+              LogicalTableScan(table=[[catalog, employees]])
+            """);
+  }
+
   @Test
   public void testWindowOrderByDefaultsNullsFirst() {
     // Window function ORDER BY without explicit NULLS FIRST/LAST defaults to NULLS FIRST,

core/src/main/java/org/opensearch/sql/ast/expression/WindowBound.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.sql.ast.expression;
 
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
 public abstract class WindowBound {
@@ -25,6 +26,7 @@ public boolean isPreceding() {
     }
   }
 
+  @EqualsAndHashCode(callSuper = false)
   public static class CurrentRowWindowBound extends WindowBound {
     CurrentRowWindowBound() {}
 

core/src/main/java/org/opensearch/sql/ast/expression/WindowFrame.java

@@ -40,6 +40,13 @@ public static WindowFrame toCurrentRow() {
         AstDSL.stringLiteral("CURRENT ROW"));
   }
 
+  public static WindowFrame rangeToCurrentRow() {
+    return WindowFrame.of(
+        FrameType.RANGE,
+        AstDSL.stringLiteral("UNBOUNDED PRECEDING"),
+        AstDSL.stringLiteral("CURRENT ROW"));
+  }
+
   public static WindowFrame of(FrameType type, String lower, String upper) {
     return WindowFrame.of(type, AstDSL.stringLiteral(lower), AstDSL.stringLiteral(upper));
   }

sql/src/main/java/org/opensearch/sql/sql/parser/AstExpressionBuilder.java

@@ -222,7 +222,14 @@ public UnresolvedExpression visitWindowFunctionClause(WindowFunctionClauseContex
               .map(item -> ImmutablePair.of(createSortOption(item), visit(item.expression())))
               .collect(Collectors.toList());
     }
-    return new WindowFunction(visit(ctx.function), partitionByList, sortList);
+    UnresolvedExpression function = visit(ctx.function);
+    WindowFunction windowFunction = new WindowFunction(function, partitionByList, sortList);
+
+    // Aggregate window with ORDER BY defaults to a running RANGE frame (ranking ignores it).
+    if (function instanceof AggregateFunction && !sortList.isEmpty()) {
+      windowFunction.setWindowFrame(WindowFrame.rangeToCurrentRow());
+    }
+    return windowFunction;
   }
 
   @Override

sql/src/test/java/org/opensearch/sql/sql/parser/AstExpressionBuilderTest.java

@@ -44,6 +44,8 @@
 import org.opensearch.sql.ast.expression.DataType;
 import org.opensearch.sql.ast.expression.Literal;
 import org.opensearch.sql.ast.expression.RelevanceFieldList;
+import org.opensearch.sql.ast.expression.WindowFrame;
+import org.opensearch.sql.ast.expression.WindowFunction;
 import org.opensearch.sql.ast.tree.Sort.SortOption;
 import org.opensearch.sql.common.antlr.CaseInsensitiveCharStream;
 import org.opensearch.sql.common.antlr.SyntaxAnalysisErrorListener;
@@ -308,12 +310,23 @@ public void canBuildWindowFunctionWithoutOrderBy() {
 
   @Test
   public void canBuildAggregateWindowFunction() {
+    WindowFunction expected =
+        new WindowFunction(
+            aggregate("AVG", qualifiedName("age")),
+            ImmutableList.of(qualifiedName("state")),
+            ImmutableList.of(ImmutablePair.of(new SortOption(null, null), qualifiedName("age"))));
+    expected.setWindowFrame(WindowFrame.rangeToCurrentRow());
+    assertEquals(expected, buildExprAst("AVG(age) OVER (PARTITION BY state ORDER BY age)"));
+  }
+
+  @Test
+  public void canBuildAggregateWindowFunctionWithoutOrderBy() {
     assertEquals(
         window(
             aggregate("AVG", qualifiedName("age")),
             ImmutableList.of(qualifiedName("state")),
-            ImmutableList.of(ImmutablePair.of(new SortOption(null, null), qualifiedName("age")))),
-        buildExprAst("AVG(age) OVER (PARTITION BY state ORDER BY age)"));
+            ImmutableList.of()),
+        buildExprAst("AVG(age) OVER (PARTITION BY state)"));
   }
 
   @Test