窗口聚合

SESSION窗口表值函数聚合在不同VVR版本中的区别

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

SESSION(TABLE data, DESCRIPTOR(timecol), gap)

实时计算引擎VVR 8.x与VVR 11.x的SESSION窗口表值函数使用区别如下：

以下示例中SQL语法是等价的，都将使用item字段作为SESSION窗口函数的分区字段：

-- tables must have time attribute, e.g. `bidtime` in this table
> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- VVR 11.x
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;
  
-- VVR 8.x
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

示例对比说明：

窗口表值函数和聚合语句无法合并的场景

以下场景均以SESSION窗口为例，同样适用于其他窗口表值函数。

• 在窗口表值函数和聚合语句之间，包含对window_start、window_end和window_time字段的过滤或计算。例如：

  -- 包含对window_start的过滤
  > SELECT window_start, window_end, item, SUM(price) AS total_price
      FROM
      (SELECT item, price, window_start, window_end FROM
      TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
      where window_start >= TIMESTAMP '2020-04-15 08:06:00.000')
      GROUP BY item, window_start, window_end;
    
  -- 包含对window_start的计算
  > SELECT window_start, window_end, item, SUM(price) AS total_price
      FROM
      (SELECT item, price, window_start + (INTERVAL '1' SECOND) as window_start, window_end FROM
      TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
      GROUP BY item, window_start, window_end;
  
  -- 包含对window_start的计算
  > SELECT window_start, window_end, item, SUM(price) AS total_price
      FROM
      (SELECT item, price, CAST(window_start as varchar) as window_start, window_end FROM
      TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
      GROUP BY item, window_start, window_end;
  

• 窗口表值函数和UDTF同时使用。例如：

  > SELECT window_start, window_end, category, SUM(price) AS total_price
      FROM
      (SELECT category, price, window_start, window_end FROM
      TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)),
      LATERAL TABLE(category_udtf(item)) as T(category))
      GROUP BY category, window_start, window_end;

• 聚合语句的GROUP KEY中未同时包含window_start和window_end。例如：

   SELECT window_start, item, SUM(price) AS total_price
    FROM TABLE(
        SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    GROUP BY item, window_start;

• 聚合函数使用python UDAF。

• 聚合函数使用GROUPING SETS、CUBE和ROLLUP语法，导致window_start和window_end不在同一组GROUP KEY中。例如：

  > SELECT item, SUM(price) AS total_price
    FROM TABLE(
        SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    GROUP BY GROUPING SETS((item), (window_start), (window_end));
    
  > SELECT item, SUM(price) AS total_price
    FROM TABLE(
        SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    GROUP BY CUBE (item, window_start, window_end);
    
  > SELECT item, SUM(price) AS total_price
    FROM TABLE(
        SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    GROUP BY ROLLUP (item, window_start, window_end);

• 聚合语句中，聚合函数使用窗口列window_start、window_end和window_time进行计算。例如：

  > SELECT window_start, window_end, item, SUM(price) AS total_price, max(window_end) AS max_end
    FROM TABLE(
        SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
    GROUP BY item, window_start, window_end;
    

条件 1：聚合阶段策略允许两阶段

table.optimizer.agg-phase-strategy 设置为 AUTO（默认）或 TWO_PHASE。若设为 ONE_PHASE，则强制禁用两阶段优化。

条件 2：窗口基于事件时间

窗口必须基于事件时间（Event Time / Rowtime），不支持处理时间窗口。

条件 3：窗口类型非 Session

支持 TUMBLE、HOP、CUMULATE 窗口类型。Session 窗口不支持两阶段优化。

条件 4：聚合函数支持 Partial Merge

所有聚合函数必须支持合并操作。内置的 SUM、COUNT、MIN、MAX、AVG 均支持。自定义 UDAF 需实现 merge() 方法。

条件 5：输入流为 Insert-Only 且窗口可转换为 TVF 形式

需同时满足以下条件：

• 输入流为 Insert-Only。

• table.exec.emit.early-fire.enabled 为 false（默认）。

• table.exec.emit.late-fire.enabled 为 false（默认）。

• HOP 窗口必须对齐（窗口大小可被滑动步长整除）。

条件 6：数据分布未满足分区要求

输入数据的分布尚未满足聚合所需的分区要求。若数据已按分区键分布，优化器判断无需额外预聚合，不会生成 Local Aggregate 节点。

参数速查