实时计算Flink版：Group Aggregate

可兼容的变更

• 新增、删除、修改非Distinct的统计指标（Aggregate Function）。

  • 对于新增统计指标，属于部分兼容变更，从当前作业启动时开始累计。

  • 对于删除统计指标，属于完全兼容变更。删除的统计指标对应的状态数据会被丢弃。

  • 对于既有新增又有删除统计指标，则属于部分兼容变更。新增的统计指标从当前作业启动时开始累计，删除的统计指标对应的状态数据会被丢弃。

  • 对于修改统计指标，被视为删除和新增两个操作，属于部分兼容变更。新增的统计指标从当前作业启动时开始累计，删除的统计指标对应的状态数据会被丢弃。

  说明

  对于未进行任何变更的统计指标，复用状态数据后计算的结果和基于历史数据运行的结果是一致的。

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;
  
  -- 新增统计指标：count(c)，属于部分兼容变更。
  -- sum(b)、max(c) 的计算结果不受影响，count(c) 的值在作业启动时从 0 开始累计
  SELECT a, SUM(b), MAX(c), COUNT(c) FROM MyTable GROUP BY a;
  
  -- 删除统计指标：sum(b)，属于完全兼容变更。
  -- max(c)的计算结果不受影响。
  SELECT a, MAX(c) FROM MyTable GROUP BY a;
  
  -- 修改统计指标：max(c) -> min(c)，属于部分兼容变更。
  -- sum(b) 的计算结果不受影响。max(c) 被认为删除，其对应的状态数据会被丢弃。
  -- min(c) 被认为是新增指标，其值在作业启动时开始计算。
  SELECT a, SUM(b), MIN(c) FROM MyTable GROUP BY a;

• 调整非Distinct的统计指标位置，该修改属于完全兼容变更。

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;
  
  -- 调整统计指标位置：sum(b)、max(c)，属于完全兼容变更。
  -- sum(b)、max(c)的计算结果不受影响。
  SELECT a, MAX(c), SUM(b) FROM MyTable GROUP BY a;

• 当非Distinct的统计指标涉及的字段有计算逻辑变化时，该统计指标被认为做了修改操作，该修改属于部分兼容变更。

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;
  
  -- 修改统计指标：max(c) -> max(substring(c, 1, 5))，属于部分兼容变更。
  -- sum(b)的计算结果不受影响。max(c)被认为删除，其对应的状态数据会被丢弃。
  -- max(substring(c, 1, 5))被认为新增指标，其值在作业启动时开始计算。
  SELECT a, SUM(b), MAX(c) FROM (
    SELECT a, b, SUBSTRING(c, 1, 5) AS c FROM MyTable
  ) GROUP BY a;

• 多个包含DISTINCT聚合函数和非DISTINCT聚合函数更换顺序，但是DISTINCT聚合函数相对顺序一致，属于完全兼容变更。

  -- 原始SQL。
  INSERT INTO MySink
  SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;
  
  -- SUM(DISTINCT b)仍然在COUNT(DISTINCT b)前，相对顺序没变，属于完全兼容变更。
  INSERT INTO MySink
  SELECT a, SUM(DISTINCT b), COUNT(DISTINCT b), MAX(b)  FROM MyTable GROUP BY a;

• 作业变更前后都没有统计指标，该修改属于完全兼容变更。

