Como decodificar arquivos de log usando wal2json - PolarDB: Central de Documentação do Alibaba Cloud

Escopo de aplicação

As seguintes versões menores do mecanismo do PolarDB for PostgreSQL suportam o wal2json:

PostgreSQL 18 (versão menor do mecanismo 2.0.18.1.1.0 e posteriores)
PostgreSQL 17 (versão menor do mecanismo 2.0.17.7.5.0 e posteriores)
PostgreSQL 16 (versão menor do mecanismo 2.0.16.6.2.0 e posteriores)
PostgreSQL 15 (versão menor do mecanismo 2.0.15.12.4.0 e posteriores)
PostgreSQL 14 (versão menor do mecanismo 2.0.14.5.1.0 e posteriores)
PostgreSQL 11 (versão menor do mecanismo 2.0.11.9.29.0 e posteriores)

Nota

Você pode visualize the minor engine version no console ou executando a instrução SHOW polardb_version;. Caso a versão menor do mecanismo não atenda aos requisitos, upgrade the minor engine version。

Informações de fundo

O wal2json é um plugin de saída de decodificação lógica que oferece os seguintes recursos:

Acesso às tuplas geradas por INSERT e UPDATE.
Acesso a versões antigas de linhas de UPDATE e DELETE, conforme a identidade de réplica configurada.
Consumo de alterações via protocolo de streaming (slots de replicação lógica) ou por uma API SQL dedicada.

O plugin wal2json gera um objeto JSON para cada transação. Esse objeto contém todas as tuplas novas e antigas. Opções adicionais permitem incluir propriedades como timestamps de transação, schemas qualificados, tipos de dados e IDs de transação. Para mais detalhes, consulte Retrieve JSON objects by using SQL.

Observações de uso

Como o PolarDB for PostgreSQL utiliza REPLICA_IDENTITY_FULL como método de replicação, todos os dados da linha aparecem durante atualizações e exclusões, e não apenas as colunas alteradas. Para registrar somente as colunas modificadas, desative o parâmetro polar_create_table_with_full_replica_identity . Não é possível modificar esse parâmetro pelo console. Contact us para obter assistência.
O plugin wal2json depende do recurso de decodificação lógica. O valor do parâmetro wal_level deve ser definido como logical.

Nota
É possível definir o parâmetro wal_level no console. Para mais informações, consulte Defina cluster parameters. Após a modificação desse parâmetro, o cluster será reiniciado. Planeje suas operações e proceda com cautela.

Obtenção de objetos JSON via SQL

A instalação do plugin wal2json não exige CREATE EXTENSION. Em vez disso, ele é carregado por meio de um slot de replicação lógica.

Crie um slot de replicação lógica com o plugin wal2json e execute os comandos abaixo para obter objetos JSON do WAL.

-- Create tables with and without primary keys
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);

-- Create a logical replication slot of the wal2json type
SELECT 'init' FROM pg_create_logical_replication_slot('test_slot', 'wal2json');

-- Commit the transaction to write to 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;

-- Retrieve JSON objects from WAL
SELECT data FROM pg_logical_slot_get_changes('test_slot', NULL, NULL, 'pretty-print', '1');

A seguinte saída é retornada:

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

Exclua o slot de replicação chamado test_slot e retorne a string 'stop'.
```
SELECT 'stop' FROM pg_drop_replication_slot('test_slot');
```

Parâmetros

A tabela a seguir descreve os parâmetros do wal2json.

