PARTITION BY with TUMBLE导致FlinkSQL中的异常

Question

我想使用FlinkSQL选择处理窗口的最后一个元素。我尝试在Blink planner中使用ROW_NUMBER来实现这一点。已尝试以下查询：

SELECT * FROM (
  SELECT key, value, ROW_NUMBER() OVER w AS rn
    FROM InputTable
  WINDOW w AS (PARTITION BY key, TUMBLE(rt, INTERVAL '15' MINUTE) ORDER BY -ts)
) WHERE rn = 1

// rt = ts.rowtime, ts is Long

不幸的是，这会导致异常

org.apache.flink.table.planner.codegen.CodeGenException: Unsupported call: TUMBLE(TIMESTAMP(3) *ROWTIME*, INTERVAL SECOND(3) NOT NULL) 
If you think this function should be supported, you can create an issue and start a discussion for it.

你知道我做错了什么吗？我认为TUMBLE函数与计算rowtime % interval是“等价的”。

Answer 1

TUMBLE (以及HOP和SESSION)是Flink SQL (版本2.11)中的特殊内置函数，只能在GROUP BY子句中使用。原则上您是对的，在这个上下文中使用TUMBLE应该是可以的，但在这一点上根本不支持它。

您可以实现一个用户定义的函数来重新实现TUMBLE的分组逻辑，但是，我不建议这样做，因为查询将不能很好地执行。Flink SQL不会意识到一个分区(PARTITION BY key, TUMBLE(rt, INTERVAL '15' MINUTE)只会在35分钟内处于“活动”状态，并且永远保持其状态。因此，随着时间的推移，查询会积累越来越多的状态，这会减慢检查点和恢复的速度。在未来，这种基于时间的OVER分区应该会被支持，但AFAIK还不支持。