聊聊flink Table的Group Windows

Table table = input
    .window([Window w].as("w"))  // 定义窗口并为其赋予别名 w
    .groupBy("w")  // 按窗口 w 分组表
    .select("b.sum");  // 聚合

Table table = input
    .window([Window w].as("w"))  // 定义窗口并为其赋予别名 w
    .groupBy("w, a")  // 按属性 a 和窗口 w 分组表
    .select("a, b.sum");  // 聚合

Table table = input
    .window([Window w].as("w"))  // 定义窗口并为其赋予别名 w
    .groupBy("w, a")  // 按属性 a 和窗口 w 分组表
    .select("a, w.start, w.end, w.rowtime, b.count"); // 聚合并添加窗口的开始、结束和行时间戳

// 事件时间的Tumbling窗口
.window(Tumble.over("10.minutes").on("rowtime").as("w"));

// 处理时间的Tumbling窗口（假设有一个处理时间属性 "proctime"）
.window(Tumble.over("10.minutes").on("proctime").as("w"));

// 基于行数的Tumbling窗口（假设有一个处理时间属性 "proctime"）
.window(Tumble.over("10.rows").on("proctime").as("w"));

// 事件时间的Sliding窗口
.window(Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w"));

// 处理时间的Sliding窗口（假设有一个处理时间属性 "proctime"）
.window(Slide.over("10.minutes").every("5.minutes").on("proctime").as("w"));

// 基于行数的Sliding窗口（假设有一个处理时间属性 "proctime"）
.window(Slide.over("10.rows").every("5.rows").on("proctime").as("w"));

// 事件时间的Session窗口
.window(Session.withGap("10.minutes").on("rowtime").as("w"));

// 处理时间的Session窗口（假设有一个处理时间属性 "proctime"）
.window(Session.withGap("10.minutes").on("proctime").as("w"));

class Table(
    private[flink] val tableEnv: TableEnvironment,
    private[flink] val logicalPlan: LogicalNode) {

  //......

  def window(window: Window): WindowedTable = {
    new WindowedTable(this, window)
  }

  //......
}

class WindowedTable(
    private[flink] val table: Table,
    private[flink] val window: Window) {

  def groupBy(fields: Expression*): WindowGroupedTable = {
    val fieldsWithoutWindow = fields.filterNot(window.alias.equals(_))
    if (fields.size != fieldsWithoutWindow.size + 1) {
      throw new ValidationException("GroupBy must contain exactly one window alias.")
    }

    new WindowGroupedTable(table, fieldsWithoutWindow, window)
  }

  def groupBy(fields: String): WindowGroupedTable = {
    val fieldsExpr = ExpressionParser.parseExpressionList(fields)
    groupBy(fieldsExpr: _*)
  }

}

class WindowGroupedTable(
    private[flink] val table: Table,
    private[flink] val groupKeys: Seq[Expression],
    private[flink] val window: Window) {

  def select(fields: Expression*): Table = {
    val expandedFields = expandProjectList(fields, table.logicalPlan, table.tableEnv)
    val (aggNames, propNames) = extractAggregationsAndProperties(expandedFields, table.tableEnv)

    val projectsOnAgg = replaceAggregationsAndProperties(
      expandedFields, table.tableEnv, aggNames, propNames)

    val projectFields = extractFieldReferences(expandedFields ++ groupKeys :+ window.timeField)

    new Table(table.tableEnv,
      Project(
        projectsOnAgg,
        WindowAggregate(
          groupKeys,
          window.toLogicalWindow,
          propNames.map(a => Alias(a._1, a._2)).toSeq,
          aggNames.map(a => Alias(a._1, a._2)).toSeq,
          Project(projectFields, table.logicalPlan).validate(table.tableEnv)
        ).validate(table.tableEnv),
        // required for proper resolution of the time attribute in multi-windows
        explicitAlias = true
      ).validate(table.tableEnv))
  }

  def select(fields: String): Table = {
    val fieldExprs = ExpressionParser.parseExpressionList(fields)
    //get the correct expression for AggFunctionCall
    val withResolvedAggFunctionCall = fieldExprs.map(replaceAggFunctionCall(_, table.tableEnv))
    select(withResolvedAggFunctionCall: _*)
  }
}