Parâmetro	Descrição
change	Entrada WAL referente a uma única operação DML, como INSERT, UPDATE, DELETE ou TRUNCATE.
changeset	Conjunto de entradas de alteração.
include-xids	Define se o ID da transação (xid) será adicionado a cada changeset. Valor padrão: false. Valores válidos: true: adiciona o xid a cada changeset. false (padrão): não adiciona o xid a cada changeset.
include-timestamp	Define se um timestamp será adicionado a cada changeset. Valor padrão: false. Valores válidos: true: adiciona um timestamp a cada changeset. false (padrão): não adiciona um timestamp a cada changeset.
include-schemas	Define se o nome do schema será adicionado a cada alteração. Valor padrão: true. Valores válidos: true (padrão): adiciona o nome do schema a cada alteração. false: não adiciona o nome do schema a cada alteração.
include-types	Define se os tipos de dados serão adicionados a cada alteração. Valor padrão: true. Valores válidos: true (padrão): adiciona os tipos de dados a cada alteração. false: não adiciona os tipos de dados a cada alteração.
include-typmod	Define se modificadores de tipo serão adicionados aos tipos que os possuem, como varchar(20) em vez de apenas varchar. Valor padrão: true. Valores válidos: true (padrão): adiciona modificadores aos tipos que os possuem. false: não adiciona modificadores aos tipos que os possuem.
include-type-oids	Define se os OIDs de tipo serão adicionados. Valor padrão: false. Valores válidos: true: adiciona os OIDs de tipo. false (padrão): não adiciona os OIDs de tipo.
include-not-null	Define se as informações de restrição `not null` serão adicionadas como columnoptionals. Valor padrão: false. Valores válidos: true: adiciona as informações de restrição `not null` como columnoptionals. false (padrão): não adiciona as informações de restrição `not null` como columnoptionals.
pretty-print	Define se espaços em branco e recuos serão adicionados para formatar a saída JSON. Valor padrão: false. Valores válidos: true: adiciona espaços em branco e recuos para formatar a saída JSON. false (padrão): não adiciona espaços em branco nem recuos para formatar a saída JSON.
write-in-chunks	Define se a saída será emitida após cada alteração em vez de após cada changeset. Valor padrão: false. Valores válidos: true: emite a saída após cada alteração em vez de após cada changeset. false (padrão): emite a saída após cada changeset em vez de após cada alteração.
include-lsn	Define se o próximo LSN (nextlsn) será adicionado a cada changeset. Valor padrão: false. Valores válidos: true: adiciona o nextlsn a cada changeset. false (padrão): não adiciona o nextlsn a cada changeset.
filter-tables	Exclui tabelas específicas. Valor padrão: vazio, o que significa que nenhuma tabela é filtrada. Nota Separe várias tabelas com vírgulas. Cada tabela deve incluir o nome do schema. `.foo` corresponde à tabela foo em todos os schemas, e `bar.` corresponde a todas as tabelas do schema. Caracteres especiais (espaços, aspas simples, vírgulas, pontos e asteriscos) devem ser escapados com barra invertida. Os nomes de schemas e tabelas diferenciam maiúsculas de minúsculas. A tabela Foo bar no schema public deve ser especificada como `public.Foo\bar`.
add-tables	Especifica as tabelas a serem decodificadas. Por padrão, todas as tabelas de todos os schemas são decodificadas. A sintaxe é a mesma de filter-tables.
filter-msg-prefixes	Exclui linhas com prefixos de mensagem específicos. Este parâmetro é geralmente usado na função `pg_logical_slot_peek_changes()`. Valor padrão: vazio, o que significa que nenhuma mensagem é filtrada. Separe vários prefixos com vírgulas.
add-msg-prefixes	Inclui apenas linhas com prefixos de mensagem específicos. Este parâmetro é geralmente usado na função `pg_logical_slot_peek_changes()`. Valor padrão: todos os prefixos. Separe vários prefixos com vírgulas. É necessário usar `filter-msg-prefixes` antes deste parâmetro.
format-version	Especifica a versão do formato de saída. Valor padrão: 1. Valores válidos: 1: usa a versão 1 do formato de saída. 2: usa a versão 2 do formato de saída.
actions	Especifica as operações a serem incluídas na saída. Valor padrão: all (INSERT, UPDATE, DELETE e TRUNCATE). Se você usar `format-version 1`, o TRUNCATE não esteja habilitado.

Exemplo

Esta seção utiliza include-xids como exemplo para demonstrar o uso dos parâmetros.

Crie uma tabela e um slot de replicação lógica e insira uma linha.

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

Especifique o nome e o valor do parâmetro na função.

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

Princípios de design

Para mais informações e princípios de design, consulte a Documentação oficial.