Hologres:Hologres バイナリログのサブスクライブ

Hologres バイナリログをサブスクライブする際には、次の制限事項にご注意ください。

• Hologres V0.9 以降のみがバイナリログのサブスクリプションをサポートしています。お使いのインスタンスがそれ以前のバージョンの場合は、オンラインサポートの DingTalk グループにご参加ください。詳細については、「オンラインサポートの利用方法」をご参照ください。

• Hologres V0.9 および V0.10 では、プロパティを変更して既存のテーブルのバイナリログを有効にすることはできません。テーブルを再作成する必要があります。V1.1 からは、オンデマンドでバイナリログを有効にできます。

• Hologres V1.3.14 および V1.1.82 より前では、スーパーユーザーのみがバイナリログを消費できました。権限の少ないアカウントを使用すると、permission denied for table hg_replication_slot_properties というエラーが発生します。V1.3.14 および V1.1.82 からは、Flink でバイナリログを消費する場合、アカウントにはテーブルに対する SELECT 権限のみが必要です。JDBC でバイナリログを消費するには、アカウントがレプリケーションロールに属している必要があります。

• Hologres は、行指向テーブルと列指向テーブルの両方でテーブルレベルのバイナリログをサポートしています。次の表に、サポートされている消費方法を示します。

  Flink のカテゴリ	行指向テーブル	列指向テーブル	行列ハイブリッドテーブル (V1.1 以降でサポート)
  Blink	サポート	サポート	サポート
  Realtime Compute for Apache Flink	サポート	サポート	サポート
  Apache Flink	非サポート	非サポート	非サポート
  JDBC	V1.1 以降でサポート	V1.1 以降でサポート	V1.1 以降でサポート

• Blink は、Hologres バイナリログを消費する際にTIMESTAMP データ型をサポートしていません。Hologres でテーブルを作成する際はTIMESTAMPTZ 型を使用してください。SMALLINT などの他の特殊な型もサポートされていません。

• パーティションテーブルの親テーブルからバイナリログを消費することはできません。子テーブルまたは通常の (パーティション化されていない) テーブルを使用してください。Hologres V1.3.24 からは、子テーブルのバイナリログ TTL をオンデマンドで変更できます。子テーブルにバイナリログ TTL を明示的に指定しない場合、親テーブルから TTL を継承します。バイナリログの TTL は正確なタイマーではない点にご注意ください。システムは、バイナリログが期限切れになった直後に削除されることを保証するものではありません。システムは、期限切れになった後のいずれかの時点でバイナリログを削除します。

• 書き込み負荷が高いワークロードの場合、列指向テーブルでバイナリログを有効にすると、理論的には行指向テーブルよりも多くのオーバーヘッドが発生します。したがって、バイナリログを有効にする場合は、行指向テーブルの使用を推奨します。テーブルが OLAP クエリにも使用される場合は、行列ハイブリッドストレージ形式を使用してください。詳細については、「テーブルストレージ形式」をご参照ください。

• Hologres の内部テーブルのみがバイナリログをサポートしています。外部テーブルはサポートしていません。

バイナリログレコードには、システムフィールドとユーザーテーブルの列が含まれます。次の表で、各フィールドについて説明します。

フィールド名	型	説明
hg_binlog_lsn	BIGINT	ログシーケンス番号 (LSN) を表すバイナリログのシステムフィールドです。値は単調増加ですが、シャード内で連続していることは保証されません。シャード間で一意であることや順序が保証されているわけではありません。 説明 バイナリログデータがシャード間でどのように分散されるかの詳細については、「分散キー」をご参照ください。
hg_binlog_event_type	BIGINT	変更イベントのタイプを示すバイナリログのシステムフィールドです。 hg_binlog_event_type の有効な値は次のとおりです： 5 (INSERT)：新しい行の挿入を表します。 2 (DELETE)：既存の行の削除を表します。 3 (BEFORE_UPDATE)：更新前の既存の行の状態を表します。 7 (AFTER_UPDATE)：更新後の既存の行の状態を表します。
hg_binlog_timestamp_us	BIGINT	システムタイムスタンプをマイクロ秒 (us) 単位で表すバイナリログのシステムフィールドです。
user_table_column_1	ユーザー定義	ユーザーテーブルの列です。
...	...	...
user_table_column_n	ユーザー定義	ユーザーテーブルの列です。

デフォルトでは、Hologres のバイナリログ機能は無効になっています。テーブルプロパティのbinlog.level と binlog.ttl を設定することで有効にできます。以下の例は、バイナリログを有効にする方法を示しています。テーブル作成パラメーターの詳細については、「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' -- バイナリログの 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');-- テーブルプロパティを設定してバイナリログ機能を有効にします。
  call set_table_property('test_message_src', 'binlog.ttl', '86400');-- バイナリログの TTL (秒単位)。
  commit;

