Window range - corner case of offset equals to window size

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

@@ -455,6 +455,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] = aggFactories[i].finalizeComputation(aggs[i].get());
+ }
+ ++resultStorageIndex;
+ }
+
  // Prime the results
  if (rowIdProvider.get() < numRows) {
  for (int i = 0; i < aggs.length; i++) {
@@ -542,6 +551,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] = aggFactories[i].finalizeComputation(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]