Implement remaming types of window frames

h2/src/docsrc/help/help.csv

@@ -2550,11 +2550,8 @@ They also may require a lot of memory for large queries.
 "
 
 "Other Grammar","Window frame","
-[RANGE BETWEEN {
-    UNBOUNDED PRECEDING AND CURRENT ROW
-    |CURRENT ROW AND UNBOUNDED FOLLOWING
-    |UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-    }]
+ROWS|RANGE|GROUP
+{windowFramePreceding|BETWEEN windowFramePreceding AND windowFrameFollowing}
 [EXCLUDE {CURRENT ROW|GROUP|TIES|NO OTHERS}]
 ","
 A window frame clause.
@@ -2563,6 +2560,28 @@ Is currently supported only in aggregates and FIRST_VALUE(), LAST_VALUE(), and N
 RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW EXCLUDE GROUP
 "
 
+"Other Grammar","Window frame preceding","
+UNBOUNDED PRECEDING|value PRECEDING|CURRENT ROW
+","
+A window frame preceding clause.
+If value is specified it should be non-negative value or parameter.
+","
+UNBOUNDED PRECEDING
+1 PRECEDING
+CURRENT ROW
+"
+
+"Other Grammar","Window frame following","
+UNBOUNDED FOLLOWING|value FOLLOWING|CURRENT ROW
+","
+A window frame following clause.
+If value is specified it should be non-negative value or parameter.
+","
+UNBOUNDED FOLLOWING
+1 FOLLOWING
+CURRENT ROW
+"
+
 "Other Grammar","Term","
 value
     | columnName

h2/src/main/org/h2/command/Parser.java

@@ -3132,13 +3132,13 @@ public class Parser {
     private WindowFrameBound readWindowFrameStarting() {
         if (readIf("UNBOUNDED")) {
             read("PRECEDING");
-            return new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, 0);
+            return new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, null);
         }
         if (readIf("CURRENT")) {
             read("ROW");
-            return new WindowFrameBound(WindowFrameBoundType.CURRENT_ROW, 0);
+            return new WindowFrameBound(WindowFrameBoundType.CURRENT_ROW, null);
         }
-        int value = readNonNegativeInt();
+        Expression value = readValueOrParameter();
         read("PRECEDING");
         return new WindowFrameBound(WindowFrameBoundType.VALUE, value);
     }
@@ -3146,17 +3146,27 @@ public class Parser {
     private WindowFrameBound readWindowFrameFollowing() {
         if (readIf("UNBOUNDED")) {
             read("FOLLOWING");
-            return new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, 0);
+            return new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, null);
         }
         if (readIf("CURRENT")) {
             read("ROW");
-            return new WindowFrameBound(WindowFrameBoundType.CURRENT_ROW, 0);
+            return new WindowFrameBound(WindowFrameBoundType.CURRENT_ROW, null);
         }
-        int value = readNonNegativeInt();
+        Expression value = readValueOrParameter();
         read("FOLLOWING");
         return new WindowFrameBound(WindowFrameBoundType.VALUE, value);
     }
 
