wal2json を使用した論理ログファイルのデコード方法 - PolarDB

PolarDB for PostgreSQL およびは、論理ログファイルを JSON 形式で出力する wal2json プラグインを提供しています。

適用範囲

以下の PolarDB for PostgreSQL のマイナーエンジンバージョンが wal2json をサポートしています。

PostgreSQL 18 (マイナーエンジンバージョン 2.0.18.1.1.0 以降)
PostgreSQL 17 (マイナーエンジンバージョン 2.0.17.7.5.0 以降)
PostgreSQL 16 (マイナーエンジンバージョン 2.0.16.6.2.0 以降)
PostgreSQL 15 (マイナーエンジンバージョン 2.0.15.12.4.0 以降)
PostgreSQL 14 (マイナーエンジンバージョン 2.0.14.5.1.0 以降)
PostgreSQL 11 (マイナーエンジンバージョン 2.0.11.9.29.0 以降)

説明

コンソールで、または SHOW polardb_version; ステートメントを実行してマイナーエンジンバージョンを確認できます。マイナーエンジンバージョンが要件を満たさない場合は、マイナーエンジンバージョンをアップグレードしてください。

背景情報

wal2json は論理デコーディングの出力プラグインで、以下の機能を提供します。

INSERT および UPDATE によって生成されたタプルへのアクセス
設定されたレプリカアイデンティティに基づく、UPDATE および DELETE の古い行バージョンへのアクセス
ストリーミングプロトコル (論理レプリケーションスロット) または専用の SQL API を使用した変更の取得

wal2json プラグインは、トランザクションごとに JSON オブジェクトを生成します。JSON オブジェクトには、すべての新旧タプルが含まれます。追加オプションにより、トランザクションのタイムスタンプ、スキーマ名、データ型、トランザクション ID などのプロパティを含めることもできます。詳細については、「SQL による JSON オブジェクトの取得」をご参照ください。

使用上の注意

PolarDB for PostgreSQL はレプリケーション方式として REPLICA_IDENTITY_FULL を使用するため、更新および削除の際に、変更された列のみではなく行全体のデータが出力されます。変更された列のみをログに記録するには、polar_create_table_with_full_replica_identity パラメータを無効にします。このパラメータはコンソールで変更できません。サポートが必要な場合は、お問い合わせください。
wal2json プラグインは、論理デコーディング機能に依存します。このため、wal_level パラメータの値を logical に設定する必要があります。

説明
wal_level パラメータはコンソールで設定できます。詳細については、「クラスターパラメーターの設定」または「」をご参照ください。このパラメータを変更すると、クラスターが再起動します。作業を計画し、注意して実行してください。

SQL による JSON オブジェクトの取得

wal2json プラグインのインストールに CREATE EXTENSION は必要ありません。代わりに、論理レプリケーションスロットを介してロードされます。

wal2json プラグインを使用して論理レプリケーションスロットを作成し、以下のコマンドを実行して WAL から JSON オブジェクトを取得します。

-- プライマリキーあり/なしのテーブルを作成
CREATE TABLE table2_with_pk (a SERIAL, b VARCHAR(30), c TIMESTAMP NOT NULL, PRIMARY KEY(a, c));
CREATE TABLE table2_without_pk (a SERIAL, b NUMERIC(5,2), c TEXT);

-- wal2json タイプの論理レプリケーションスロットを作成
SELECT 'init' FROM pg_create_logical_replication_slot('test_slot', 'wal2json');

-- トランザクションをコミットして WAL に書き込む
BEGIN;
INSERT INTO table2_with_pk (b, c) VALUES('Backup and Restore', now());
INSERT INTO table2_with_pk (b, c) VALUES('Tuning', now());
INSERT INTO table2_with_pk (b, c) VALUES('Replication', now());
DELETE FROM table2_with_pk WHERE a < 3;

INSERT INTO table2_without_pk (b, c) VALUES(2.34, 'Tapir');
UPDATE table2_without_pk SET c = 'Anta' WHERE c = 'Tapir';
COMMIT;