• 在Retract流后删除Retract聚合函数，属于完全兼容变更。

  -- 原始SQL。
  SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
      (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
  GROUP BY c/2;
  
  -- 删除Retract聚合函数，属于完全兼容变更。
  SELECT c/2, AVG(avg_a) AS avg_avg_a FROM
      (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
  GROUP BY c/2;

不兼容的变更

• 新增、删除、修改统计维度（group key）或者统计维度涉及字段的计算逻辑发生变化，该修改属于不兼容修改。

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;
  
  -- 新增统计维度：d，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a, d;
  
  -- 删除统计维度：a，当前修改属于不兼容变更。
  SELECT SUM(b), MAX(c) FROM MyTable;
  
  -- 修改统计维度：a -> d，当前修改属于不兼容变更。
  SELECT d, SUM(b), MIN(c) FROM MyTable GROUP BY d;
  
  -- 修改了统计维度：a -> a + 1，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c) FROM (
    SELECT a + 1 AS a, b, c FROM MyTable 
  ) GROUP BY a;

• 新增、删除、修改Distinct统计指标（Distinct Aggregate Function）或者Distinct统计指标涉及字段的计算逻辑发生变化，该修改属于不兼容变更。

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;
  
  -- 新增distinct统计指标：count(distinct b)，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;
  
  -- 删除distinct统计指标：sum(distinct b)，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c), COUNT(DISTINCT c) FROM MyTable GROUP BY a;
  
  -- 修改distinct统计指标：sum(distinct b) -> avg(distinct b)，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c), AVG(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;
  
  -- 修改distinct统计指标：count(distinct c) -> count(distinct avg(c))，当前修改属于不兼容变更。
  SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM (
       SELECT a, b, AVG(c) AS c from MyTable GROUP BY a, b
  ) GROUP BY a;

• 级联聚合时，新增统计指标，该修改属于不兼容变更。（级联聚合会发生消息撤回，新增统计指标结果不可预期。）

  -- 原始SQL。
  SELECT a/2, AVG(b), MIN(c) FROM (
      SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
  ) GROUP BY a/2;
  
  -- 新增统计指标：count(c)，当前修改属于不兼容变更。
  SELECT a/2, AVG(b), MIN(c), COUNT(c) FROM (
      SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
  ) GROUP BY a/2;

• 删除了所有统计指标，该修改属于不兼容变更。（所有统计指标的状态数据都被丢弃，没有状态复用。）

  -- 原始SQL。
  SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;
  
  -- 删除所有统计指标：sum(b), max(c)，当前修改属于不兼容变更。
  SELECT a FROM MyTable GROUP BY a;

• 多个包含DISTINCT聚合函数和非DISTINCT聚合函数更换顺序，但是DISTINCT聚合函数相对顺序变化，属于不兼容变更。

  -- 原始SQL。
  INSERT INTO MySink
  SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;
  
  --SUM(DISTINCT b)被交换到了COUNT(DISTINCT b)后，相对顺序发生变化，属于完全兼容变更。
  INSERT INTO MySink
  SELECT COUNT(DISTINCT b), a, MAX(b), SUM(DISTINCT b) FROM MyTable GROUP BY a;

• 变更前没有任何一个统计指标，而新作业加入新的统计指标，该修改属于不兼容变更。

  -- 原始SQL。
  INSERT INTO MySink
  SELECT a, b FROM MyTable GROUP BY a,b;
  
  -- 加入新的聚合函数，属于不兼容变更。
  INSERT INTO MySink
  SELECT a, b, SUM(b) FROM MyTable GROUP BY a,b;

• 变更后仅有一个统计指标，并且修改了计算逻辑，该修改属于不兼容变更。

  -- 原始SQL。
  INSERT INTO MySink
  SELECT a, SUM(b), MAX(b), MAX(c) FROM MyTable GROUP BY a;
  
  --修改统计指标和计算逻辑，属于不兼容变更。
  INSERT INTO MySink 
  SELECT a, MAX(c) FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable) GROUP BY a;

• 变更前后作业任何一个统计指标都不同，属于不兼容变更。

  -- 原始SQL。
  INSERT INTO MySink
  SELECT a, b, MAX(c) FROM MyTable GROUP BY a,b;
  
  -- 修改统计指标，属于不兼容变更。
  INSERT INTO MySink
  SELECT a, b, MIN(c) FROM MyTable GROUP BY a,b;

• 在Retract流后新增或变更Retract聚合函数，属于不兼容变更。

  -- 原始SQL。
  SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
      (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
  GROUP BY c/2;
  
  -- 新增Retract聚合函数，属于不兼容变更。
  SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) min_max_b, MAX(max_b) max_max_b FROM
      (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
  GROUP BY c/2;
  
  -- 变更Retract聚合函数，属于不兼容变更。
  SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) max_max_b FROM
      (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
  GROUP BY c/2;

未知兼容变更

存在Python自定义聚合函数（UDAF）时，任何变更都属于未知兼容变更。

-- 作业变更前后，存在例如python自定义weighter_avg的函数，属于未知兼容。
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b), weighted_avg(a, b) 
FROM MyTable GROUP BY a, b;