Flink SQL选择ROWTIME字段时类型转换异常问题求助

问题描述 #

通过CREATE TABLE语句创建表并配置Watermark列：

CREATE TABLE IF NOT EXISTS MY_TABLE
  (
      id                    INT NOT NULL,
      ...
      _ingested_timestamp   TIMESTAMP(3),
      WATERMARK FOR _ingested_timestamp AS _ingested_timestamp - INTERVAL '10' SECOND,
      PRIMARY KEY(id) NOT ENFORCED
  )
WITH(...);

执行以下查询时抛出Conversion to relational algebra failed to preserve datatypes异常：

SELECT
 item.id,
 ...
 item._ingested_timestamp as event_time, 
FROM MY_TABLE as item JOIN ....;

异常核心是类型不匹配：验证阶段预期event_time带有*ROWTIME*标识，但转换后的关系代数中该标识丢失：

validated type:
RecordType(INTEGER NOT NULL id, ..., TIMESTAMP(3) *ROWTIME* event_time) NOT NULL
converted type:
RecordType(INTEGER NOT NULL id, ..., TIMESTAMP(3) event_time) NOT NULL

问题原因 #

_ingested_timestamp是绑定了ROWTIME元数据的事件时间列（因配置了Watermark），但在SELECT语句中为其设置别名event_time后，Flink默认不会将ROWTIME元数据传递给别名列。当查询涉及JOIN等依赖事件时间属性的操作时，查询计划器验证阶段预期event_time保留ROWTIME标识，而实际转换后的关系代数中该列无此标识，导致类型不匹配异常。

解决方法 #

方法一：直接使用原始ROWTIME列 #

避免对ROWTIME列设置别名，直接用原始列名参与查询，保留其ROWTIME属性：

SELECT
 item.id,
 ...
 item._ingested_timestamp, 
FROM MY_TABLE as item JOIN ....;

方法二：为别名列重新声明Watermark #

若必须使用别名，需在查询中重新为别名列配置Watermark，显式赋予其ROWTIME属性：

WITH processed_item AS (
    SELECT
        id,
        ...,
        _ingested_timestamp AS event_time
    FROM MY_TABLE
)
SELECT
    id,
    ...,
    event_time
FROM processed_item
WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
JOIN ....;

方法三：JOIN条件中直接引用原始列 #

如果JOIN逻辑依赖事件时间，可在JOIN条件中直接使用原始的_ingested_timestamp列，SELECT中别名化不影响核心逻辑：

SELECT
 item.id,
 ...
 item._ingested_timestamp AS event_time, 
FROM MY_TABLE as item
JOIN other_table ot
ON item.id = ot.id
AND item._ingested_timestamp BETWEEN ot.start_time AND ot.end_time;

内容的提问来源于stack exchange，提问作者mdm