次の表に、パラメーターについて説明します。

パラメーター	説明
binlog_level または binlog.level	バイナリログを有効にするかどうかを指定します。有効な値は次のとおりです： replica：有効。 none：無効。
binlog_ttl または binlog.ttl	バイナリログの TTL (秒単位)。デフォルト値：2,592,000 (30日)。

Hologres V1.1 からは、バイナリログを有効または無効にできます。また、TTL を設定して、さまざまな保持要件に対応することもできます。既存のテーブルを再作成することなく、バイナリログを有効にできます。

Hologres のバイナリログデータは厳密に型付けされています。特定のテーブルのバイナリログをクエリするには、組み込みのバイナリログシステムフィールドと元のテーブルの列を組み合わせます。Hologres はまた、最も古い、または最新のバイナリログをクエリしたり、既知の LSN またはタイムスタンプによってバイナリログ情報をクエリしたりするための関数も提供します。

組み込みフィールドによるクエリ

次のステートメントを使用して、組み込みのバイナリログフィールドと元のテーブルフィールドを組み合わせてバイナリログをクエリします。

SELECT hg_binlog_lsn, hg_binlog_event_type, hg_binlog_timestamp_us, * FROM test_message_src;

次のコードは、結果のサンプルです。

postgres=# select hg_binlog_lsn, hg_binlog_event_type, hg_binlog_timestamp_us, * from test_message_src ;
 hg_binlog_lsn | hg_binlog_event_type | hg_binlog_timestamp_us | id |  title  |       body
---------------+----------------------+------------------------+----+---------+-------------------
          1081 |                    5 |       1626856018921653 |  1 | title 1 | body 1
          1092 |                    3 |       1626856054284918 |  1 | title 1 | body 1
          1093 |                    7 |       1626856054284918 |  1 | title 1 | body after update
          1095 |                    2 |       1626856059747536 |  1 | title 1 | body after update
(4 rows)

最も古い、または最新のバイナリログのクエリ

hg_get_binlog_cursor 関数を使用して、特定のシャードで最も古い、または最新のバイナリログをクエリします。次のコードは構文を示します。

-- OLDEST：このシャードで最も古いバイナリログをクエリします。
SELECT * FROM hg_get_binlog_cursor('<table_name>','OLDEST',<shard_id>);
-- LATEST：このシャードで最新のバイナリログをクエリします。
SELECT * FROM hg_get_binlog_cursor('<table_name>','LATEST',<shard_id>);

次のコードは例を示します。

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

次のコードは、結果のサンプルです。

test=# select * from hg_get_binlog_cursor('test_message_src','OLDEST',0);
 hg_shard_id | hg_binlog_lsn | hg_binlog_timestamp_us
-------------+---------------+------------------------
           0 |           152 |       1716204893791755
(1 row)

LSNによるタイムスタンプのクエリ

hg_get_binlog_cursor_by_lsn 関数を使用して、バイナリログのタイムスタンプをクエリします。この関数は、指定された LSN 以上の LSN を持つ最初のバイナリログレコードに関する情報を返します。指定された 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);

次のコードは、結果のサンプルです。

test=# select * from hg_get_binlog_cursor_by_lsn('test_message_src',152,0);
 hg_shard_id | hg_binlog_lsn | hg_binlog_timestamp_us
-------------+---------------+------------------------
           0 |           152 |      1716204893791755
(1 row)

タイムスタンプによる LSN のクエリ

hg_get_binlog_cursor_by_timestamp 関数を使用して、バイナリログの LSN をクエリします。この関数は、指定された時刻以上のタイムスタンプを持つ最初のバイナリログレコードに関する情報を返します。指定された時刻が最新のバイナリログエントリよりも後の場合、結果の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);

次のコードは、結果のサンプルです。

test=# 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);
 hg_shard_id | hg_binlog_lsn | hg_binlog_timestamp_us |          to_timestamp
-------------+---------------+------------------------+-------------------------------
           0 |           152 |    1716204893791755 | 2024-05-20 19:34:53.791755+08
(1 row)

Flink、Blink、および JDBC (Holo Client を含む) を使用して Hologres バイナリログを消費できます。詳細については、次のトピックをご参照ください。

• Flink または Blink を使用してリアルタイムでバイナリログを消費するには、「FlinkまたはBlinkを使用したHologresバイナリログのリアルタイム消費」をご参照ください。

• JDBC を使用してバイナリログを消費するには、「JDBCを使用したHologresバイナリログの消費」をご参照ください。

次の SQL 文を使用して、どのテーブルでバイナリログが有効になっているかを確認できます。

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

以下は結果のサンプルです。クエリはtable_namespace、table_name、property_key、property_value の4つの列を返します。これらの列は、バイナリログが有効になっているすべてのテーブルに関する情報を提供します。