[SPARK-58980][SQL] Support GROUPS window frames

common/utils/src/main/resources/error/error-conditions.json

@@ -1953,6 +1953,21 @@
           "Filter expression <filter> of type <type> is not a boolean."
         ]
       },
+      "GROUPS_FRAME_NEGATIVE_OFFSET" : {
+        "message" : [
+          "The offset of a groups frame must be non-negative."
+        ]
+      },
+      "GROUPS_FRAME_NULL_OFFSET" : {
+        "message" : [
+          "The <location> bound of a groups frame must not be null."
+        ]
+      },
+      "GROUPS_FRAME_WITHOUT_ORDER" : {
+        "message" : [
+          "A groups window frame cannot be used in an unordered window specification."
+        ]
+      },
       "HASH_MAP_TYPE" : {
         "message" : [
           "Input to the function <functionName> cannot contain elements of the \"MAP\" type. In Spark, same maps may have different hashcode, thus hash expressions are prohibited on \"MAP\" elements. To restore previous behavior set \"spark.sql.legacy.allowHashOnMapType\" to \"true\"."

docs/sql-ref-ansi-compliance.md

@@ -584,6 +584,7 @@ Below is a list of all the keywords in Spark SQL.
 |GRANT|reserved|non-reserved|reserved|
 |GROUP|reserved|non-reserved|reserved|
 |GROUPING|non-reserved|non-reserved|reserved|
+|GROUPS|non-reserved|non-reserved|reserved|
 |HANDLER|non-reserved|non-reserved|non-reserved|
 |HAVING|reserved|non-reserved|reserved|
 |HISTORY|non-reserved|non-reserved|non-reserved|

docs/sql-ref-syntax-qry-select-window.md

@@ -66,7 +66,7 @@ window_function [ nulls_option ] OVER
 
     **Syntax:**
 
-    `{ RANGE | ROWS } { frame_start | BETWEEN frame_start AND frame_end }`
+    `{ RANGE | ROWS | GROUPS } { frame_start | BETWEEN frame_start AND frame_end }`
 
     * `frame_start` and `frame_end` have the following syntax:
 
@@ -78,6 +78,13 @@ window_function [ nulls_option ] OVER
 
     **Note:** If `frame_end` is omitted it defaults to `CURRENT ROW`.
 
+    **Note:** `GROUPS` offsets count peer groups: rows with equal values for all window
+    `ORDER BY` expressions within a partition. `GROUPS` requires `ORDER BY` and supports multiple
+    ordering expressions. `CURRENT ROW` starts at the first row of the current peer group when
+    used as a frame start, and ends at its last row when used as a frame end.
+    `0 PRECEDING` and `0 FOLLOWING` are equivalent to `CURRENT ROW`. Offsets must be constant,
+    non-null, non-negative integer expressions.
+
 ### Examples
 
 ```sql
@@ -217,6 +224,25 @@ SELECT id, v,
 | 7|   v|   v|   v|        y|          x|         v|
 | 8|NULL|NULL|NULL|        y|          x|         v|
 +--+----+----+----+---------+-----------+----------+
+
+CREATE TABLE batches (batch_id INT, amount INT);
+
+INSERT INTO batches VALUES (1, 10), (1, 15), (2, 20), (3, 25), (3, 30), (9, 40);
+
+SELECT batch_id, amount,
+    SUM(amount) OVER (ORDER BY batch_id GROUPS BETWEEN 1 PRECEDING AND CURRENT ROW) AS moving_sum
+    FROM batches
+    ORDER BY batch_id, amount;
++--------+------+----------+
+|batch_id|amount|moving_sum|
++--------+------+----------+
+|       1|    10|        25|
+|       1|    15|        25|
+|       2|    20|        45|
+|       3|    25|        75|
+|       3|    30|        75|
+|       9|    40|        95|
++--------+------+----------+
 ```
 
 ### Related Statements

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4

@@ -300,6 +300,7 @@ GLOBAL: 'GLOBAL';
 GRANT: 'GRANT';
 GROUP: 'GROUP';
 GROUPING: 'GROUPING';
+GROUPS: 'GROUPS';
 HANDLER: 'HANDLER';
 HAVING: 'HAVING';
 BINARY_HEX: 'X';

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -1866,8 +1866,10 @@ windowSpec
 windowFrame
     : frameType=RANGE start=frameBound
     | frameType=ROWS start=frameBound
+    | frameType=GROUPS start=frameBound
     | frameType=RANGE BETWEEN start=frameBound AND end=frameBound
     | frameType=ROWS BETWEEN start=frameBound AND end=frameBound
+    | frameType=GROUPS BETWEEN start=frameBound AND end=frameBound
     ;
 
 frameBound
@@ -2257,6 +2259,7 @@ ansiNonReserved
     | GEOMETRY
     | GLOBAL
     | GROUPING
+    | GROUPS
     | HANDLER
     | HISTORY
     | HOUR
@@ -2701,6 +2704,7 @@ nonReserved
     | GRANT
     | GROUP
     | GROUPING
+    | GROUPS
     | HANDLER
     | HAVING
     | HISTORY

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala

@@ -91,6 +91,8 @@ case class WindowSpecDefinition(
             "valueBoundaryType" -> toSQLType(f.valueBoundary.head.dataType)
           )
         )
+      case f: SpecifiedWindowFrame if f.frameType == GroupFrame && orderSpec.isEmpty =>
+        DataTypeMismatch(errorSubClass = "GROUPS_FRAME_WITHOUT_ORDER")
       case _ => TypeCheckSuccess
     }
   }
