Cara mendekode file log dengan menggunakan wal2json - PolarDB

PolarDB for PostgreSQL menyediakan plugin wal2json yang menghasilkan file log logis dalam format JSON.

Ruang lingkup penerapan

Versi mesin minor PolarDB for PostgreSQL berikut mendukung wal2json:

PostgreSQL 18 (versi mesin minor 2.0.18.1.1.0 dan yang lebih baru)
PostgreSQL 17 (versi mesin minor 2.0.17.7.5.0 dan yang lebih baru)
PostgreSQL 16 (versi mesin minor 2.0.16.6.2.0 dan yang lebih baru)
PostgreSQL 15 (versi mesin minor 2.0.15.12.4.0 dan yang lebih baru)
PostgreSQL 14 (versi mesin minor 2.0.14.5.1.0 dan yang lebih baru)
PostgreSQL 11 (versi mesin minor 2.0.11.9.29.0 dan yang lebih baru)

Catatan

Anda dapat melihat versi mesin minor di Konsol atau dengan menjalankan pernyataan SHOW polardb_version;. Jika versi mesin minor tidak memenuhi persyaratan, lakukan upgrade versi mesin minor.

Informasi latar belakang

wal2json adalah plugin output decoding logis yang menyediakan fitur-fitur berikut:

Mengakses tupel yang dihasilkan oleh INSERT dan UPDATE.
Mengakses versi baris lama dari UPDATE dan DELETE berdasarkan identitas replika yang dikonfigurasi.
Mengonsumsi perubahan melalui protokol streaming (slot replikasi logis) atau API SQL khusus.

Plugin wal2json menghasilkan objek JSON untuk setiap transaksi. Objek tersebut berisi semua tupel baru dan lama. Opsi tambahan dapat mencakup properti seperti timestamp transaksi, skema berkualifikasi, tipe data, dan ID transaksi. Untuk informasi selengkapnya, lihat Mengambil objek JSON dengan menggunakan SQL.

Catatan penggunaan

Karena PolarDB for PostgreSQL menggunakan metode replikasi REPLICA_IDENTITY_FULL, data baris lengkap ditampilkan saat operasi update dan delete, bukan hanya kolom yang berubah. Untuk mencatat hanya kolom yang berubah, nonaktifkan parameter polar_create_table_with_full_replica_identity. Parameter ini tidak dapat diubah melalui Konsol. Hubungi kami untuk bantuan.
Plugin wal2json bergantung pada fitur logical decoding. Nilai parameter wal_level harus diatur ke logical.

Catatan
Anda dapat mengatur parameter wal_level di Konsol. Untuk informasi selengkapnya, lihat Setel parameter kluster. Setelah mengubah parameter ini, kluster akan restart. Rencanakan operasi Anda dan lanjutkan dengan hati-hati.

Mengambil objek JSON dengan menggunakan SQL

Plugin wal2json tidak memerlukan perintah CREATE EXTENSION untuk instalasi. Plugin ini dimuat melalui slot replikasi logis.

Buat slot replikasi logis dengan plugin wal2json, lalu jalankan perintah berikut untuk mengambil objek JSON dari WAL.

-- Buat tabel dengan dan tanpa primary key
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);

-- Buat slot replikasi logis tipe wal2json
SELECT 'init' FROM pg_create_logical_replication_slot('test_slot', 'wal2json');

-- Commit transaksi untuk menulis ke 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;

-- Ambil objek JSON dari WAL
SELECT data FROM pg_logical_slot_get_changes('test_slot', NULL, NULL, 'pretty-print', '1');

Output berikut dikembalikan:

{
    "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"]
        }
    ]
}

Hapus slot replikasi bernama test_slot dan kembalikan string 'stop'.
```
SELECT 'stop' FROM pg_drop_replication_slot('test_slot');
```

Parameter

Tabel berikut menjelaskan parameter wal2json.

