订阅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版本开始支持）

  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。若该表同时涉及OLAP查询，则建议采用行列共存的存储方式，详情请参见表的存储格式。

• 仅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';

返回结果示例如下。