@@ -159,6 +161,39 @@ case object RangeFrame extends FrameType {
   override def sql: String = "RANGE"
 }
 
+/**
+ * A frame whose integral offsets count peer groups in the window ordering.
+ */
+case object GroupFrame extends FrameType {
+  override def inputType: AbstractDataType = IntegerType
+  override def sql: String = "GROUPS"
+}
+
+/**
+ * Validates a SQL GROUPS offset before PRECEDING negates it. Keep the original magnitude through
+ * analysis so parameter binding cannot turn a negative offset into a valid signed boundary.
+ */
+case class GroupFrameOffset(child: Expression)
+  extends RuntimeReplaceable with UnaryLike[Expression] {
+  override def replacement: Expression = child
+  override def foldable: Boolean = child.foldable
+  override def sql: String = child.sql
+  // Frame descriptions should display the offset, not its analysis-only validation wrapper.
+  override def toString: String = child.toString
+
+  override def checkInputDataTypes(): TypeCheckResult = {
+    if (child.foldable && child.dataType == IntegerType &&
+        Option(child.eval()).exists(_.asInstanceOf[Int] < 0)) {
+      DataTypeMismatch(errorSubClass = "GROUPS_FRAME_NEGATIVE_OFFSET")
+    } else {
+      TypeCheckSuccess
+    }
+  }
+
+  override protected def withNewChildInternal(newChild: Expression): GroupFrameOffset =
+    copy(child = newChild)
+}
+
 /**
  * The trait used to represent special boundaries used in a window frame.
  */
@@ -272,6 +307,8 @@ case class SpecifiedWindowFrame(
   private def boundarySql(expr: Expression): String = expr match {
     case e: SpecialFrameBoundary => e.sql
     case UnaryMinus(n, _) => n.sql + " PRECEDING"
+    case IntegerLiteral(offset) if frameType == GroupFrame && offset < 0 =>
+      s"${-offset} PRECEDING"
     case e: Expression => e.sql + " FOLLOWING"
   }
 
@@ -302,6 +339,11 @@ case class SpecifiedWindowFrame(
           "expectedType" -> toSQLType(frameType.inputType)
         )
       )
+    case e: Expression if frameType == GroupFrame && e.eval() == null =>
+      DataTypeMismatch(
+        errorSubClass = "GROUPS_FRAME_NULL_OFFSET",
+        messageParameters = Map("location" -> location)
+      )
     case _ => TypeCheckSuccess
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -4384,11 +4384,12 @@ class AstBuilder extends DataTypeAstBuilder
     val partition = ctx.partition.asScala.map(expression)
     val order = ctx.sortItem.asScala.map(visitSortItem)
 
-    // RANGE/ROWS BETWEEN ...
+    // RANGE/ROWS/GROUPS BETWEEN ...
     val frameSpecOption = Option(ctx.windowFrame).map { frame =>
       val frameType = frame.frameType.getType match {
         case SqlBaseParser.RANGE => RangeFrame
         case SqlBaseParser.ROWS => RowFrame
+        case SqlBaseParser.GROUPS => GroupFrame
       }
 
       SpecifiedWindowFrame(
@@ -4412,7 +4413,12 @@ class AstBuilder extends DataTypeAstBuilder
       if (!(e.resolved && e.foldable || e.isInstanceOf[Parameter])) {
         throw QueryParsingErrors.invalidWindowFrameBoundError(ctx)
       }
-      e
+      if (ctx.getParent.asInstanceOf[WindowFrameContext].frameType.getType ==
+          SqlBaseParser.GROUPS) {
+        GroupFrameOffset(e)
+      } else {
+        e
+      }
     }
 
     ctx.boundType.getType match {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala

@@ -583,6 +583,79 @@ class ExpressionParserSuite extends AnalysisTest {
         stop = 59))
   }
 
