Trino Delta Lake连接器如何实现类似Spark的变更数据读取控制？

问题说明 #

Trino 408及以上版本支持通过change_data_feed_enabled表属性关联开启了CDC的Delta表，但即使在413版本中设置该属性为True，默认查询仍只返回记录的最新镜像，和设置False无区别。而Spark SQL可以通过readChangeFeed选项切换读取最新镜像或完整变更日志，需要在Trino中实现类似的控制逻辑。

关联Delta表的示例SQL：

CREATE TABLE delta.table_collection.table_name (
    id varchar,
    value_1 varchar,
    value_2 integer,
    log_status varchar,
    ts bigint
) WITH (
    location = 's3://path/to/table',
    checkpoint_interval = 7,
    change_data_feed_enabled = true
);

实现方法 #

目前Trino Delta连接器暂不支持通过会话参数或查询选项直接切换CDC读取模式，要获取完整变更日志，需直接读取Delta表的CDC日志存储目录：

1. 创建CDC日志外部表 #

Delta表开启CDC后，变更日志默认存储在表路径下的_change_data目录，格式为Parquet。在Trino中创建指向该目录的外部表，即可读取所有变更记录：

CREATE TABLE delta.table_collection.table_name_cdc (
    id varchar,
    value_1 varchar,
    value_2 integer,
    log_status varchar,
    ts bigint,
    _change_type varchar, -- 标记变更类型：insert/update/delete
    _commit_version bigint, -- 变更对应的Delta提交版本
    _commit_timestamp timestamp -- 变更提交时间
) WITH (
    location = 's3://path/to/table/_change_data',
    format = 'parquet'
);

查询这个表就能获取所有历史变更操作记录，替代Spark的readChangeFeed全量读取场景。

2. 筛选特定版本区间的变更 #

如果需要获取指定版本范围内的变更，可通过_commit_version字段过滤，模拟Spark中指定起始/结束版本的readChangeFeed用法：

SELECT *
FROM delta.table_collection.table_name_cdc
WHERE _commit_version BETWEEN 3 AND 8; -- 读取版本3到8之间的所有变更

关键注意点 #

• 必须确保原Delta表在Spark侧已开启CDC（设置delta.enableChangeDataFeed = true），否则_change_data目录不会生成日志文件。
• Trino的change_data_feed_enabled属性仅用于标识表支持CDC，不会改变默认查询返回最新镜像的行为。
• 若需要更贴近Spark的readChangeFeed体验，需等待Trino Delta连接器后续版本新增对应的语法或参数支持。

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