[multistage] Add ROWS support for aggregation window functions with ORDER BY within OVER

pinot-query-planner/src/main/java/org/apache/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java

@@ -177,8 +177,9 @@ private void validateWindowFrames(Window.Group windowGroup) {
     // Has ROWS only aggregation call kind (e.g. ROW_NUMBER)?
     boolean isRowsOnlyTypeAggregateCall = isRowsOnlyAggregationCallType(windowGroup.aggCalls);
     // For Phase 1 only the default frame is supported
-    Preconditions.checkState(!windowGroup.isRows || isRowsOnlyTypeAggregateCall,
-        "Default frame must be of type RANGE and not ROWS unless this is a ROWS only aggregation function");
+    Preconditions.checkState(!windowGroup.isRows || isRowsOnlyTypeAggregateCall
+            || !windowGroup.orderKeys.getKeys().isEmpty(), "Default frame must be of type RANGE and not ROWS unless "
+        + "this is a ROWS only aggregation function or ORDER BY is included");
     Preconditions.checkState(windowGroup.lowerBound.isPreceding() && windowGroup.lowerBound.isUnbounded(),
         String.format("Lower bound must be UNBOUNDED PRECEDING but it is: %s", windowGroup.lowerBound));
     if (windowGroup.orderKeys.getKeys().isEmpty() && !isRowsOnlyTypeAggregateCall) {

pinot-query-planner/src/test/java/org/apache/pinot/query/QueryEnvironmentTestBase.java

@@ -89,7 +89,7 @@ public void setUp() {
 
   @DataProvider(name = "testQueryDataProvider")
   protected Object[][] provideQueries() {
-    return new Object[][]{
+    return new Object[][] {
         new Object[]{"SELECT * FROM a UNION SELECT * FROM b"},
         new Object[]{"SELECT * FROM a UNION ALL SELECT * FROM b"},
         new Object[]{"SELECT * FROM a INTERSECT SELECT * FROM b"},
@@ -171,6 +171,10 @@ protected Object[][] provideQueries() {
         new Object[]{"SELECT RANK() OVER(PARTITION BY a.col2 ORDER BY a.col1) FROM a"},
         new Object[]{"SELECT DENSE_RANK() OVER(ORDER BY a.col1) FROM a"},
         new Object[]{"SELECT a.col1, SUM(a.col3) OVER (ORDER BY a.col2), MIN(a.col3) OVER (ORDER BY a.col2) FROM a"},
+        new Object[]{"SELECT SUM(a.col3) OVER(PARTITION BY a.col1 ORDER BY a.col2 ROWS BETWEEN UNBOUNDED "
+            + "PRECEDING AND CURRENT ROW) FROM a"},
+        new Object[]{"SELECT SUM(a.col3) OVER(ORDER BY a.col2 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT "
+            + "ROW) FROM a"},
         new Object[]{
             "SELECT /*+ aggOptions(is_partitioned_by_group_by_keys='true') */ a.col3, a.col1, SUM(b.col3) "
                 + "FROM a JOIN b ON a.col3 = b.col3 GROUP BY a.col3, a.col1"