-- WAL から JSON オブジェクトを取得
SELECT data FROM pg_logical_slot_get_changes('test_slot', NULL, NULL, 'pretty-print', '1');

次の出力が返されます。

{
    "change": [
        {
            "kind": "insert",
            "schema": "public",
            "table": "table2_with_pk",
            "columnnames": ["a", "b", "c"],
            "columntypes": ["integer", "character varying(30)", "timestamp without time zone"],
            "columnvalues": [1, "Backup and Restore", "2018-03-27 12:05:29.914496"]
        }
        ,{
            "kind": "insert",
            "schema": "public",
            "table": "table2_with_pk",
            "columnnames": ["a", "b", "c"],
            "columntypes": ["integer", "character varying(30)", "timestamp without time zone"],
            "columnvalues": [2, "Tuning", "2018-03-27 12:05:29.914496"]
        }
        ,{
            "kind": "insert",
            "schema": "public",
            "table": "table2_with_pk",
            "columnnames": ["a", "b", "c"],
            "columntypes": ["integer", "character varying(30)", "timestamp without time zone"],
            "columnvalues": [3, "Replication", "2018-03-27 12:05:29.914496"]
        }
        ,{
            "kind": "delete",
            "schema": "public",
            "table": "table2_with_pk",
            "oldkeys": {
                "keynames": ["a", "c"],
                "keytypes": ["integer", "timestamp without time zone"],
                "keyvalues": [1, "2018-03-27 12:05:29.914496"]
            }
        }
        ,{
            "kind": "delete",
            "schema": "public",
            "table": "table2_with_pk",
            "oldkeys": {
                "keynames": ["a", "c"],
                "keytypes": ["integer", "timestamp without time zone"],
                "keyvalues": [2, "2018-03-27 12:05:29.914496"]
            }
        }
        ,{
            "kind": "insert",
            "schema": "public",
            "table": "table2_without_pk",
            "columnnames": ["a", "b", "c"],
            "columntypes": ["integer", "numeric(5,2)", "text"],
            "columnvalues": [1, 2.34, "Tapir"]
        }
        ,{
            "kind": "update",
            "schema": "public",
            "table": "table2_without_pk",
            "oldkeys": {
                "keynames": ["a", "b", "c"],
                "keytypes": ["integer", "numeric(5,2)", "text"],
                "keyvalues": [1, 2.34, "Tapir"]
            },
            "columnnames": ["a", "b", "c"],
            "columntypes": ["integer", "numeric(5,2)", "text"],
            "columnvalues": [1, 2.34, "Anta"]
        }
    ]
}

test_slot という名前のレプリケーションスロットを削除し、文字列 'stop' を返します。
```
SELECT 'stop' FROM pg_drop_replication_slot('test_slot');
```

パラメータ

以下の表に、wal2json のパラメータを示します。

