实时数仓Hologres：订阅Hologres Binlog

在Hologres中订阅Hologres Binlog需要注意如下事项：

• 仅Hologres V0.9及以上版本支持订阅Hologres Binlog，如果您的实例是V0.9以下版本，请您加入在线支持钉钉群，详情请参见如何获取更多的在线支持？。

• Hologres V0.9以及V0.10版本，已存在的表无法修改表属性开启Binlog，需重新建表。从V1.1版本开始，可以按需开启Binlog。

• Hologres V1.3.14和V1.1.82之前的版本仅支持Superuser消费Binlog，如果使用其他低权限账号消费Binlog会出现没有权限的报错：permission denied for table hg_replication_slot_properties。从Hologres V1.3.14和V1.1.82版本开始，若使用Flink消费Binlog，仅需账号具备查询表的权限即可。对于使用JDBC消费Hologres Binlog需要账号在Replication Role中。

• Hologres支持单表级别的Binlog功能，支持行存表和列存表。当前订阅Hologres Binlog支持的情况如下表。

  Flink分类	Hologres行存表Binlog	Hologres列存表Binlog	Hologres行列共存表Binlog（从Hologres V1.1版本开始支持）
  实时计算Blink	支持	支持	支持
  全托管Flink	支持	支持	支持
  开源Flink	不支持	不支持	不支持
  JDBC	Hologres V1.1版本开始支持	Hologres V1.1版本开始支持	Hologres V1.1版本开始支持

• Blink消费Hologres Binlog暂不支持Hologres的TIMESTAMP类型，在Hologres建表时，请使用TIMESTAMPTZ类型。同时也不支持SMALLINT等特殊类型。

• 不支持消费分区表父表的Binlog，请使用分区子表或者普通表（非分区表）。Hologres从 V1.3.24版本开始，支持按需修改分区子表的Binlog TTL，若是没有显式指定分区子表的Binlog TTL，则与父表的Binlog TTL保持一致。同时需要注意Binlog TTL不是精准的时间，系统不会强保证Binlog到期后立马删除Binlog，将会在过期后的某个时间删除。

• 对于更新频繁的场景，理论上列存表开启Binlog的开销要大于行存表的开销，所以建议使用行存表开启Binlog。

• 仅Hologres内部表支持开启Binlog，外部表不支持开启Binlog。

Binlog记录由Binlog系统字段和用户Table字段组成，具体字段定义如下表。

Hologres中，Binlog功能默认关闭，您可以通过设置表属性binlog.level和binlog.ttl开启该功能。如下示例开启Binlog，更多关于创建表的参数说明，请参见CREATE TABLE。

• V2.1版本起支持的语法：

  binlog.level和binlog.ttl表属性名更新为binlog_level和binlog_ttl。

  CREATE TABLE test_message_src (
      id int PRIMARY KEY,
      title text NOT NULL,
      body text
  )
  WITH (
      orientation = 'row',
      clustering_key = 'id',
      binlog_level = 'replica',
      binlog_ttl = '86400' -- Binlog的TTL，单位为秒
  );

• 所有版本支持的语法：

  begin;
  create table test_message_src(
    id int primary key, 
    title text not null, 
    body text);
  call set_table_property('test_message_src', 'orientation', 'row');--创建行存表test_message_src
  call set_table_property('test_message_src', 'clustering_key', 'id');--在id列建立聚簇索引
  call set_table_property('test_message_src', 'binlog.level', 'replica');--设置表属性开启Binlog功能
  call set_table_property('test_message_src', 'binlog.ttl', '86400');--binlog.ttl，Binlog的TTL，单位为秒
  commit;

参数说明如下。

从Hologres V1.1版本开始，可以根据业务需要选择开启/关闭Binlog能力，同时支持配置TTL满足不同业务场景对Binlog保留时间的诉求，已有表无需重新建表就能开启Binlog，操作方便快捷。

Hologres表的Binlog数据采用强Schema格式。要查询特定表的Binlog，可通过将Binlog内置额外字段与表原有字段进行组合来实现。此外，Hologres还提供了一些函数，以便查询最新或最早的Binlog，或者通过已知的LSN和TIMESTAMP来查询相应的Binlog信息。

通过内置特殊字段直接查询Binlog

使用Binlog内置额外字段与表原有字段组合的方式查询Binlog，示例如下。

SELECT hg_binlog_lsn,hg_binlog_event_type,hg_binlog_timestamp_us,* FROM test_message_src;

查询结果示例如下。

查询表指定Shard上最早或最新的Binlog

使用hg_get_binlog_cursor函数的方式查询Binlog，语法如下。

-- OLDEST，此shard上最早的Binlog
SELECT * FROM hg_get_binlog_cursor('<table_name>','OLDEST',<shard_id>);

-- LATEST，此shard上最新的Binlog
SELECT * FROM hg_get_binlog_cursor('<table_name>','LATEST',<shard_id>);

示例如下。

SELECT * FROM hg_get_binlog_cursor('test_message_src','OLDEST',0);

查询结果示例如下。

通过Binlog LSN查询相应Binlog的时间戳

使用hg_get_binlog_cursor_by_lsn函数查询相应Binlog的时间戳，可返回大于等于所查询LSN（日志序列号）的第一条Binlog的信息，如果当前LSN不存在，返回结果中的hg_binlog_timestamp_us字段是当前时间。语法如下。

SELECT * FROM hg_get_binlog_cursor_by_lsn('<table_name>',<lsn>,<shard_id>);--LSN值（bigint类型）

示例如下。

SELECT * FROM hg_get_binlog_cursor_by_lsn('test_message_src',152,0);

查询结果示例如下。

通过Binlog TIMESTAMP查询相应Binlog的LSN

使用hg_get_binlog_cursor_by_timestamp函数查询相应Binlog的LSN，可返回大于等于所查询时间的第一条Binlog的信息，如果已经是最新的，返回结果中的hg_binlog_timestamp_us字段是当前时间，hg_binlog_lsn字段是下一条数据插入时将会分配的LSN。语法如下。

SELECT * FROM hg_get_binlog_cursor_by_timestamp('<table_name>',<timestamp>,<shard_id>);

示例如下。

SELECT *,to_timestamp(hg_binlog_timestamp_us/1000000.0) FROM hg_get_binlog_cursor_by_timestamp('test_message_src','2024-05-20 19:34:53.791+08',0);

查询结果示例如下。

当前支持通过Flink、Blink以及JDBC（包括Holo Client）消费Hologres Binlog，详情请参见文档：

• Flink和Blink实时消费Binlog，请参见Flink/Blink实时消费Hologres Binlog。

• 有关JDBC如何实时消费Binlog，请参见通过JDBC消费Hologres Binlog。

您可以使用如下SQL查看哪些表开启了Binlog。

SELECT
    *
FROM
    hologres.hg_table_properties
WHERE
    property_key = 'binlog.level'
    AND property_value = 'replica';

返回结果示例如下。