Fixing corner case when only one of upper or lower range is unbounded…

processing/src/main/java/org/apache/druid/query/rowsandcols/semantic/DefaultFramedOnHeapAggregatable.java

@@ -454,6 +454,15 @@ private AppendableRowsAndColumns computeCumulativeAggregates(AggregatorFactory[]
       rowIdProvider.incrementAndGet();
     }
 
+    // Corner case when the numRows=upperOffset
+    // In such a case the priming of results is not needed
+    if (rowIdProvider.get() == numRows) {
+      for (int i = 0; i < aggs.length; i++) {
+        results[i][resultStorageIndex] = aggs[i].get();
+      }
+      ++resultStorageIndex;
+    }
+
     // Prime the results
     if (rowIdProvider.get() < numRows) {
       for (int i = 0; i < aggs.length; i++) {
@@ -541,6 +550,15 @@ private AppendableRowsAndColumns computeReverseCumulativeAggregates(AggregatorFa
       rowIdProvider.decrementAndGet();
     }
 
+    // Corner case when the numRows=loweOffset
+    // In such a case the priming of results is not needed
+    if (rowIdProvider.get() < 0) {
+      for (int i = 0; i < aggs.length; i++) {
+        results[i][resultStorageIndex] = aggs[i].get();
+      }
+      --resultStorageIndex;
+    }
+
     // Prime the results
     if (rowIdProvider.get() >= 0) {
       for (int i = 0; i < aggs.length; i++) {

processing/src/test/java/org/apache/druid/query/rowsandcols/semantic/FramedOnHeapAggregatableTest.java

@@ -392,6 +392,8 @@ public void testUnboundedWindowedAggregation()
         .validate(results);
   }
 
+
+
   @Test
   public void testCumulativeAggregation()
   {
@@ -460,6 +462,73 @@ public void testReverseCumulativeAggregation()
         .validate(results);
   }
 
+  @Test
+  public void testCumulativeAggregationUnbounded()
+  {
+    Map<String, Column> map = new LinkedHashMap<>();
+    map.put("intCol", new IntArrayColumn(new int[]{0}));
+    map.put("doubleCol", new DoubleArrayColumn(new double[]{0}));
+    map.put("objectCol", new ObjectArrayColumn(
+                new String[]{"a"},
+                ColumnType.STRING
+            )
+    );
+
+    RowsAndColumns rac = make(MapOfColumnsRowsAndColumns.fromMap(map));
+
+    FramedOnHeapAggregatable agger = FramedOnHeapAggregatable.fromRAC(rac);
+
+    final RowsAndColumns results = agger.aggregateAll(
+        new WindowFrame(WindowFrame.PeerType.ROWS, true, 1, false, 1, null),
+        new AggregatorFactory[]{
+            new LongMaxAggregatorFactory("cummMax", "intCol"),
+            new DoubleSumAggregatorFactory("cummSum", "doubleCol")
+        }
+    );
+
+    new RowsAndColumnsHelper()
+        .expectColumn("intCol", new int[]{0})
+        .expectColumn("doubleCol", new double[]{0})
+        .expectColumn("objectCol", new String[]{"a"}, ColumnType.STRING)
+        .expectColumn("cummMax", new long[]{0})
+        .expectColumn("cummSum", new double[]{0})
+        .allColumnsRegistered()
+        .validate(results);
+  }
+
+  @Test
+  public void testReverseCumulativeAggregationUnbound()
+  {
+    Map<String, Column> map = new LinkedHashMap<>();
+    map.put("intCol", new IntArrayColumn(new int[]{0}));
+    map.put("doubleCol", new DoubleArrayColumn(new double[]{0}));
+    map.put("objectCol", new ObjectArrayColumn(
+                new String[]{"a"},
+                ColumnType.STRING
+            )
+    );
+
+    RowsAndColumns rac = make(MapOfColumnsRowsAndColumns.fromMap(map));
+
+    FramedOnHeapAggregatable agger = FramedOnHeapAggregatable.fromRAC(rac);
+
+    final RowsAndColumns results = agger.aggregateAll(
+        new WindowFrame(WindowFrame.PeerType.ROWS, false, 1, true, 0, null),
+        new AggregatorFactory[]{
+            new LongMaxAggregatorFactory("cummMax", "intCol"),
+            new DoubleSumAggregatorFactory("cummSum", "doubleCol")
+        }
+    );
+
+    new RowsAndColumnsHelper()
+        .expectColumn("intCol", new int[]{0})
+        .expectColumn("doubleCol", new double[]{0})
+        .expectColumn("objectCol", new String[]{"a"}, ColumnType.STRING)
+        .expectColumn("cummMax", new long[]{0})
+        .expectColumn("cummSum", new double[]{0})
+        .allColumnsRegistered()
+        .validate(results);
+  }
 
 
   @Test

sql/src/test/resources/calcite/tests/window/window_unbounded_lower.sqlTest

@@ -0,0 +1,16 @@
+type: "operatorValidation"
+
+sql: |
+   select countryName, cityName,
+   count(cityName) over w cnt
+   from wikipedia
+   where countryName in ('Austria', 'Republic of Korea') and (cityName in ('Horsching', 'Vienna', 'Seoul', 'Jeonju'))
+   group by countryName, cityName
+   window w as (partition by cityName order by countryName asc rows between unbounded preceding and 1 following)
+
+
+expectedResults:
+  - ["Austria","Horsching",1]
+  - ["Republic of Korea","Jeonju",1]
+  - ["Republic of Korea","Seoul",1]
+  - ["Austria","Vienna",1]

sql/src/test/resources/calcite/tests/window/window_unbounded_upper.sqlTest

@@ -0,0 +1,16 @@
+type: "operatorValidation"
+
+sql: |
+   select countryName, cityName,
+   count(cityName) over w cnt
+   from wikipedia
+   where countryName in ('Austria', 'Republic of Korea') and (cityName in ('Horsching', 'Vienna', 'Seoul', 'Jeonju'))
+   group by countryName, cityName
+   window w as (partition by cityName order by countryName asc rows between 1 preceding and unbounded following)
+
+
+expectedResults:
+  - ["Austria","Horsching",1]
+  - ["Republic of Korea","Jeonju",1]
+  - ["Republic of Korea","Seoul",1]
+  - ["Austria","Vienna",1]