+  private def groupsBoundary(boundary: Expression): Expression = boundary match {
+    case s: SpecialFrameBoundary => s
+    case UnaryMinus(offset, _) => -GroupFrameOffset(offset)
+    case IntegerLiteral(offset) if offset < 0 => -GroupFrameOffset(Literal(-offset))
+    case offset => GroupFrameOffset(offset)
+  }
+
+  test("groups window function expressions") {
+    val func = $"foo".function(star())
+    def windowed(
+        partitioning: Seq[Expression] = Seq.empty,
+        ordering: Seq[SortOrder] = Seq.empty,
+        frame: WindowFrame = UnspecifiedFrame): Expression = {
+      WindowExpression(func, WindowSpecDefinition(partitioning, ordering, frame))
+    }
+
+    val boundaries = Seq(
+      // Single-bound forms.
+      ("unbounded preceding", UnboundedPreceding, CurrentRow),
+      ("10 preceding", -Literal(10), CurrentRow),
+      ("3 + 1 preceding", -Add(Literal(3), Literal(1)), CurrentRow),
+      ("0 preceding", -Literal(0), CurrentRow),
+      ("current row", CurrentRow, CurrentRow),
+      ("0 following", Literal(0), CurrentRow),
+      ("3 + 1 following", Add(Literal(3), Literal(1)), CurrentRow),
+      ("10 following", Literal(10), CurrentRow),
+      ("unbounded following", UnboundedFollowing, CurrentRow), // Will fail during analysis
+
+      // BETWEEN forms.
+      ("between unbounded preceding and 5 following", UnboundedPreceding, Literal(5)),
+      ("between unbounded preceding and current row", UnboundedPreceding, CurrentRow),
+      ("between 10 preceding and current row", -Literal(10), CurrentRow),
+      ("between 0 preceding and current row", -Literal(0), CurrentRow),
+      ("between current row and current row", CurrentRow, CurrentRow),
+      ("between current row and 0 following", CurrentRow, Literal(0)),
+      ("between current row and 5 following", CurrentRow, Literal(5)),
+      ("between current row and unbounded following", CurrentRow, UnboundedFollowing),
+      ("between 10 preceding and unbounded following", -Literal(10), UnboundedFollowing),
+      ("between 0 preceding and unbounded following", -Literal(0), UnboundedFollowing),
+      ("between 10 preceding and 5 following", -Literal(10), Literal(5)),
+      ("between unbounded preceding and unbounded following",
+        UnboundedPreceding, UnboundedFollowing)
+    )
+    boundaries.foreach {
+      case (boundarySql, begin, end) =>
+        val query = s"foo(*) over (partition by a order by b groups $boundarySql)"
+        val expr = windowed(Seq($"a"), Seq($"b".asc),
+          SpecifiedWindowFrame(GroupFrame, groupsBoundary(begin), groupsBoundary(end)))
+        assertEqual(query, expr)
+    }
+  }
+
+  test("GROUPS window frame sql output re-parses to an equal tree") {
+    val func = $"foo".function(star())
+    val frames = Seq(
+      SpecifiedWindowFrame(GroupFrame, UnboundedPreceding, CurrentRow),
+      SpecifiedWindowFrame(GroupFrame, CurrentRow, CurrentRow),
+      SpecifiedWindowFrame(GroupFrame, UnboundedPreceding, UnboundedFollowing),
+      SpecifiedWindowFrame(GroupFrame, CurrentRow, UnboundedFollowing),
+      SpecifiedWindowFrame(GroupFrame, Literal(-2), Literal(-1)),
+      SpecifiedWindowFrame(GroupFrame, Literal(1), Literal(2)),
+      SpecifiedWindowFrame(GroupFrame, Literal(0), CurrentRow)
+    )
+    frames.foreach { frame =>
+      val expr = WindowExpression(func,
+        WindowSpecDefinition(Seq($"a"), Seq($"b".asc),
+          frame.copy(lower = groupsBoundary(frame.lower), upper = groupsBoundary(frame.upper))))
+      val sqlText = s"foo(*) over (partition by a order by b ${frame.sql})"
+      val reparsed = defaultParser.parseExpression(sqlText)
+      compareExpressions(reparsed, expr)
+    }
+  }
+
   test("row constructor") {
     // Note that '(a)' will be interpreted as a nested expression.
     assertEqual("(a, b)", CreateStruct(Seq($"a", $"b")))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala

@@ -1222,6 +1222,23 @@ class PlanParserSuite extends AnalysisTest {
     parsePlan("select approx, asof, distance, exact, nearest, similarity from t")
   }
 
+  test("GROUPS keyword is non-reserved (usable as identifier)") {
+    def checkGroupsAsIdentifier(): Unit = {
+      parsePlan("select groups from t")
+      parsePlan("select a as groups from t")
+      parsePlan("select * from groups")
+      parsePlan("select foo(*) over (order by groups) from t")
+      parsePlan(
+        "select foo(*) over (order by groups groups between 1 preceding and current row) from t")
+    }
+    checkGroupsAsIdentifier()
+    withSQLConf(
+        SQLConf.ANSI_ENABLED.key -> "true",
+        SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
+      checkGroupsAsIdentifier()
+    }
+  }
+
   test("sampled relations") {
     val sql = "select * from t"
     assertEqual(s"$sql tablesample(100 rows)",