Evaluate window aggregate function only once per each group when possible

h2/src/main/org/h2/expression/aggregate/AbstractAggregate.java

@@ -19,6 +19,7 @@ import org.h2.expression.analysis.WindowFrame;
 import org.h2.expression.analysis.WindowFrameBound;
 import org.h2.expression.analysis.WindowFrameBoundType;
 import org.h2.expression.analysis.WindowFrameExclusion;
+import org.h2.expression.analysis.WindowFrameUnits;
 import org.h2.table.ColumnResolver;
 import org.h2.table.TableFilter;
 import org.h2.value.Value;
@@ -86,9 +87,16 @@ public abstract class AbstractAggregate extends DataAnalysisOperation {
     protected void getOrderedResultLoop(Session session, HashMap<Integer, Value> result, ArrayList<Value[]> ordered,
             int rowIdColumn) {
         WindowFrame frame = over.getWindowFrame();
+        /*
+         * With RANGE (default) or GROUPS units and EXCLUDE GROUP or EXCLUDE NO
+         * OTHERS (default) exclusion all rows in the group have the same value
+         * of window aggregate function.
+         */
+        boolean grouped = frame == null
+                || frame.getUnits() != WindowFrameUnits.ROWS && frame.getExclusion().isGroupOrNoOthers();
         if (frame == null) {
             assert over.getOrderBy() != null;
-            aggregateFastPartition(session, result, ordered, rowIdColumn);
+            aggregateFastPartition(session, result, ordered, rowIdColumn, grouped);
             return;
         }
         if (frame.getStarting().getType() == WindowFrameBoundType.UNBOUNDED_PRECEDING
@@ -97,28 +105,28 @@ public abstract class AbstractAggregate extends DataAnalysisOperation {
             if (following != null && following.getType() == WindowFrameBoundType.UNBOUNDED_FOLLOWING) {
                 aggregateWholePartition(session, result, ordered, rowIdColumn);
             } else {
-                aggregateFastPartition(session, result, ordered, rowIdColumn);
+                aggregateFastPartition(session, result, ordered, rowIdColumn, grouped);
             }
             return;
         }
         // All other types of frames (slow)
         int size = ordered.size();
-        for (int i = 0; i < size; i++) {
+        for (int i = 0; i < size;) {
             Object aggregateData = createAggregateData();
             for (Iterator<Value[]> iter = WindowFrame.iterator(over, session, ordered, getOverOrderBySort(), i,
                     false); iter.hasNext();) {
                 updateFromExpressions(session, aggregateData, iter.next());
             }
-            result.put(ordered.get(i)[rowIdColumn].getInt(), getAggregatedValue(session, aggregateData));
+            i = processGroup(session, result, ordered, rowIdColumn, i, size, aggregateData, grouped);
         }
     }
 
     private void aggregateFastPartition(Session session, HashMap<Integer, Value> result, ArrayList<Value[]> ordered,
-            int rowIdColumn) {
+            int rowIdColumn, boolean grouped) {
         Object aggregateData = createAggregateData();
         int size = ordered.size();
         int lastIncludedRow = -1;
-        for (int i = 0; i < size; i++) {
+        for (int i = 0; i < size;) {
             int newLast = WindowFrame.getEndIndex(over, session, ordered, getOverOrderBySort(), i);
             assert newLast >= lastIncludedRow;
             if (newLast > lastIncludedRow) {
@@ -127,10 +135,21 @@ public abstract class AbstractAggregate extends DataAnalysisOperation {
                 }
                 lastIncludedRow = newLast;
             }
-            result.put(ordered.get(i)[rowIdColumn].getInt(), getAggregatedValue(session, aggregateData));
+            i = processGroup(session, result, ordered, rowIdColumn, i, size, aggregateData, grouped);
         }
     }
 
+    private int processGroup(Session session, HashMap<Integer, Value> result, ArrayList<Value[]> ordered,
+            int rowIdColumn, int i, int size, Object aggregateData, boolean grouped) {
+        Value[] firstRowInGroup = ordered.get(i), currentRowInGroup = firstRowInGroup;
+        Value r = getAggregatedValue(session, aggregateData);
+        do {
+            result.put(currentRowInGroup[rowIdColumn].getInt(), r);
+        } while (++i < size && grouped
+                && overOrderBySort.compare(firstRowInGroup, currentRowInGroup = ordered.get(i)) == 0);
+        return i;
+    }
+
     private void aggregateWholePartition(Session session, HashMap<Integer, Value> result, ArrayList<Value[]> ordered,
             int rowIdColumn) {
         // Aggregate values from the whole partition

h2/src/main/org/h2/expression/analysis/WindowFrameExclusion.java

@@ -38,6 +38,17 @@ public enum WindowFrameExclusion {
         this.sql = sql;
     }
 
+    /**
+     * Returns true if this exclusion clause excludes or includes the whole
+     * group.
+     *
+     * @return true if this exclusion clause is {@link #EXCLUDE_GROUP} or
+     *         {@link #EXCLUDE_NO_OTHERS}
+     */
+    public boolean isGroupOrNoOthers() {
+        return this == WindowFrameExclusion.EXCLUDE_GROUP || this == EXCLUDE_NO_OTHERS;
+    }
+
     /**
      * Returns SQL representation.
      *