パラメータ	説明
change	INSERT、UPDATE、DELETE、TRUNCATE などの単一の DML 操作に対する WAL エントリです。
changeset	change エントリのコレクションです。
include-xids	各チェンジセットにトランザクション ID (xid) を追加するかどうかを指定します。デフォルト値は false です。有効な値： true：各チェンジセットに xid を追加します。 false (デフォルト)：各チェンジセットに xid を追加しません。
include-timestamp	各チェンジセットにタイムスタンプを追加するかどうかを指定します。デフォルト値は false です。有効な値： true：各チェンジセットにタイムスタンプを追加します。 false (デフォルト)：各チェンジセットにタイムスタンプを追加しません。
include-schemas	各 change にスキーマ名を追加するかどうかを指定します。デフォルト値は true です。有効な値： true (デフォルト)：各 change にスキーマ名を追加します。 false：各 change にスキーマ名を追加しません。
include-types	各 change にデータ型を追加するかどうかを指定します。デフォルト値は true です。有効な値： true (デフォルト)：各 change にデータ型を追加します。 false：各 change にデータ型を追加しません。
include-typmod	varchar ではなく varchar(20) のように、修飾子を持つ型に対して型修飾子を追加するかどうかを指定します。デフォルト値は true です。有効な値： true (デフォルト)：修飾子を持つ型に型修飾子を追加します。 false：修飾子を持つ型に型修飾子を追加しません。
include-type-oids	型 OID を追加するかどうかを指定します。デフォルト値は false です。有効な値： true：型 OID を追加します。 false (デフォルト)：型 OID を追加しません。
include-not-null	columnoptionals として `NOT NULL` 制約情報を追加するかどうかを指定します。デフォルト値は false です。有効な値： true：columnoptionals として `NOT NULL` 制約情報を追加します。 false (デフォルト)：columnoptionals として `NOT NULL` 制約情報を追加しません。
pretty-print	JSON 出力を整形するために、空白とインデントを追加するかどうかを指定します。デフォルト値は false です。有効な値： true：JSON 出力を整形するために、空白とインデントを追加します。 false (デフォルト)：JSON 出力を整形するための空白やインデントを追加しません。
write-in-chunks	各チェンジセットの後ではなく、各 change の後に出力するかどうかを指定します。デフォルト値は false です。有効な値： true：各チェンジセットの後ではなく、各 change の後に出力します。 false (デフォルト)：各 change の後ではなく、各チェンジセットの後に出力します。
include-lsn	各チェンジセットに次の LSN (nextlsn) を追加するかどうかを指定します。デフォルト値は false です。有効な値： true：各チェンジセットに nextlsn を追加します。 false (デフォルト)：各チェンジセットに nextlsn を追加しません。
filter-tables	特定のテーブルを除外します。デフォルト値は空で、テーブルをフィルタリングしないことを意味します。説明複数のテーブルは、カンマで区切ります。各テーブルにはスキーマ名を含める必要があります。 `.foo` はすべてのスキーマにあるテーブル foo に一致し、`bar.` は bar スキーマ内のすべてのテーブルに一致します。特殊文字 (スペース、単一引用符、カンマ、ピリオド、アスタリスク) は、バックスラッシュでエスケープする必要があります。スキーマ名とテーブル名では大文字と小文字が区別されます。 public スキーマ内のテーブル Foo bar は、`public.Foo\bar` として指定する必要があります。
add-tables	デコードするテーブルを指定します。デフォルトでは、すべてのスキーマ内のすべてのテーブルがデコードされます。構文は filter-tables と同じです。
filter-msg-prefixes	特定のメッセージプレフィックスを持つ行を除外します。このパラメータは通常、`pg_logical_slot_peek_changes()` 関数で使用されます。デフォルト値は空で、メッセージをフィルタリングしないことを意味します。複数のプレフィックスはカンマで区切ります。
add-msg-prefixes	特定のメッセージプレフィックスを持つ行のみを含めます。このパラメータは通常、`pg_logical_slot_peek_changes()` 関数で使用されます。デフォルト値はすべてのプレフィックスです。複数のプレフィックスはカンマで区切ります。このパラメータを使用する前に、`filter-msg-prefixes` を使用する必要があります。
format-version	出力形式のバージョンを指定します。デフォルト値は 1 です。有効な値： 1：出力形式バージョン 1 を使用します。 2：出力形式バージョン 2 を使用します。
actions	出力に含める操作を指定します。デフォルト値はすべて (INSERT、UPDATE、DELETE、TRUNCATE) です。`format-version 1` を使用する場合、TRUNCATE は有効になりません。

例

このセクションでは、include-xids を例に、パラメータの使用方法を説明します。

テーブルと論理レプリケーションスロットを作成し、行を 1 件挿入します。

DROP TABLE IF EXISTS tbl;
CREATE TABLE tbl (id int);
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'wal2json');
INSERT INTO tbl VALUES (1);

関数でパラメータ名と値を指定します。

SELECT
count(*) = 1,
count(distinct ((data::json)->'xid')::text) = 1
FROM pg_logical_slot_get_changes(
'regression_slot', NULL, NULL,
'format-version', '1',
'include-xids', '1');

設計原則

詳細および設計原則については、「公式ドキュメント」をご参照ください。