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 count. Default to
ROWS UNBOUNDED PRECEDING .. CURRENT ROW for aggregates in the SQL parser
(ranking functions ignore frames), and add equals/hashCode to
CurrentRowWindowBound. 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,32 @@ GROUP BY department HAVING MAX(age) > 30 AND MIN(age) < 50
             """);
   }
 
+  @Test
+  public void testCountDistinctWindowWithOrderBy() {
+    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 ROWS UNBOUNDED PRECEDING)])
+              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 ROWS UNBOUNDED PRECEDING)])
+              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

@@ -28,6 +28,16 @@ public boolean isPreceding() {
   public static class CurrentRowWindowBound extends WindowBound {
     CurrentRowWindowBound() {}
 
+    @Override
+    public boolean equals(Object o) {
+      return this == o || o instanceof CurrentRowWindowBound;
+    }
+
+    @Override
+    public int hashCode() {
+      return CurrentRowWindowBound.class.hashCode();
+    }
+
     @Override
     public String toString() {
       return "CURRENT ROW";

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

@@ -222,7 +222,16 @@ 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);
+    // SQL standard: an aggregate window function with ORDER BY but no explicit frame defaults to a
+    // running window (ROWS UNBOUNDED PRECEDING AND CURRENT ROW), not the whole partition. The
+    // grammar exposes no frame clause and the AST default is whole-partition (added for PPL), so
+    // apply the running default here. Ranking functions (RANK/DENSE_RANK/ROW_NUMBER) ignore frames.
+    if (function instanceof AggregateFunction && !sortList.isEmpty()) {
+      windowFunction.setWindowFrame(WindowFrame.toCurrentRow());
+    }
+    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,15 @@ public void canBuildWindowFunctionWithoutOrderBy() {
 
   @Test
   public void canBuildAggregateWindowFunction() {
-    assertEquals(
-        window(
+    WindowFunction expected =
+        new WindowFunction(
             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(ImmutablePair.of(new SortOption(null, null), qualifiedName("age"))));
+    // Aggregate window with ORDER BY defaults to a running frame (UNBOUNDED PRECEDING .. CURRENT
+    // ROW) rather than the whole partition.
+    expected.setWindowFrame(WindowFrame.toCurrentRow());
+    assertEquals(expected, buildExprAst("AVG(age) OVER (PARTITION BY state ORDER BY age)"));
   }
 
   @Test