+    private Expression readValueOrParameter() {
+        int index = parseIndex;
+        Expression value = readExpression();
+        if (!(value instanceof ValueExpression) && !(value instanceof Parameter)) {
+            parseIndex = index;
+            throw getSyntaxError();
+        }
+        return value;
+    }
+
     private AggregateType getAggregateType(String name) {
         if (!identifiersToUpper) {
             // if not yet converted to uppercase, do it now

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

@@ -445,7 +445,7 @@ public abstract class AbstractAggregate extends Expression {
             int size = ordered.size();
             for (int i = 0; i < size; i++) {
                 Object aggregateData = createAggregateData();
-                for (Iterator<Value[]> iter = frame.iterator(ordered, getOverOrderBySort(), i, false); iter
+                for (Iterator<Value[]> iter = frame.iterator(session, ordered, getOverOrderBySort(), i, false); iter
                         .hasNext();) {
                     updateFromExpressions(session, aggregateData, iter.next());
                 }

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

@@ -66,7 +66,7 @@ public final class Window {
         this.partitionBy = partitionBy;
         this.orderBy = orderBy;
         if (frame == null) {
-            frame = new WindowFrame(WindowFrameUnits.RANGE, new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, 0),
+            frame = new WindowFrame(WindowFrameUnits.RANGE, new WindowFrameBound(WindowFrameBoundType.UNBOUNDED, null),
                     null, WindowFrameExclusion.EXCLUDE_NO_OTHERS);
         }
         this.frame = frame;

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

@@ -11,6 +11,10 @@ import java.util.Collections;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
 
+import org.h2.engine.Session;
+import org.h2.expression.BinaryOperation;
+import org.h2.expression.BinaryOperation.OpType;
+import org.h2.expression.ValueExpression;
 import org.h2.expression.aggregate.WindowFrameBound.WindowFrameBoundType;
 import org.h2.message.DbException;
 import org.h2.result.SortOrder;
@@ -154,6 +158,54 @@ public final class WindowFrame {
 
     private final WindowFrameExclusion exclusion;
 
+    private static int toGroupStart(ArrayList<Value[]> orderedRows, SortOrder sortOrder, int offset, int minOffset) {
+        Value[] row = orderedRows.get(offset);
+        while (offset > minOffset && sortOrder.compare(row, orderedRows.get(offset - 1)) == 0) {
+            offset--;
+        }
+        return offset;
+    }
+
+    private static int toGroupEnd(ArrayList<Value[]> orderedRows, SortOrder sortOrder, int offset, int maxOffset) {
+        Value[] row = orderedRows.get(offset);
+        while (offset < maxOffset && sortOrder.compare(row, orderedRows.get(offset + 1)) == 0) {
+            offset++;
+        }
+        return offset;
+    }
+
+    private static int getIntOffset(WindowFrameBound bound, Session session) {
+        int value = bound.getValue().getValue(session).getInt();
+        if (value < 0) {
+            throw DbException.getInvalidValueException("unsigned", value);
+        }
+        return value;
+    }
+
+    private static Value plus(Session session, ArrayList<Value[]> orderedRows, int currentRow, WindowFrameBound bound,
+            int index) {
+        return new BinaryOperation(OpType.PLUS, //
+                ValueExpression.get(orderedRows.get(currentRow)[index]),
+                ValueExpression.get(getValueOffset(bound, session))) //
+                        .optimize(session).getValue(session);
+    }
+
+    private static Value minus(Session session, ArrayList<Value[]> orderedRows, int currentRow, WindowFrameBound bound,
+            int index) {
+        return new BinaryOperation(OpType.MINUS, //
+                ValueExpression.get(orderedRows.get(currentRow)[index]),
+                ValueExpression.get(getValueOffset(bound, session))) //
+                        .optimize(session).getValue(session);
+    }
+
+    private static Value getValueOffset(WindowFrameBound bound, Session session) {
+        Value value = bound.getValue().getValue(session);
+        if (value.getSignum() < 0) {
+            throw DbException.getInvalidValueException("unsigned", value.getTraceSQL());
+        }
+        return value;
+    }
+
     /**
      * Creates new instance of window frame clause.
      *
@@ -202,6 +254,8 @@ public final class WindowFrame {
     /**
      * Returns iterator.
      *
+     * @param session
+     *            the session
      * @param orderedRows
      *            ordered rows
      * @param sortOrder
@@ -210,13 +264,15 @@ public final class WindowFrame {
      *            index of the current row
      * @param reverse
      *            whether iterator should iterate in reverse order
+     *
      * @return iterator
      */
-    public Iterator<Value[]> iterator(ArrayList<Value[]> orderedRows, SortOrder sortOrder, int currentRow,
-            boolean reverse) {
+    public Iterator<Value[]> iterator(Session session, ArrayList<Value[]> orderedRows, SortOrder sortOrder,
+            int currentRow, boolean reverse) {
         int size = orderedRows.size();
-        final int startIndex;
-        switch (starting.getType()) {
+        WindowFrameBound bound = starting;
+        int startIndex;
+        switch (bound.getType()) {
         case UNBOUNDED:
             startIndex = 0;
             break;
@@ -226,26 +282,53 @@ public final class WindowFrame {
         case VALUE:
             switch (units) {
             case ROWS: {
-                int value = starting.getValue();
+                int value = getIntOffset(bound, session);
                 startIndex = value > currentRow ? 0 : currentRow - value;
                 break;
             }
+            case GROUPS: {
+                int value = getIntOffset(bound, session);
+                startIndex = toGroupStart(orderedRows, sortOrder, currentRow, 0);
+                while (value > 0 && startIndex > 0) {
+                    value--;
+                    startIndex = toGroupStart(orderedRows, sortOrder, startIndex - 1, 0);
+                }
+                break;
+            }
+            case RANGE: {
+                startIndex = currentRow;
+                int index = sortOrder.getQueryColumnIndexes()[0];
+                if ((sortOrder.getSortTypes()[0] & SortOrder.DESCENDING) != 0) {
+                    Value c = plus(session, orderedRows, currentRow, bound, index);
+                    while (startIndex > 0
+                            && session.getDatabase().compare(c, orderedRows.get(startIndex - 1)[index]) >= 0) {
+                        startIndex--;
+                    }
+                } else {
+                    Value c = minus(session, orderedRows, currentRow, bound, index);
+                    while (startIndex > 0
+                            && session.getDatabase().compare(c, orderedRows.get(startIndex - 1)[index]) <= 0) {
+                        startIndex--;
+                    }
+                }
+                break;
+            }
             default:
-                // TODO
                 throw DbException.getUnsupportedException("units=" + units);
             }
             break;
         default:
-            throw DbException.getUnsupportedException("window frame bound type=" + starting.getType());
+            throw DbException.getUnsupportedException("window frame bound type=" + bound.getType());
         }
-
+        bound = following;
         int endIndex;
-        if (following == null) {
+        if (bound == null) {
             endIndex = currentRow;
         } else {
-            switch (following.getType()) {
+            int last = size - 1;
+            switch (bound.getType()) {
             case UNBOUNDED:
-                endIndex = size - 1;
+                endIndex = last;
                 break;
             case CURRENT_ROW:
                 endIndex = currentRow;
@@ -253,18 +336,44 @@ public final class WindowFrame {
             case VALUE:
                 switch (units) {
                 case ROWS: {
-                    int value = following.getValue();
-                    int rem = size - currentRow - 1;
-                    endIndex = value > rem ? size - 1 : currentRow + value;
+                    int value = getIntOffset(bound, session);
+                    int rem = last - currentRow;
+                    endIndex = value > rem ? last : currentRow + value;
+                    break;
+                }
+                case GROUPS: {
+                    int value = getIntOffset(bound, session);
+                    endIndex = toGroupEnd(orderedRows, sortOrder, currentRow, last);
+                    while (value > 0 && endIndex < last) {
+                        value--;
+                        endIndex = toGroupEnd(orderedRows, sortOrder, endIndex + 1, last);
+                    }
+                    break;
+                }
+                case RANGE: {
+                    endIndex = currentRow;
+                    int index = sortOrder.getQueryColumnIndexes()[0];
+                    if ((sortOrder.getSortTypes()[0] & SortOrder.DESCENDING) != 0) {
+                        Value c = minus(session, orderedRows, currentRow, bound, index);
+                        while (endIndex < last
+                                && session.getDatabase().compare(c, orderedRows.get(endIndex + 1)[index]) <= 0) {
+                            endIndex++;
+                        }
+                    } else {
+                        Value c = plus(session, orderedRows, currentRow, bound, index);
+                        while (endIndex < last
+                                && session.getDatabase().compare(c, orderedRows.get(endIndex + 1)[index]) >= 0) {
+                            endIndex++;
+                        }
+                    }
                     break;
                 }
                 default:
-                    // TODO
                     throw DbException.getUnsupportedException("units=" + units);
                 }
                 break;
             default:
-                throw DbException.getUnsupportedException("window frame bound type=" + following.getType());
+                throw DbException.getUnsupportedException("window frame bound type=" + bound.getType());
             }
         }
         if (exclusion != WindowFrameExclusion.EXCLUDE_NO_OTHERS) {
@@ -285,15 +394,8 @@ public final class WindowFrame {
             break;
         case EXCLUDE_GROUP:
         case EXCLUDE_TIES: {
-            int exStart = currentRow;
-            Value[] row = orderedRows.get(currentRow);
-            while (exStart > startIndex && sortOrder.compare(row, orderedRows.get(exStart - 1)) == 0) {
-                exStart--;
-            }
-            int exEnd = currentRow;
-            while (exEnd < endIndex && sortOrder.compare(row, orderedRows.get(exEnd + 1)) == 0) {
-                exEnd++;
-            }
+            int exStart = toGroupStart(orderedRows, sortOrder, currentRow, startIndex);
+            int exEnd = toGroupEnd(orderedRows, sortOrder, currentRow, endIndex);
             set.clear(exStart, exEnd + 1);
             if (exclusion == WindowFrameExclusion.EXCLUDE_TIES) {
                 set.set(currentRow);

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

@@ -37,7 +37,7 @@ public class WindowFrameBound {
 
     private final WindowFrameBoundType type;
 
-    private final int value;
+    private final Expression value;
 
     /**
      * Creates new instance of window frame bound.
@@ -47,15 +47,12 @@ public class WindowFrameBound {
      * @param value
      *            bound value, if any
      */
-    public WindowFrameBound(WindowFrameBoundType type, int value) {
+    public WindowFrameBound(WindowFrameBoundType type, Expression value) {
         this.type = type;
         if (type == WindowFrameBoundType.VALUE) {
-            if (value < 0) {
-                throw DbException.getInvalidValueException("unsigned", value);
-            }
             this.value = value;
         } else {
-            this.value = 0;
+            this.value = null;
         }
     }
 
@@ -73,7 +70,7 @@ public class WindowFrameBound {
      *
      * @return the value
      */
-    public int getValue() {
+    public Expression getValue() {
         return value;
     }
 
@@ -93,7 +90,7 @@ public class WindowFrameBound {
         case CURRENT_ROW:
             return "CURRENT ROW";
         case VALUE:
-            return value + (following ? " FOLLOWING" : " PRECEDING");
+            return value.getSQL() + (following ? " FOLLOWING" : " PRECEDING");
         default:
             throw DbException.throwInternalError("type=" + type);
         }

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

@@ -321,11 +321,11 @@ public class WindowFunction extends AbstractAggregate {
             Value v;
             switch (type) {
             case FIRST_VALUE: {
-                v = getNthValue(frame.iterator(ordered, getOverOrderBySort(), i, false), 0, ignoreNulls);
+                v = getNthValue(frame.iterator(session, ordered, getOverOrderBySort(), i, false), 0, ignoreNulls);
                 break;
             }
             case LAST_VALUE:
-                v = getNthValue(frame.iterator(ordered, getOverOrderBySort(), i, true), 0, ignoreNulls);
+                v = getNthValue(frame.iterator(session, ordered, getOverOrderBySort(), i, true), 0, ignoreNulls);
                 break;
             case NTH_VALUE: {
                 int n = row[1].getInt();
@@ -333,7 +333,7 @@ public class WindowFunction extends AbstractAggregate {
                     throw DbException.getInvalidValueException("nth row", n);
                 }
                 n--;
-                Iterator<Value[]> iter = frame.iterator(ordered, getOverOrderBySort(), i, fromLast);
+                Iterator<Value[]> iter = frame.iterator(session, ordered, getOverOrderBySort(), i, fromLast);
                 v = getNthValue(iter, n, ignoreNulls);
                 break;
             }

h2/src/test/org/h2/test/scripts/functions/aggregate/array-agg.sql

@@ -275,3 +275,44 @@ SELECT
 
 DROP TABLE TEST;
 > ok
+
+CREATE TABLE TEST(ID INT, VALUE INT);
+> ok
+
+INSERT INTO TEST VALUES
+    (1, 1),
+    (2, 1),
+    (3, 5),
+    (4, 8),
+    (5, 8),
+    (6, 8),
+    (7, 9),
+    (8, 9);
+> update count: 8
+
+SELECT *,
+    ARRAY_AGG(ID) OVER (ORDER BY VALUE ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) R_ID,
+    ARRAY_AGG(VALUE) OVER (ORDER BY VALUE ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) R_V,
+    ARRAY_AGG(ID) OVER (ORDER BY VALUE RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) V_ID,
+    ARRAY_AGG(VALUE) OVER (ORDER BY VALUE RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) V_V,
+    ARRAY_AGG(VALUE) OVER (ORDER BY VALUE DESC RANGE BETWEEN 1 PRECEDING AND 1 FOLLOWING) V_V_R,
+    ARRAY_AGG(ID) OVER (ORDER BY VALUE GROUPS BETWEEN 1 PRECEDING AND 1 FOLLOWING) G_ID,
+    ARRAY_AGG(VALUE) OVER (ORDER BY VALUE GROUPS BETWEEN 1 PRECEDING AND 1 FOLLOWING) G_V
+    FROM TEST;
+> ID VALUE R_ID      R_V       V_ID            V_V             V_V_R           G_ID               G_V
+> -- ----- --------- --------- --------------- --------------- --------------- ------------------ ------------------
+> 1  1     (1, 2)    (1, 1)    (1, 2)          (1, 1)          (1, 1)          (1, 2, 3)          (1, 1, 5)
+> 2  1     (1, 2, 3) (1, 1, 5) (1, 2)          (1, 1)          (1, 1)          (1, 2, 3)          (1, 1, 5)
+> 3  5     (2, 3, 4) (1, 5, 8) (3)             (5)             (5)             (1, 2, 3, 4, 5, 6) (1, 1, 5, 8, 8, 8)
+> 4  8     (3, 4, 5) (5, 8, 8) (4, 5, 6, 7, 8) (8, 8, 8, 9, 9) (9, 9, 8, 8, 8) (3, 4, 5, 6, 7, 8) (5, 8, 8, 8, 9, 9)
+> 5  8     (4, 5, 6) (8, 8, 8) (4, 5, 6, 7, 8) (8, 8, 8, 9, 9) (9, 9, 8, 8, 8) (3, 4, 5, 6, 7, 8) (5, 8, 8, 8, 9, 9)
+> 6  8     (5, 6, 7) (8, 8, 9) (4, 5, 6, 7, 8) (8, 8, 8, 9, 9) (9, 9, 8, 8, 8) (3, 4, 5, 6, 7, 8) (5, 8, 8, 8, 9, 9)
+> 7  9     (6, 7, 8) (8, 9, 9) (4, 5, 6, 7, 8) (8, 8, 8, 9, 9) (9, 9, 8, 8, 8) (4, 5, 6, 7, 8)    (8, 8, 8, 9, 9)
+> 8  9     (7, 8)    (9, 9)    (4, 5, 6, 7, 8) (8, 8, 8, 9, 9) (9, 9, 8, 8, 8) (4, 5, 6, 7, 8)    (8, 8, 8, 9, 9)
+> rows (ordered): 8
+
+SELECT *, ARRAY_AGG(ID) OVER (ORDER BY VALUE ROWS -1 PRECEDING) FROM TEST;
+> exception INVALID_VALUE_2
+
+DROP TABLE TEST;
+> ok