Parameter	Deskripsi
change	Entri WAL untuk satu operasi DML, seperti INSERT, UPDATE, DELETE, atau TRUNCATE.
changeset	Kumpulan entri change.
include-xids	Menentukan apakah akan menambahkan ID transaksi (xid) ke setiap changeset. Nilai default: false. Nilai yang valid: true: menambahkan xid ke setiap changeset. false (default): tidak menambahkan xid ke setiap changeset.
include-timestamp	Menentukan apakah akan menambahkan timestamp ke setiap changeset. Nilai default: false. Nilai yang valid: true: menambahkan timestamp ke setiap changeset. false (default): tidak menambahkan timestamp ke setiap changeset.
include-schemas	Menentukan apakah akan menambahkan nama skema ke setiap change. Nilai default: true. Nilai yang valid: true (default): menambahkan nama skema ke setiap change. false: tidak menambahkan nama skema ke setiap change.
include-types	Menentukan apakah akan menambahkan tipe data ke setiap change. Nilai default: true. Nilai yang valid: true (default): menambahkan tipe data ke setiap change. false: tidak menambahkan tipe data ke setiap change.
include-typmod	Menentukan apakah akan menambahkan pengubah tipe ke tipe yang memiliki pengubah, seperti varchar(20) alih-alih varchar. Nilai default: true. Nilai yang valid: true (default): menambahkan pengubah tipe ke tipe yang memiliki pengubah. false: tidak menambahkan pengubah tipe ke tipe yang memiliki pengubah.
include-type-oids	Menentukan apakah akan menambahkan OID tipe. Nilai default: false. Nilai yang valid: true: menambahkan OID tipe. false (default): tidak menambahkan OID tipe.
include-not-null	Menentukan apakah akan menambahkan informasi constraint `not null` sebagai columnoptionals. Nilai default: false. Nilai yang valid: true: menambahkan informasi constraint `not null` sebagai columnoptionals. false (default): tidak menambahkan informasi constraint `not null` sebagai columnoptionals.
pretty-print	Menentukan apakah akan menambahkan spasi putih dan indentasi untuk memformat output JSON. Nilai default: false. Nilai yang valid: true: menambahkan spasi putih dan indentasi untuk memformat output JSON. false (default): tidak menambahkan spasi putih atau indentasi untuk memformat output JSON.
write-in-chunks	Menentukan apakah akan mengeluarkan output setelah setiap change alih-alih setelah setiap changeset. Nilai default: false. Nilai yang valid: true: mengeluarkan output setelah setiap change alih-alih setelah setiap changeset. false (default): mengeluarkan output setelah setiap changeset alih-alih setelah setiap change.
include-lsn	Menentukan apakah akan menambahkan LSN berikutnya (nextlsn) ke setiap changeset. Nilai default: false. Nilai yang valid: true: menambahkan nextlsn ke setiap changeset. false (default): tidak menambahkan nextlsn ke setiap changeset.
filter-tables	Mengecualikan tabel tertentu. Nilai default: kosong, artinya tidak ada tabel yang difilter. Catatan Pisahkan beberapa tabel dengan koma. Setiap tabel harus mencantumkan nama skema. `.foo` cocok dengan tabel foo di semua skema, dan `bar.` cocok dengan semua tabel di skema tersebut. Karakter khusus (spasi, tanda kutip tunggal, koma, titik, dan tanda bintang) harus di-escape dengan backslash. Nama skema dan tabel bersifat case-sensitive. Tabel Foo bar di skema public harus ditentukan sebagai `public.Foo\bar`.
add-tables	Menentukan tabel yang akan didekode. Secara default, semua tabel di semua skema didekode. Sintaksnya sama dengan filter-tables.
filter-msg-prefixes	Mengecualikan baris dengan awalan pesan tertentu. Parameter ini biasanya digunakan dalam fungsi `pg_logical_slot_peek_changes()`. Nilai default: kosong, artinya tidak ada pesan yang difilter. Pisahkan beberapa awalan dengan koma.
add-msg-prefixes	Hanya menyertakan baris dengan awalan pesan tertentu. Parameter ini biasanya digunakan dalam fungsi `pg_logical_slot_peek_changes()`. Nilai default: semua awalan. Pisahkan beberapa awalan dengan koma. Anda harus menggunakan `filter-msg-prefixes` sebelum parameter ini.
format-version	Menentukan versi format output. Nilai default: 1. Nilai yang valid: 1: menggunakan format output versi 1. 2: menggunakan format output versi 2.
actions	Menentukan operasi yang disertakan dalam output. Nilai default: all (INSERT, UPDATE, DELETE, dan TRUNCATE). Jika Anda menggunakan `format-version 1`, TRUNCATE tidak diaktifkan.

Contoh

Bagian ini menggunakan parameter include-xids sebagai contoh untuk menunjukkan cara penggunaannya.

Buat tabel dan slot replikasi logis, lalu masukkan satu baris.

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);

Tentukan nama dan nilai parameter dalam fungsi.

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');

Prinsip desain

Untuk informasi selengkapnya mengenai prinsip desain, lihat Dokumentasi resmi.