diff --git a/.github/workflows/run-tests-tiered.yml b/.github/workflows/run-tests-tiered.yml index dcb2777b7..366579712 100644 --- a/.github/workflows/run-tests-tiered.yml +++ b/.github/workflows/run-tests-tiered.yml @@ -149,7 +149,10 @@ jobs: - cdc-endpos-between-transaction - cdc-filtering - cdc-wal2json + - cdc-pgoutput + - cdc-filtering-pgoutput - follow-wal2json + - follow-pgoutput - follow-standby - follow-9.6 - follow-data-only diff --git a/docs/include/clone.rst b/docs/include/clone.rst index 478dd1028..b49349eeb 100644 --- a/docs/include/clone.rst +++ b/docs/include/clone.rst @@ -39,7 +39,8 @@ --not-consistent Allow taking a new snapshot on the source database --snapshot Use snapshot obtained with pg_export_snapshot --follow Implement logical decoding to replay changes - --plugin Output plugin to use (test_decoding, wal2json) + --plugin Output plugin to use (test_decoding, wal2json, pgoutput) + --publication Publication to use with the pgoutput plugin --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin --slot-name Use this Postgres replication slot name --create-slot Create the replication slot diff --git a/docs/include/follow.rst b/docs/include/follow.rst index 253cc1741..42e667169 100644 --- a/docs/include/follow.rst +++ b/docs/include/follow.rst @@ -11,7 +11,8 @@ --resume Allow resuming operations after a failure --not-consistent Allow taking a new snapshot on the source database --snapshot Use snapshot obtained with pg_export_snapshot - --plugin Output plugin to use (test_decoding, wal2json) + --plugin Output plugin to use (test_decoding, wal2json, pgoutput) + --publication Publication to use with the pgoutput plugin --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin --slot-name Use this Postgres replication slot name --create-slot Create the replication slot diff --git a/docs/include/snapshot.rst b/docs/include/snapshot.rst index 46eac2feb..e227133da 100644 --- a/docs/include/snapshot.rst +++ b/docs/include/snapshot.rst @@ -6,7 +6,9 @@ --source Postgres URI to the source database --dir Work directory to use --follow Implement logical decoding to replay changes - --plugin Output plugin to use (test_decoding, wal2json) + --plugin Output plugin to use (test_decoding, wal2json, pgoutput) + --publication Publication to use with the pgoutput plugin + --filters Use the filters defined in --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin --slot-name Use this Postgres replication slot name diff --git a/docs/include/stream-setup.rst b/docs/include/stream-setup.rst index d8835399a..9e848e72e 100644 --- a/docs/include/stream-setup.rst +++ b/docs/include/stream-setup.rst @@ -10,7 +10,8 @@ --resume Allow resuming operations after a failure --not-consistent Allow taking a new snapshot on the source database --snapshot Use snapshot obtained with pg_export_snapshot - --plugin Output plugin to use (test_decoding, wal2json) + --plugin Output plugin to use (test_decoding, wal2json, pgoutput) + --publication Publication to use with the pgoutput plugin --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin --slot-name Stream changes recorded by this slot --origin Name of the Postgres replication origin diff --git a/docs/ref/pgcopydb_clone.rst b/docs/ref/pgcopydb_clone.rst index ac79bc392..811f49a6b 100644 --- a/docs/ref/pgcopydb_clone.rst +++ b/docs/ref/pgcopydb_clone.rst @@ -638,8 +638,14 @@ The following options are available to ``pgcopydb clone``: mostly historical in pgcopydb, it should not make a user visible difference whether you use the default test_decoding or wal2json. + It is also possible to use `pgoutput`__, which is built into Postgres core + since version 10. Use pgoutput when you cannot install an extension on the + source server. See :ref:`pgcopydb_follow` for the ``--publication`` option + that goes with it. + __ https://www.postgresql.org/docs/current/test-decoding.html __ https://github.com/eulerto/wal2json/ + __ https://www.postgresql.org/docs/current/protocol-logical-replication.html --wal2json-numeric-as-string diff --git a/docs/ref/pgcopydb_follow.rst b/docs/ref/pgcopydb_follow.rst index 9838920d9..da138c0f3 100644 --- a/docs/ref/pgcopydb_follow.rst +++ b/docs/ref/pgcopydb_follow.rst @@ -465,8 +465,59 @@ The following options are available to ``pgcopydb follow``: mostly historical in pgcopydb, it should not make a user visible difference whether you use the default test_decoding or wal2json. + It is also possible to use `pgoutput`__, which is built into Postgres core + since version 10. Use pgoutput when you cannot install an extension on the + source server. pgoutput sends a compact binary protocol, so it uses less + network bandwidth and less CPU on the source server than the other two + plugins. + + pgoutput only sends the changes for the tables of a publication. See the + ``--publication`` option for how pgcopydb manages that publication. + __ https://www.postgresql.org/docs/current/test-decoding.html __ https://github.com/eulerto/wal2json/ + __ https://www.postgresql.org/docs/current/protocol-logical-replication.html + +--publication + + Name of the publication to use with the ``--plugin pgoutput`` option. This + option does nothing with the other output plugins. + + When you pass ``--publication``, the publication must already exist on the + source database. pgcopydb never changes it and never drops it. + + When you omit ``--publication``, pgcopydb creates a publication named after + the replication slot, and drops it again during ``pgcopydb stream cleanup``. + The publication lists the tables that the ``--filters`` option selects, so + the source server does the filtering. + + The publication is created together with the replication slot. In a + multi-step migration that is the ``pgcopydb snapshot`` step, not the + ``pgcopydb clone`` step, so pass ``--filters`` to ``pgcopydb snapshot`` as + well:: + + $ pgcopydb snapshot --follow --plugin pgoutput --filters filters.ini + $ pgcopydb stream setup + $ pgcopydb clone --filters filters.ini + + Without ``--filters`` on the snapshot step the publication lists every + table. pgcopydb still filters the changes before it applies them, so the + target stays correct, but the source server decodes and sends rows that are + then discarded. ``pgcopydb clone --follow`` takes the filters in a single + command and does not need this. + + Two limits apply to the publication that pgcopydb creates: + + - The table list is read once, when the replication slot is created. A + table that you create on the source during the migration is not in the + publication, so its changes are not replicated. Add such a table to the + publication yourself, or use ``--publication`` and manage the + publication yourself. + + - ``CREATE PUBLICATION`` is a DDL statement, so it needs a read-write + source server and ownership of every listed table. When the source is a + standby server, create the publication on the primary and then pass + ``--publication``. --wal2json-numeric-as-string diff --git a/docs/ref/pgcopydb_snapshot.rst b/docs/ref/pgcopydb_snapshot.rst index fb2bbefca..b369af060 100644 --- a/docs/ref/pgcopydb_snapshot.rst +++ b/docs/ref/pgcopydb_snapshot.rst @@ -54,8 +54,14 @@ The following options are available to ``pgcopydb snapshot``: mostly historical in pgcopydb, it should not make a user visible difference whether you use the default test_decoding or wal2json. + It is also possible to use `pgoutput`__, which is built into Postgres core + since version 10. Use pgoutput when you cannot install an extension on the + source server. See :ref:`pgcopydb_follow` for the ``--publication`` option + that goes with it. + __ https://www.postgresql.org/docs/current/test-decoding.html __ https://github.com/eulerto/wal2json/ + __ https://www.postgresql.org/docs/current/protocol-logical-replication.html --wal2json-numeric-as-string diff --git a/docs/ref/pgcopydb_stream.rst b/docs/ref/pgcopydb_stream.rst index e291e8200..e7484096b 100644 --- a/docs/ref/pgcopydb_stream.rst +++ b/docs/ref/pgcopydb_stream.rst @@ -311,8 +311,14 @@ The following options are available to ``pgcopydb stream`` sub-commands: mostly historical in pgcopydb, it should not make a user visible difference whether you use the default test_decoding or wal2json. + It is also possible to use `pgoutput`__, which is built into Postgres core + since version 10. Use pgoutput when you cannot install an extension on the + source server. See :ref:`pgcopydb_follow` for the ``--publication`` option + that goes with it. + __ https://www.postgresql.org/docs/current/test-decoding.html __ https://github.com/eulerto/wal2json/ + __ https://www.postgresql.org/docs/current/protocol-logical-replication.html --wal2json-numeric-as-string diff --git a/src/bin/pgcopydb/cli_clone_follow.c b/src/bin/pgcopydb/cli_clone_follow.c index d0c19b4c9..70ca9e619 100644 --- a/src/bin/pgcopydb/cli_clone_follow.c +++ b/src/bin/pgcopydb/cli_clone_follow.c @@ -62,7 +62,8 @@ " --not-consistent Allow taking a new snapshot on the source database\n" \ " --snapshot Use snapshot obtained with pg_export_snapshot\n" \ " --follow Implement logical decoding to replay changes\n" \ - " --plugin Output plugin to use (test_decoding, wal2json)\n" \ + " --plugin Output plugin to use (test_decoding, wal2json, pgoutput)\n" \ + " --publication Publication to use with the pgoutput plugin\n" \ " --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin\n" \ " --slot-name Use this Postgres replication slot name\n" \ " --create-slot Create the replication slot\n" \ @@ -107,7 +108,8 @@ CommandLine follow_command = " --resume Allow resuming operations after a failure\n" " --not-consistent Allow taking a new snapshot on the source database\n" " --snapshot Use snapshot obtained with pg_export_snapshot\n" - " --plugin Output plugin to use (test_decoding, wal2json)\n" + " --plugin Output plugin to use (test_decoding, wal2json, pgoutput)\n" + " --publication Publication to use with the pgoutput plugin\n" " --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin\n" " --slot-name Use this Postgres replication slot name\n" " --create-slot Create the replication slot\n" diff --git a/src/bin/pgcopydb/cli_common.c b/src/bin/pgcopydb/cli_common.c index bc5f75362..c60d5bd72 100644 --- a/src/bin/pgcopydb/cli_common.c +++ b/src/bin/pgcopydb/cli_common.c @@ -521,14 +521,55 @@ cli_read_previous_options(CopyDBOptions *options, CopyFilePaths *cfPaths) return false; } + if (!IS_EMPTY_STRING_BUFFER(options->slot.publicationName) && + !streq(options->slot.publicationName, onFileSlot.publicationName)) + { + log_error("Failed to ensure consistency of --publication"); + log_error("Previous run was done with publication \"%s\" and " + "current run is using --publication \"%s\"", + onFileSlot.publicationName, + options->slot.publicationName); + return false; + } + /* copy the onFileSlot over to our options, wholesale */ options->slot = onFileSlot; } if (options->slot.plugin == STREAM_PLUGIN_UNKNOWN) { - log_fatal("Unknown replication plugin \"%s\", please use either " - "test_decoding (the default) or wal2json", + log_fatal("Unknown replication plugin \"%s\", please use one of " + "test_decoding (the default), wal2json, or pgoutput", + OutputPluginToString(options->slot.plugin)); + return false; + } + + if (options->slot.plugin == STREAM_PLUGIN_PGOUTPUT) + { + /* + * pgoutput only sends the tables of a publication. Without + * --publication, pgcopydb creates one named after the replication + * slot and drops it again in "pgcopydb stream cleanup". + */ + if (IS_EMPTY_STRING_BUFFER(options->slot.publicationName)) + { + strlcpy(options->slot.publicationName, + options->slot.slotName, + sizeof(options->slot.publicationName)); + + options->slot.publicationAutoManaged = true; + } + + log_notice("Using pgoutput with publication \"%s\" (%s)", + options->slot.publicationName, + options->slot.publicationAutoManaged + ? "managed by pgcopydb" + : "managed by the user"); + } + else if (!IS_EMPTY_STRING_BUFFER(options->slot.publicationName)) + { + log_fatal("Option --publication requires --plugin pgoutput, " + "current plugin is \"%s\"", OutputPluginToString(options->slot.plugin)); return false; } @@ -659,6 +700,7 @@ cli_copy_db_getopts(int argc, char **argv) { "defer-validate-fks", no_argument, NULL, 259 }, { "prune-threshold", required_argument, NULL, 260 }, { "prune-min-age", required_argument, NULL, 261 }, + { "publication", required_argument, NULL, 262 }, { "help", no_argument, NULL, 'h' }, { NULL, 0, NULL, 0 } }; @@ -1200,6 +1242,15 @@ cli_copy_db_getopts(int argc, char **argv) break; } + case 262: + { + strlcpy(options.slot.publicationName, optarg, + sizeof(options.slot.publicationName)); + + log_trace("--publication %s", options.slot.publicationName); + break; + } + case '?': default: { diff --git a/src/bin/pgcopydb/cli_snapshot.c b/src/bin/pgcopydb/cli_snapshot.c index 7868ad3ba..03b80dbc1 100644 --- a/src/bin/pgcopydb/cli_snapshot.c +++ b/src/bin/pgcopydb/cli_snapshot.c @@ -32,7 +32,9 @@ CommandLine snapshot_command = " --source Postgres URI to the source database\n" " --dir Work directory to use\n" " --follow Implement logical decoding to replay changes\n" - " --plugin Output plugin to use (test_decoding, wal2json)\n" + " --plugin Output plugin to use (test_decoding, wal2json, pgoutput)\n" + " --publication Publication to use with the pgoutput plugin\n" + " --filters Use the filters defined in \n" " --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin\n" " --slot-name Use this Postgres replication slot name\n", cli_create_snapshot_getopts, @@ -53,6 +55,8 @@ cli_create_snapshot_getopts(int argc, char **argv) { "follow", no_argument, NULL, 'f' }, { "plugin", required_argument, NULL, 'p' }, { "wal2json-numeric-as-string", no_argument, NULL, 'w' }, + { "publication", required_argument, NULL, 262 }, + { "filters", required_argument, NULL, 'F' }, { "slot-name", required_argument, NULL, 's' }, { "version", no_argument, NULL, 'V' }, { "verbose", no_argument, NULL, 'v' }, @@ -73,7 +77,7 @@ cli_create_snapshot_getopts(int argc, char **argv) exit(EXIT_CODE_BAD_ARGS); } - while ((c = getopt_long(argc, argv, "S:D:fp:ws:Vvdzqh", + while ((c = getopt_long(argc, argv, "S:D:fp:wF:s:Vvdzqh", long_options, &option_index)) != -1) { switch (c) @@ -127,6 +131,29 @@ cli_create_snapshot_getopts(int argc, char **argv) break; } + case 262: + { + strlcpy(options.slot.publicationName, optarg, + sizeof(options.slot.publicationName)); + + log_trace("--publication %s", options.slot.publicationName); + break; + } + + case 'F': + { + strlcpy(options.filterFileName, optarg, MAXPGPATH); + log_trace("--filters \"%s\"", options.filterFileName); + + if (!file_exists(options.filterFileName)) + { + log_error("Filters file \"%s\" does not exists", + options.filterFileName); + ++errors; + } + break; + } + case 'V': { /* keeper_cli_print_version prints version and exits. */ @@ -291,6 +318,21 @@ cli_create_snapshot(int argc, char **argv) exit(EXIT_CODE_INTERNAL_ERROR); } + /* + * The pgoutput publication is created here, so the filters have to be + * known now. Without this the publication would list every table and the + * later --filters of the clone step would arrive too late. + */ + if (!IS_EMPTY_STRING_BUFFER(createSNoptions.filterFileName)) + { + if (!parse_filters(createSNoptions.filterFileName, &(copySpecs.filters))) + { + log_error("Failed to parse filters in file \"%s\"", + createSNoptions.filterFileName); + exit(EXIT_CODE_BAD_ARGS); + } + } + /* * We have two ways to create a snapshot: * diff --git a/src/bin/pgcopydb/cli_stream.c b/src/bin/pgcopydb/cli_stream.c index 65d0f153b..1338ca75a 100644 --- a/src/bin/pgcopydb/cli_stream.c +++ b/src/bin/pgcopydb/cli_stream.c @@ -56,7 +56,8 @@ static CommandLine stream_setup_command = " --resume Allow resuming operations after a failure\n" " --not-consistent Allow taking a new snapshot on the source database\n" " --snapshot Use snapshot obtained with pg_export_snapshot\n" - " --plugin Output plugin to use (test_decoding, wal2json)\n" + " --plugin Output plugin to use (test_decoding, wal2json, pgoutput)\n" + " --publication Publication to use with the pgoutput plugin\n" " --wal2json-numeric-as-string Print numeric data type as string when using wal2json output plugin\n" " --slot-name Stream changes recorded by this slot\n" " --origin Name of the Postgres replication origin\n", @@ -207,6 +208,7 @@ cli_stream_getopts(int argc, char **argv) { "dir", required_argument, NULL, 'D' }, { "plugin", required_argument, NULL, 'p' }, { "wal2json-numeric-as-string", no_argument, NULL, 'w' }, + { "publication", required_argument, NULL, 262 }, { "slot-name", required_argument, NULL, 's' }, { "snapshot", required_argument, NULL, 'N' }, { "origin", required_argument, NULL, 'o' }, @@ -297,6 +299,15 @@ cli_stream_getopts(int argc, char **argv) break; } + case 262: + { + strlcpy(options.slot.publicationName, optarg, + sizeof(options.slot.publicationName)); + + log_trace("--publication %s", options.slot.publicationName); + break; + } + case 'N': { strlcpy(options.snapshot, optarg, sizeof(options.snapshot)); diff --git a/src/bin/pgcopydb/copydb.h b/src/bin/pgcopydb/copydb.h index cfd73b810..8e934ad34 100644 --- a/src/bin/pgcopydb/copydb.h +++ b/src/bin/pgcopydb/copydb.h @@ -339,6 +339,8 @@ bool copydb_create_logical_replication_slot(CopyDataSpec *copySpecs, const char *logrep_pguri, ReplicationSlot *slot); +bool snapshot_prepare_publication(CopyDataSpec *copySpecs, + ReplicationSlot *slot); bool snapshot_write_slot(const char *filename, ReplicationSlot *slot); bool snapshot_read_slot(const char *filename, ReplicationSlot *slot); diff --git a/src/bin/pgcopydb/ld_pgoutput.c b/src/bin/pgcopydb/ld_pgoutput.c new file mode 100644 index 000000000..6a1c2a2fb --- /dev/null +++ b/src/bin/pgcopydb/ld_pgoutput.c @@ -0,0 +1,1234 @@ +/* + * src/bin/pgcopydb/ld_pgoutput.c + * pgoutput logical decoding plugin support for pgcopydb. + * + * pgoutput uses a binary wire protocol (proto_version=1) with typed messages. + * All integers are big-endian (network byte order). + * + * Message layout for proto_version=1 (no streaming, no xid in DML): + * + * BEGIN: 'B' u64(final_lsn) u64(commit_time) u32(xid) + * COMMIT: 'C' u8(flags) u64(commit_lsn) u64(end_lsn) u64(commit_time) + * RELATION: 'R' u32(relOid) cstr(nspname) cstr(relname) u8(replident) + * u16(natts) per-col[u8(flags) cstr(name) u32(typeOid) i32(typmod)] + * INSERT: 'I' u32(relOid) 'N' tuple + * UPDATE: 'U' u32(relOid) [('K'|'O') old_tuple] 'N' new_tuple + * DELETE: 'D' u32(relOid) ('K'|'O') old_tuple + * TRUNCATE: 'T' u32(nrelids) u8(flags) nrelids*u32(relOid) + * TYPE: 'Y' ... (filtered out) + * ORIGIN: 'O' ... (filtered out) + * + * Tuple: u16(ncols) per-col[u8(status) if-'t': i32(len) + len bytes] + * status: 'n'=null, 'u'=unchanged TOAST, 't'=text value + * + * The decoded message is serialised as a wal2json-shaped JSON object so that + * the transform and apply steps reuse the existing wal2json code path. + */ + +#include +#include +#include + +#include "postgres.h" +#include "postgres_fe.h" +#include "access/xlogdefs.h" +#include "pqexpbuffer.h" + +#include "defaults.h" +#include "ld_pgoutput.h" +#include "ld_stream.h" +#include "log.h" +#include "pgsql.h" +#include "string_utils.h" + +/* + * Replica identity flag from proto.c (not exported in a header we can use). + */ +#define PGOUT_IS_REPLICA_IDENTITY 0x01 + +/* + * The only two type OIDs the transform step keys on: json takes a ::text cast + * in the WHERE clause, bytea needs wal2json's \x handling. Every other type is + * reported with no "type" property, which binds the value as text. + */ +#define PGOUT_BYTEAOID 17 +#define PGOUT_JSONOID 114 + + +/* + * Big-endian readers. All bounds-check against bufLen and return 0/NULL on + * overflow. + */ +static uint8_t +pgout_u8(const char *buf, int *pos, int bufLen) +{ + if (*pos + 1 > bufLen) + { + log_error("pgoutput: buffer underflow reading u8 at pos %d (len %d)", + *pos, bufLen); + return 0; + } + + uint8_t v = (uint8_t) buf[*pos]; + *pos += 1; + + return v; +} + + +static int16_t +pgout_i16(const char *buf, int *pos, int bufLen) +{ + if (*pos + 2 > bufLen) + { + log_error("pgoutput: buffer underflow reading i16 at pos %d (len %d)", + *pos, bufLen); + return 0; + } + + uint16_t v = + ((uint16_t) (uint8_t) buf[*pos] << 8) | + ((uint16_t) (uint8_t) buf[*pos + 1]); + + *pos += 2; + + return (int16_t) v; +} + + +static uint32_t +pgout_u32(const char *buf, int *pos, int bufLen) +{ + if (*pos + 4 > bufLen) + { + log_error("pgoutput: buffer underflow reading u32 at pos %d (len %d)", + *pos, bufLen); + return 0; + } + + uint32_t v = + ((uint32_t) (uint8_t) buf[*pos] << 24) | + ((uint32_t) (uint8_t) buf[*pos + 1] << 16) | + ((uint32_t) (uint8_t) buf[*pos + 2] << 8) | + ((uint32_t) (uint8_t) buf[*pos + 3]); + + *pos += 4; + + return v; +} + + +static int32_t +pgout_i32(const char *buf, int *pos, int bufLen) +{ + return (int32_t) pgout_u32(buf, pos, bufLen); +} + + +static uint64_t +pgout_u64(const char *buf, int *pos, int bufLen) +{ + if (*pos + 8 > bufLen) + { + log_error("pgoutput: buffer underflow reading u64 at pos %d (len %d)", + *pos, bufLen); + return 0; + } + + uint64_t v = 0; + + for (int i = 0; i < 8; i++) + { + v = (v << 8) | (uint64_t) (uint8_t) buf[*pos + i]; + } + + *pos += 8; + + return v; +} + + +/* + * pgout_cstr returns a pointer into buf at the current position and advances + * *pos past the terminating NUL byte. The returned pointer is not a copy. + */ +static const char * +pgout_cstr(const char *buf, int *pos, int bufLen) +{ + int start = *pos; + + while (*pos < bufLen && buf[*pos] != '\0') + { + ++(*pos); + } + + if (*pos >= bufLen) + { + log_error("pgoutput: unterminated string at pos %d (len %d)", + start, bufLen); + return NULL; + } + + /* skip the NUL byte */ + ++(*pos); + + return buf + start; +} + + +/* ---------- + * Relation cache management. + * ---------- + */ + +/* + * pgoutput_cache_relation parses a RELATION ('R') message starting at pos and + * inserts the relation in privateContext->pgoutputRelationCache. + */ +static bool +pgoutput_cache_relation(StreamContext *privateContext, + const char *buf, int bufLen, int pos) +{ + /* pos is already past the 'R' type byte */ + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + + const char *nspname = pgout_cstr(buf, &pos, bufLen); + + if (nspname == NULL) + { + return false; + } + + /* empty namespace means pg_catalog */ + if (nspname[0] == '\0') + { + nspname = "pg_catalog"; + } + + const char *relname = pgout_cstr(buf, &pos, bufLen); + + if (relname == NULL) + { + return false; + } + + uint8_t replident = pgout_u8(buf, &pos, bufLen); + + int16_t natts = pgout_i16(buf, &pos, bufLen); + + /* a relation is re-sent when its definition changes: drop the stale entry */ + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, &relOid, rel); + + if (rel != NULL) + { + HASH_DEL(privateContext->pgoutputRelationCache, rel); + + PgoutputAttrCache *attr, *tmp; + HASH_ITER(hh, rel->attrs, attr, tmp) + { + HASH_DEL(rel->attrs, attr); + free(attr); + } + + free(rel); + } + + rel = (PgoutputRelationCache *) calloc(1, sizeof(PgoutputRelationCache)); + + if (rel == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + rel->relOid = relOid; + strlcpy(rel->nspname, nspname, sizeof(rel->nspname)); + strlcpy(rel->relname, relname, sizeof(rel->relname)); + rel->replicaIdentity = (char) replident; + rel->natts = natts; + + for (int i = 0; i < natts; i++) + { + uint8_t flags = pgout_u8(buf, &pos, bufLen); + const char *attname = pgout_cstr(buf, &pos, bufLen); + + if (attname == NULL) + { + free(rel); + return false; + } + + uint32_t typeOID = pgout_u32(buf, &pos, bufLen); + pgout_i32(buf, &pos, bufLen); /* typmod - not used */ + + PgoutputAttrCache *attr = + (PgoutputAttrCache *) calloc(1, sizeof(PgoutputAttrCache)); + + if (attr == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + free(rel); + return false; + } + + attr->colIndex = i; + strlcpy(attr->attname, attname, sizeof(attr->attname)); + attr->typeOID = typeOID; + attr->isReplicaIdentity = (flags & PGOUT_IS_REPLICA_IDENTITY) != 0; + + HASH_ADD_INT(rel->attrs, colIndex, attr); + } + + HASH_ADD_INT(privateContext->pgoutputRelationCache, relOid, rel); + + log_debug("pgoutput: cached relation %u %s.%s replident=%c natts=%d", + relOid, rel->nspname, rel->relname, rel->replicaIdentity, natts); + + return true; +} + + +/* ---------- + * Tuple decoder. + * ---------- + */ + +/* + * decode_tuple reads a binary tuple from buf at *pos, allocating *cols_out and + * setting *ncols_out to the tuple width reported on the wire. + */ +static bool +decode_tuple(const char *buf, int bufLen, int *pos, + PgoutputRelationCache *rel, + PgoutputColumn **cols_out, int *ncols_out) +{ + int16_t ncols = pgout_i16(buf, pos, bufLen); + + if (ncols < 0) + { + log_error("pgoutput: negative column count %d in tuple", ncols); + return false; + } + + *ncols_out = ncols; + + if (ncols == 0) + { + *cols_out = NULL; + return true; + } + + PgoutputColumn *cols = + (PgoutputColumn *) calloc(ncols, sizeof(PgoutputColumn)); + + if (cols == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + for (int i = 0; i < ncols; i++) + { + uint8_t status = pgout_u8(buf, pos, bufLen); + cols[i].status = (char) status; + + /* get the column name and type from the relation cache */ + if (rel != NULL) + { + PgoutputAttrCache *attr = NULL; + HASH_FIND_INT(rel->attrs, &i, attr); + + if (attr != NULL) + { + strlcpy(cols[i].name, attr->attname, sizeof(cols[i].name)); + cols[i].typeOID = attr->typeOID; + } + } + + if (status == 't') + { + int32_t len = pgout_i32(buf, pos, bufLen); + + if (len < 0) + { + log_error("pgoutput: negative column value length %d", len); + free(cols); + return false; + } + + if (*pos + len > bufLen) + { + log_error("pgoutput: buffer underflow reading column value " + "(need %d bytes at pos %d, bufLen %d)", + len, *pos, bufLen); + free(cols); + return false; + } + + cols[i].value = strndup(buf + *pos, len); + + if (cols[i].value == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + free(cols); + return false; + } + + *pos += len; + } + else if (status == 'b') + { + /* binary column: read and discard, we never ask for binary */ + int32_t len = pgout_i32(buf, pos, bufLen); + + if (len >= 0 && *pos + len <= bufLen) + { + *pos += len; + } + + cols[i].status = 'u'; /* treat as unchanged for our purposes */ + } + + /* 'n' and 'u' have no payload */ + } + + *cols_out = cols; + + return true; +} + + +/* ---------- + * JSON serialisation, matching the wal2json format-version 2 shape. + * ---------- + */ + +/* + * appendJSONString appends str to buf as a quoted JSON string literal. + */ +static void +appendJSONString(PQExpBuffer buf, const char *str) +{ + appendPQExpBufferChar(buf, '"'); + + for (const unsigned char *p = (const unsigned char *) str; *p; p++) + { + switch (*p) + { + case '"': + { + appendPQExpBufferStr(buf, "\\\""); + break; + } + + case '\\': + { + appendPQExpBufferStr(buf, "\\\\"); + break; + } + + case '\b': + { + appendPQExpBufferStr(buf, "\\b"); + break; + } + + case '\f': + { + appendPQExpBufferStr(buf, "\\f"); + break; + } + + case '\n': + { + appendPQExpBufferStr(buf, "\\n"); + break; + } + + case '\r': + { + appendPQExpBufferStr(buf, "\\r"); + break; + } + + case '\t': + { + appendPQExpBufferStr(buf, "\\t"); + break; + } + + default: + { + if (*p < 0x20) + { + appendPQExpBuffer(buf, "\\u%04x", *p); + } + else + { + appendPQExpBufferChar(buf, (char) *p); + } + + break; + } + } + } + + appendPQExpBufferChar(buf, '"'); +} + + +/* + * appendColumn appends one {"name":..,"type":..,"value":..} object. Values are + * emitted as JSON strings, never numbers: pgoutput already sends the type's + * text output, and a JSON number would cost precision on numeric and float8. + */ +static void +appendColumn(PQExpBuffer buf, PgoutputColumn *col, bool first) +{ + appendPQExpBufferStr(buf, first ? "{" : ",{"); + + appendPQExpBufferStr(buf, "\"name\":"); + appendJSONString(buf, col->name); + + /* bytea reports the value without \x, matching wal2json, which puts it back */ + if (col->typeOID == PGOUT_JSONOID) + { + appendPQExpBufferStr(buf, ",\"type\":\"json\""); + } + else if (col->typeOID == PGOUT_BYTEAOID) + { + appendPQExpBufferStr(buf, ",\"type\":\"bytea\""); + } + + appendPQExpBufferStr(buf, ",\"value\":"); + + if (col->status == 'n' || col->value == NULL) + { + appendPQExpBufferStr(buf, "null"); + } + else if (col->typeOID == PGOUT_BYTEAOID) + { + const char *value = col->value; + + /* strip the \x prefix, the transform step puts it back */ + if (value[0] == '\\' && value[1] == 'x') + { + value += 2; + } + + appendJSONString(buf, value); + } + else + { + appendJSONString(buf, col->value); + } + + appendPQExpBufferChar(buf, '}'); +} + + +/* + * appendTuple appends a JSON array of columns under the given property name. + * + * Status 'u' is always skipped, the value was never sent. In a 'K' section + * (REPLICA IDENTITY DEFAULT) status 'n' marks a non-key placeholder and must + * not reach the WHERE clause; in an 'O' section or a new tuple the same status + * is a genuine NULL and is kept. + */ +static void +appendTuple(PQExpBuffer buf, const char *property, + PgoutputColumn *cols, int ncols, bool keySection) +{ + appendPQExpBuffer(buf, ",\"%s\":[", property); + + bool first = true; + + for (int i = 0; i < ncols; i++) + { + if (cols[i].status == 'u') + { + continue; + } + + if (keySection && cols[i].status == 'n') + { + continue; + } + + appendColumn(buf, &(cols[i]), first); + first = false; + } + + appendPQExpBufferChar(buf, ']'); +} + + +/* + * appendRelation appends the "schema" and "table" properties. + */ +static void +appendRelation(PQExpBuffer buf, const char *nspname, const char *relname) +{ + appendPQExpBufferStr(buf, ",\"schema\":"); + appendJSONString(buf, nspname); + + appendPQExpBufferStr(buf, ",\"table\":"); + appendJSONString(buf, relname); +} + + +/* + * finishJSON returns a copy of the buffer contents and destroys the buffer. + */ +static char * +finishJSON(PQExpBuffer buf) +{ + if (PQExpBufferBroken(buf)) + { + log_error("pgoutput: failed to prepare JSON message: out of memory"); + destroyPQExpBuffer(buf); + return NULL; + } + + char *json = strdup(buf->data); + + destroyPQExpBuffer(buf); + + if (json == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + } + + return json; +} + + +/* + * pgoutputMessageToJSON serialises the decoded message as a wal2json-shaped + * JSON object. + */ +static char * +pgoutputMessageToJSON(PgoutputMessage *msg) +{ + PQExpBuffer buf = createPQExpBuffer(); + + if (buf == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return NULL; + } + + appendPQExpBuffer(buf, "{\"action\":\"%c\",\"xid\":%u", + msg->action, msg->xid); + + switch (msg->action) + { + case 'B': + case 'C': + { + /* transaction control messages only carry the xid */ + break; + } + + case 'I': + { + appendRelation(buf, msg->nspname, msg->relname); + appendTuple(buf, "columns", msg->new_cols, msg->ncols_new, false); + break; + } + + case 'U': + { + appendRelation(buf, msg->nspname, msg->relname); + appendTuple(buf, "columns", msg->new_cols, msg->ncols_new, false); + appendTuple(buf, "identity", msg->old_cols, msg->ncols_old, + msg->oldType == 'K'); + break; + } + + case 'D': + { + appendRelation(buf, msg->nspname, msg->relname); + appendTuple(buf, "identity", msg->old_cols, msg->ncols_old, + msg->oldType == 'K'); + break; + } + + case 'T': + { + appendRelation(buf, msg->nspname, msg->relname); + break; + } + + default: + { + break; + } + } + + appendPQExpBufferChar(buf, '}'); + + return finishJSON(buf); +} + + +/* + * pgoutputTruncateJSON builds the JSON message for one relation of a TRUNCATE + * that targets several relations at once. Index 0 is already emitted by + * preparePgoutputMessage. + */ +char * +pgoutputTruncateJSON(StreamContext *privateContext, int relIndex) +{ + PgoutputMessage *msg = &(privateContext->pgoutputMsg); + + if (relIndex < 0 || relIndex >= msg->ntruncate) + { + log_error("BUG: pgoutputTruncateJSON called with index %d of %d", + relIndex, msg->ntruncate); + return NULL; + } + + PQExpBuffer buf = createPQExpBuffer(); + + if (buf == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return NULL; + } + + appendPQExpBuffer(buf, "{\"action\":\"T\",\"xid\":%u", msg->xid); + + appendRelation(buf, + msg->truncate[relIndex].nspname, + msg->truncate[relIndex].relname); + + appendPQExpBufferChar(buf, '}'); + + return finishJSON(buf); +} + + +/* ---------- + * Public API: parsePgoutputMessageActionAndXid + * ---------- + */ + +/* + * parsePgoutputMessageActionAndXid reads the first byte of the pgoutput binary + * message in context->buffer and sets metadata->action, metadata->xid and + * metadata->filterOut. + */ +bool +parsePgoutputMessageActionAndXid(LogicalStreamContext *context) +{ + StreamContext *privateContext = (StreamContext *) context->private; + LogicalMessageMetadata *metadata = &(privateContext->metadata); + + const char *buf = context->buffer; + int bufLen = context->bufferLen; + + if (bufLen < 1) + { + log_error("pgoutput: empty message (bufLen=%d)", bufLen); + return false; + } + + char msgtype = buf[0]; + int pos = 1; + + switch (msgtype) + { + case 'B': /* BEGIN */ + { + if (bufLen < 21) /* 1 + 8 + 8 + 4 */ + { + log_error("pgoutput: BEGIN message too short (%d bytes)", + bufLen); + return false; + } + + pgout_u64(buf, &pos, bufLen); /* final_lsn */ + pgout_u64(buf, &pos, bufLen); /* commit_time */ + uint32_t xid = pgout_u32(buf, &pos, bufLen); + + metadata->action = STREAM_ACTION_BEGIN; + metadata->xid = xid; + privateContext->currentXid = xid; + break; + } + + case 'C': /* COMMIT */ + { + metadata->action = STREAM_ACTION_COMMIT; + metadata->xid = privateContext->currentXid; + break; + } + + case 'R': /* RELATION - cache it, filter out */ + { + if (!pgoutput_cache_relation(privateContext, buf, bufLen, pos)) + { + log_error("pgoutput: failed to cache RELATION message"); + return false; + } + + metadata->filterOut = true; + break; + } + + case 'Y': /* TYPE - filter out */ + case 'O': /* ORIGIN - filter out */ + { + metadata->filterOut = true; + break; + } + + case 'I': /* INSERT */ + case 'U': /* UPDATE */ + case 'D': /* DELETE */ + case 'T': /* TRUNCATE */ + { + if (bufLen < 5) + { + log_error("pgoutput: DML message too short (%d bytes)", bufLen); + return false; + } + + switch (msgtype) + { + case 'I': + { + metadata->action = STREAM_ACTION_INSERT; + break; + } + + case 'U': + { + metadata->action = STREAM_ACTION_UPDATE; + break; + } + + case 'D': + { + metadata->action = STREAM_ACTION_DELETE; + break; + } + + case 'T': + { + metadata->action = STREAM_ACTION_TRUNCATE; + break; + } + } + + metadata->xid = privateContext->currentXid; + + /* + * A TRUNCATE message starts with the relation count, not with a + * relation OID, so the pgcopydb schema check below only applies to + * the DML messages. + */ + if (msgtype == 'T') + { + break; + } + + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, &relOid, rel); + + if (rel != NULL && streq(rel->nspname, "pgcopydb")) + { + log_debug("pgoutput: filtering out %c message for pgcopydb.%s", + msgtype, rel->relname); + metadata->filterOut = true; + } + + break; + } + + default: + { + log_debug("pgoutput: unknown message type '%c' (0x%02x), " + "filtering out", + msgtype, (unsigned char) msgtype); + metadata->filterOut = true; + break; + } + } + + return true; +} + + +/* ---------- + * Public API: preparePgoutputMessage + * ---------- + */ + +/* + * preparePgoutputMessage decodes the binary pgoutput message into + * privateContext->pgoutputMsg and then serialises it into + * metadata->jsonBuffer, using the same JSON shape that wal2json produces. + */ +bool +preparePgoutputMessage(LogicalStreamContext *context) +{ + StreamContext *privateContext = (StreamContext *) context->private; + LogicalMessageMetadata *metadata = &(privateContext->metadata); + PgoutputMessage *msg = &(privateContext->pgoutputMsg); + + const char *buf = context->buffer; + int bufLen = context->bufferLen; + + /* reset the message struct */ + free_pgoutput_message(msg); + memset(msg, 0, sizeof(PgoutputMessage)); + + if (bufLen < 1) + { + log_error("pgoutput: empty message (bufLen=%d)", bufLen); + return false; + } + + char msgtype = buf[0]; + int pos = 1; + + msg->action = msgtype; + msg->xid = metadata->xid; + msg->lsn = metadata->lsn; + + switch (msgtype) + { + case 'B': + { + /* already parsed in parsePgoutputMessageActionAndXid */ + break; + } + + case 'C': + { + /* clear the XID tracking after COMMIT is decoded */ + privateContext->currentXid = 0; + break; + } + + case 'I': + { + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, &relOid, rel); + + if (rel == NULL) + { + log_error("pgoutput: INSERT for uncached relOid %u", relOid); + return false; + } + + strlcpy(msg->nspname, rel->nspname, sizeof(msg->nspname)); + strlcpy(msg->relname, rel->relname, sizeof(msg->relname)); + + uint8_t marker = pgout_u8(buf, &pos, bufLen); + + if (marker != 'N') + { + log_error("pgoutput: INSERT expected 'N' marker, got '%c'", + marker); + return false; + } + + if (!decode_tuple(buf, bufLen, &pos, rel, + &msg->new_cols, &msg->ncols_new)) + { + return false; + } + + msg->oldType = 0; + break; + } + + case 'U': + { + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, &relOid, rel); + + if (rel == NULL) + { + log_error("pgoutput: UPDATE for uncached relOid %u", relOid); + return false; + } + + strlcpy(msg->nspname, rel->nspname, sizeof(msg->nspname)); + strlcpy(msg->relname, rel->relname, sizeof(msg->relname)); + + uint8_t next = pgout_u8(buf, &pos, bufLen); + + if (next == 'K' || next == 'O') + { + msg->oldType = (char) next; + + if (!decode_tuple(buf, bufLen, &pos, rel, + &msg->old_cols, &msg->ncols_old)) + { + return false; + } + + /* read the 'N' marker for the new tuple */ + uint8_t n_marker = pgout_u8(buf, &pos, bufLen); + + if (n_marker != 'N') + { + log_error("pgoutput: UPDATE expected 'N' after old tuple, " + "got '%c'", n_marker); + return false; + } + } + else if (next == 'N') + { + /* + * No old tuple sent: REPLICA IDENTITY is DEFAULT and the key + * columns did not change. Synthesize the old key tuple from + * the new tuple below. + */ + msg->oldType = 'K'; + } + else + { + log_error("pgoutput: UPDATE unexpected marker '%c'", next); + return false; + } + + if (!decode_tuple(buf, bufLen, &pos, rel, + &msg->new_cols, &msg->ncols_new)) + { + return false; + } + + if (next == 'N') + { + /* count the replica identity columns */ + int nkey = 0; + + for (int i = 0; i < msg->ncols_new; i++) + { + PgoutputAttrCache *attr = NULL; + HASH_FIND_INT(rel->attrs, &i, attr); + + if (attr != NULL && attr->isReplicaIdentity) + { + nkey++; + } + } + + if (nkey == 0) + { + log_error("pgoutput: UPDATE without old tuple and no " + "replica identity columns for %s.%s", + rel->nspname, rel->relname); + return false; + } + + msg->ncols_old = nkey; + msg->old_cols = + (PgoutputColumn *) calloc(nkey, sizeof(PgoutputColumn)); + + if (msg->old_cols == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + int ki = 0; + + for (int i = 0; i < msg->ncols_new && ki < nkey; i++) + { + PgoutputAttrCache *attr = NULL; + HASH_FIND_INT(rel->attrs, &i, attr); + + if (attr == NULL || !attr->isReplicaIdentity) + { + continue; + } + + msg->old_cols[ki] = msg->new_cols[i]; + + /* deep-copy so both tuples own their data */ + if (msg->new_cols[i].status == 't' && + msg->new_cols[i].value != NULL) + { + msg->old_cols[ki].value = + strdup(msg->new_cols[i].value); + + if (msg->old_cols[ki].value == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + } + + ki++; + } + } + + break; + } + + case 'D': + { + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, &relOid, rel); + + if (rel == NULL) + { + log_error("pgoutput: DELETE for uncached relOid %u", relOid); + return false; + } + + strlcpy(msg->nspname, rel->nspname, sizeof(msg->nspname)); + strlcpy(msg->relname, rel->relname, sizeof(msg->relname)); + + uint8_t marker = pgout_u8(buf, &pos, bufLen); + + if (marker != 'K' && marker != 'O') + { + log_error("pgoutput: DELETE expected 'K' or 'O', got '%c'", + marker); + return false; + } + + msg->oldType = (char) marker; + + if (!decode_tuple(buf, bufLen, &pos, rel, + &msg->old_cols, &msg->ncols_old)) + { + return false; + } + + break; + } + + case 'T': + { + uint32_t nrelids = pgout_u32(buf, &pos, bufLen); + pgout_u8(buf, &pos, bufLen); /* flags (cascade, restart_seqs) */ + + if (nrelids == 0) + { + log_error("pgoutput: TRUNCATE message with no relation"); + return false; + } + + msg->truncate = (PgoutputTruncateRel *) + calloc(nrelids, sizeof(PgoutputTruncateRel)); + + if (msg->truncate == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + for (uint32_t i = 0; i < nrelids; i++) + { + uint32_t relOid = pgout_u32(buf, &pos, bufLen); + + PgoutputRelationCache *rel = NULL; + HASH_FIND_INT(privateContext->pgoutputRelationCache, + &relOid, rel); + + if (rel == NULL) + { + log_error("pgoutput: TRUNCATE for uncached relOid %u", + relOid); + return false; + } + + strlcpy(msg->truncate[i].nspname, rel->nspname, + sizeof(msg->truncate[i].nspname)); + strlcpy(msg->truncate[i].relname, rel->relname, + sizeof(msg->truncate[i].relname)); + } + + msg->ntruncate = (int) nrelids; + + /* the first relation is the one emitted by the main JSON buffer */ + strlcpy(msg->nspname, msg->truncate[0].nspname, + sizeof(msg->nspname)); + strlcpy(msg->relname, msg->truncate[0].relname, + sizeof(msg->relname)); + + break; + } + + default: + { + /* filtered-out type - nothing to decode */ + break; + } + } + + metadata->jsonBuffer = pgoutputMessageToJSON(msg); + + if (metadata->jsonBuffer == NULL) + { + log_error("pgoutput: failed to serialise the %c message", msgtype); + return false; + } + + return true; +} + + +/* + * free_pgoutput_message releases the memory owned by a decoded message. + */ +void +free_pgoutput_message(PgoutputMessage *msg) +{ + if (msg == NULL) + { + return; + } + + for (int i = 0; i < msg->ncols_old; i++) + { + if (msg->old_cols != NULL && msg->old_cols[i].value != NULL) + { + free(msg->old_cols[i].value); + } + } + + for (int i = 0; i < msg->ncols_new; i++) + { + if (msg->new_cols != NULL && msg->new_cols[i].value != NULL) + { + free(msg->new_cols[i].value); + } + } + + if (msg->old_cols != NULL) + { + free(msg->old_cols); + } + + if (msg->new_cols != NULL) + { + free(msg->new_cols); + } + + if (msg->truncate != NULL) + { + free(msg->truncate); + } + + msg->old_cols = NULL; + msg->new_cols = NULL; + msg->truncate = NULL; + msg->ncols_old = 0; + msg->ncols_new = 0; + msg->ntruncate = 0; +} diff --git a/src/bin/pgcopydb/ld_pgoutput.h b/src/bin/pgcopydb/ld_pgoutput.h new file mode 100644 index 000000000..5115f04e5 --- /dev/null +++ b/src/bin/pgcopydb/ld_pgoutput.h @@ -0,0 +1,90 @@ +/* + * src/bin/pgcopydb/ld_pgoutput.h + * pgoutput logical decoding plugin support for pgcopydb. + * + * See ld_pgoutput.c for the wire format. ld_stream.h includes this header to + * embed the relation cache in StreamContext, so it must NOT include + * ld_stream.h back. + */ + +#ifndef LD_PGOUTPUT_H +#define LD_PGOUTPUT_H + +#include +#include + +#include "pgsql.h" +#include "uthash.h" + + +typedef struct PgoutputAttrCache +{ + int colIndex; /* hash key: zero-based position */ + char attname[PG_NAMEDATALEN]; + uint32_t typeOID; + bool isReplicaIdentity; /* flags bit 0x01 */ + UT_hash_handle hh; +} PgoutputAttrCache; + + +/* built from the 'R' (RELATION) messages, keyed by relOid */ +typedef struct PgoutputRelationCache +{ + uint32_t relOid; /* hash key */ + char nspname[PG_NAMEDATALEN]; + char relname[PG_NAMEDATALEN]; + char replicaIdentity; /* 'd', 'i', 'f', 'n' */ + int natts; + PgoutputAttrCache *attrs; + UT_hash_handle hh; +} PgoutputRelationCache; + + +typedef struct PgoutputColumn +{ + char name[PG_NAMEDATALEN]; + char status; /* 't' value, 'n' null, 'u' unchanged TOAST */ + uint32_t typeOID; + char *value; /* non-NULL only when status='t' */ +} PgoutputColumn; + + +/* one TRUNCATE message may name several relations */ +typedef struct PgoutputTruncateRel +{ + char nspname[PG_NAMEDATALEN]; + char relname[PG_NAMEDATALEN]; +} PgoutputTruncateRel; + + +typedef struct PgoutputMessage +{ + /* char codes, not StreamAction, to avoid including ld_stream.h */ + char action; /* 'B','C','I','U','D','T' */ + uint32_t xid; + uint64_t lsn; + + char nspname[PG_NAMEDATALEN]; + char relname[PG_NAMEDATALEN]; + + char oldType; /* 'K' key-only, 'O' full-old, 0 absent */ + int ncols_old; + PgoutputColumn *old_cols; /* NULL when oldType==0 */ + + int ncols_new; + PgoutputColumn *new_cols; /* NULL for DELETE */ + + int ntruncate; + PgoutputTruncateRel *truncate; /* NULL unless action=='T' */ +} PgoutputMessage; + + +struct StreamContext; + +bool parsePgoutputMessageActionAndXid(LogicalStreamContext *context); +bool preparePgoutputMessage(LogicalStreamContext *context); +char * pgoutputTruncateJSON(struct StreamContext *privateContext, int relIndex); +void free_pgoutput_message(PgoutputMessage *msg); + + +#endif /* LD_PGOUTPUT_H */ diff --git a/src/bin/pgcopydb/ld_stream.c b/src/bin/pgcopydb/ld_stream.c index 6b40e6683..a12ce281e 100644 --- a/src/bin/pgcopydb/ld_stream.c +++ b/src/bin/pgcopydb/ld_stream.c @@ -133,6 +133,24 @@ stream_init_specs(StreamSpecs *specs, break; } + case STREAM_PLUGIN_PGOUTPUT: + { + KeyVal options = { + .count = 2, + .keywords = { + "proto_version", + "publication_names" + }, + .values = { + "1", + specs->slot.publicationName + } + }; + + specs->pluginOptions = options; + break; + } + default: { log_error("Unknown logical decoding output plugin \"%s\"", @@ -855,6 +873,35 @@ streamWrite(LogicalStreamContext *context) /* update internal transaction counters */ (void) updateStreamCounters(privateContext, metadata); + + /* + * A single pgoutput TRUNCATE message can target several relations at + * once, while our internal representation holds one relation per + * statement. Write one more JSON message for each extra relation. + */ + if (context->plugin == STREAM_PLUGIN_PGOUTPUT && + metadata->action == STREAM_ACTION_TRUNCATE && + privateContext->pgoutputMsg.ntruncate > 1) + { + for (int i = 1; i < privateContext->pgoutputMsg.ntruncate; i++) + { + metadata->jsonBuffer = pgoutputTruncateJSON(privateContext, i); + + if (metadata->jsonBuffer == NULL) + { + /* errors have already been logged */ + return false; + } + + if (!stream_write_json(context, previous)) + { + /* errors have already been logged */ + return false; + } + + (void) updateStreamCounters(privateContext, metadata); + } + } } if (metadata->xid > 0) @@ -1908,6 +1955,11 @@ parseMessageActionAndXid(LogicalStreamContext *context) return parseWal2jsonMessageActionAndXid(context); } + case STREAM_PLUGIN_PGOUTPUT: + { + return parsePgoutputMessageActionAndXid(context); + } + default: { log_error("BUG in parseMessageActionAndXid: unknown plugin %d", @@ -1939,6 +1991,11 @@ prepareMessageJSONbuffer(LogicalStreamContext *context) return prepareWal2jsonMessage(context); } + case STREAM_PLUGIN_PGOUTPUT: + { + return preparePgoutputMessage(context); + } + default: { log_error("BUG in prepareMessageJSONbuffer: unknown plugin %d", @@ -2545,6 +2602,27 @@ stream_cleanup_databases(CopyDataSpec *copySpecs, char *slotName, char *origin) } else { + /* + * Drop the publication when pgcopydb created it for the pgoutput + * plugin. A publication named with --publication belongs to the user + * and is left in place. + */ + ReplicationSlot slot = { 0 }; + + if (file_exists(copySpecs->cfPaths.cdc.slotfile) && + snapshot_read_slot(copySpecs->cfPaths.cdc.slotfile, &slot) && + slot.publicationAutoManaged && + !IS_EMPTY_STRING_BUFFER(slot.publicationName)) + { + if (!pgsql_drop_publication(&src, slot.publicationName)) + { + log_error("Failed to drop publication \"%s\"", + slot.publicationName); + pgsql_finish(&src); + return false; + } + } + log_info("Removing schema pgcopydb and its objects"); if (!pgsql_execute(&src, "drop schema if exists pgcopydb cascade")) diff --git a/src/bin/pgcopydb/ld_stream.h b/src/bin/pgcopydb/ld_stream.h index 4c6067c59..e937daf63 100644 --- a/src/bin/pgcopydb/ld_stream.h +++ b/src/bin/pgcopydb/ld_stream.h @@ -12,6 +12,7 @@ #include "copydb.h" #include "filtering.h" +#include "ld_pgoutput.h" #include "queue_utils.h" #include "pgsql.h" #include "schema.h" @@ -394,6 +395,11 @@ typedef struct StreamContext /* table filtering configuration */ SourceFilters *filters; + /* relation cache and current message for the pgoutput binary protocol */ + PgoutputRelationCache *pgoutputRelationCache; + PgoutputMessage pgoutputMsg; + uint32_t currentXid; /* pgoutput sends the xid in BEGIN only */ + Queue *transformQueue; PGSQL *transformPGSQL; diff --git a/src/bin/pgcopydb/pgsql.c b/src/bin/pgcopydb/pgsql.c index f67e071f9..bb9983f7b 100644 --- a/src/bin/pgcopydb/pgsql.c +++ b/src/bin/pgcopydb/pgsql.c @@ -21,8 +21,11 @@ #include "cli_root.h" #include "defaults.h" +#include "parson.h" + #include "env_utils.h" #include "file_utils.h" +#include "filtering.h" #include "log.h" #include "parsing_utils.h" #include "pgsql.h" @@ -3817,6 +3820,10 @@ OutputPluginFromString(char *plugin) { return STREAM_PLUGIN_WAL2JSON; } + else if (strcmp(plugin, "pgoutput") == 0) + { + return STREAM_PLUGIN_PGOUTPUT; + } return STREAM_PLUGIN_UNKNOWN; } @@ -3845,6 +3852,11 @@ OutputPluginToString(StreamOutputPlugin plugin) return "wal2json"; } + case STREAM_PLUGIN_PGOUTPUT: + { + return "pgoutput"; + } + default: { log_error("Unknown logical decoding output plugin %d", plugin); @@ -4506,6 +4518,7 @@ pgsql_stream_logical(LogicalStreamClient *client, LogicalStreamContext *context) /* call the consumer function */ context->cur_record_lsn = cur_record_lsn; context->buffer = copybuf + hdr_len; + context->bufferLen = r - hdr_len; context->now = client->now; /* the tracking LSN information is updated in the writeFunction */ @@ -5230,6 +5243,290 @@ pgsql_drop_replication_slot(PGSQL *pgsql, const char *slotName) } +/* + * pgsql_create_publication creates a publication FOR TABLE by + * querying the source catalogs on the live connection. System schemas and the + * pgcopydb internal schema are always excluded. + * + * When filters is non-NULL, the include-only and exclude entries are applied + * so that the publication matches the user filter. That gives server-side + * filtering for the pgoutput plugin. + * + * CREATE PUBLICATION ... FOR TABLE needs ownership of each listed table. It + * does not need superuser, unlike FOR ALL TABLES. + */ + +typedef struct PublicationTableListContext +{ + PQExpBuffer tableList; + bool hasTable; +} PublicationTableListContext; + + +static void +publication_table_list_parse(void *ctx, PGresult *result) +{ + PublicationTableListContext *context = (PublicationTableListContext *) ctx; + int nTuples = PQntuples(result); + + for (int i = 0; i < nTuples; i++) + { + char *nspname = PQgetvalue(result, i, 0); + char *relname = PQgetvalue(result, i, 1); + + if (context->hasTable) + { + appendPQExpBufferStr(context->tableList, ", "); + } + + appendPQExpBuffer(context->tableList, "\"%s\".\"%s\"", + nspname, relname); + context->hasTable = true; + } +} + + +/* + * appendStringLiteralPub appends a SQL string literal, with internal single + * quotes doubled. + */ +static void +appendStringLiteralPub(PQExpBuffer buf, const char *str) +{ + appendPQExpBufferChar(buf, '\''); + + for (const char *p = str; *p; p++) + { + if (*p == '\'') + { + appendPQExpBufferChar(buf, '\''); + } + + appendPQExpBufferChar(buf, *p); + } + + appendPQExpBufferChar(buf, '\''); +} + + +bool +pgsql_create_publication(PGSQL *pgsql, const char *pubName, + struct SourceFilters *filters) +{ + PQExpBuffer query = createPQExpBuffer(); + + if (query == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + /* + * Only ordinary and partitioned tables can be published. Reading + * pg_class/pg_namespace directly avoids the pg_tables view, which does not + * report the relkind we need to filter on. + */ + appendPQExpBufferStr(query, + "SELECT n.nspname, c.relname" + " FROM pg_class c" + " JOIN pg_namespace n ON n.oid = c.relnamespace" + " WHERE c.relkind IN ('r', 'p')" + " AND c.relpersistence = 'p'" + " AND n.nspname NOT IN " + "('pg_catalog', 'information_schema', 'pgcopydb')" + " AND n.nspname NOT LIKE 'pg_toast%'" + " AND n.nspname NOT LIKE 'pg_temp%'"); + + if (filters != NULL) + { + /* + * For an include-only filter, restrict the list to the named schemas + * and tables by OR-ing every allowed entry together. + */ + if (filters->type == SOURCE_FILTER_TYPE_INCL) + { + bool firstCond = true; + + appendPQExpBufferStr(query, " AND ("); + + for (int i = 0; i < filters->includeOnlySchemaList.count; i++) + { + if (!firstCond) + { + appendPQExpBufferStr(query, " OR "); + } + + appendPQExpBufferStr(query, "n.nspname = "); + appendStringLiteralPub( + query, filters->includeOnlySchemaList.array[i].nspname); + firstCond = false; + } + + for (int i = 0; i < filters->includeOnlyTableList.count; i++) + { + if (!firstCond) + { + appendPQExpBufferStr(query, " OR "); + } + + appendPQExpBufferStr(query, "(n.nspname = "); + appendStringLiteralPub( + query, filters->includeOnlyTableList.array[i].nspname); + appendPQExpBufferStr(query, " AND c.relname = "); + appendStringLiteralPub( + query, filters->includeOnlyTableList.array[i].relname); + appendPQExpBufferChar(query, ')'); + firstCond = false; + } + + if (firstCond) + { + /* an include-only filter with no entry includes nothing */ + appendPQExpBufferStr(query, "false"); + } + + appendPQExpBufferChar(query, ')'); + } + + /* + * For an exclude filter, strip out the named schemas and tables. + */ + if (filters->type == SOURCE_FILTER_TYPE_EXCL || + filters->type == SOURCE_FILTER_TYPE_LIST_EXCL || + filters->type == SOURCE_FILTER_TYPE_LIST_NOT_INCL) + { + for (int i = 0; i < filters->excludeSchemaList.count; i++) + { + appendPQExpBufferStr(query, " AND n.nspname != "); + appendStringLiteralPub( + query, filters->excludeSchemaList.array[i].nspname); + } + + for (int i = 0; i < filters->excludeTableList.count; i++) + { + appendPQExpBufferStr(query, " AND NOT (n.nspname = "); + appendStringLiteralPub( + query, filters->excludeTableList.array[i].nspname); + appendPQExpBufferStr(query, " AND c.relname = "); + appendStringLiteralPub( + query, filters->excludeTableList.array[i].relname); + appendPQExpBufferChar(query, ')'); + } + } + } + + appendPQExpBufferStr(query, " ORDER BY n.nspname, c.relname"); + + PQExpBuffer tableList = createPQExpBuffer(); + + if (tableList == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + destroyPQExpBuffer(query); + return false; + } + + PublicationTableListContext ctx = { + .tableList = tableList, + .hasTable = false + }; + + if (!pgsql_execute_with_params(pgsql, query->data, 0, NULL, NULL, + &ctx, &publication_table_list_parse)) + { + destroyPQExpBuffer(query); + destroyPQExpBuffer(tableList); + return false; + } + + destroyPQExpBuffer(query); + + PQExpBuffer sql = createPQExpBuffer(); + + if (sql == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + destroyPQExpBuffer(tableList); + return false; + } + + if (ctx.hasTable) + { + appendPQExpBuffer(sql, "CREATE PUBLICATION \"%s\" FOR TABLE %s", + pubName, tableList->data); + } + else + { + appendPQExpBuffer(sql, "CREATE PUBLICATION \"%s\"", pubName); + } + + destroyPQExpBuffer(tableList); + + log_info("Creating publication \"%s\"", pubName); + log_debug("%s", sql->data); + + bool result = pgsql_execute(pgsql, sql->data); + + destroyPQExpBuffer(sql); + + return result; +} + + +/* + * pgsql_drop_publication drops a publication by name. + */ +bool +pgsql_drop_publication(PGSQL *pgsql, const char *pubName) +{ + PQExpBuffer sql = createPQExpBuffer(); + + if (sql == NULL) + { + log_error(ALLOCATION_FAILED_ERROR); + return false; + } + + appendPQExpBuffer(sql, "DROP PUBLICATION IF EXISTS \"%s\"", pubName); + + log_info("Dropping publication \"%s\"", pubName); + + bool result = pgsql_execute(pgsql, sql->data); + + destroyPQExpBuffer(sql); + + return result; +} + + +/* + * pgsql_publication_exists checks that a publication with the given name + * exists on the source server. + */ +bool +pgsql_publication_exists(PGSQL *pgsql, const char *pubName, bool *exists) +{ + SingleValueResultContext context = { { 0 }, PGSQL_RESULT_BOOL, false }; + + char *sql = + "SELECT true FROM pg_publication WHERE pubname = $1"; + + Oid paramTypes[1] = { TEXTOID }; + const char *paramValues[1] = { pubName }; + + if (!pgsql_execute_with_params(pgsql, sql, + 1, paramTypes, paramValues, + &context, &parseSingleValueResult)) + { + return false; + } + + *exists = context.parsedOk && context.boolVal; + + return true; +} + + /* * pgsql_table_exists checks that a table with the given name exists on the * Postgres server. diff --git a/src/bin/pgcopydb/pgsql.h b/src/bin/pgcopydb/pgsql.h index bd553e94f..8bc314175 100644 --- a/src/bin/pgcopydb/pgsql.h +++ b/src/bin/pgcopydb/pgsql.h @@ -421,7 +421,8 @@ typedef enum { STREAM_PLUGIN_UNKNOWN = 0, STREAM_PLUGIN_TEST_DECODING, - STREAM_PLUGIN_WAL2JSON + STREAM_PLUGIN_WAL2JSON, + STREAM_PLUGIN_PGOUTPUT } StreamOutputPlugin; typedef struct LogicalTrackLSN @@ -441,6 +442,7 @@ typedef struct LogicalStreamContext uint32_t WalSegSz; const char *buffer; /* expose internal buffer */ + int bufferLen; /* byte length, pgoutput sends binary data */ StreamOutputPlugin plugin; bool forceFeedback; @@ -515,11 +517,21 @@ typedef struct ReplicationSlot char snapshot[BUFSIZE]; StreamOutputPlugin plugin; bool wal2jsonNumericAsString; + char publicationName[BUFSIZE]; /* pgoutput publication name */ + bool publicationAutoManaged; /* pgcopydb created it, drop on cleanup */ } ReplicationSlot; bool pgsql_create_logical_replication_slot(LogicalStreamClient *client, ReplicationSlot *slot); +/* filtering.h includes this header, so only forward-declare the filter set */ +struct SourceFilters; + +bool pgsql_create_publication(PGSQL *pgsql, const char *pubName, + struct SourceFilters *filters); +bool pgsql_drop_publication(PGSQL *pgsql, const char *pubName); +bool pgsql_publication_exists(PGSQL *pgsql, const char *pubName, bool *exists); + bool pgsql_timestamptz_to_string(TimestampTz ts, char *str, size_t size); bool pgsql_start_replication(LogicalStreamClient *client); diff --git a/src/bin/pgcopydb/snapshot.c b/src/bin/pgcopydb/snapshot.c index 5b915d683..2b7bf352c 100644 --- a/src/bin/pgcopydb/snapshot.c +++ b/src/bin/pgcopydb/snapshot.c @@ -558,6 +558,20 @@ copydb_create_logical_replication_slot(CopyDataSpec *copySpecs, return false; } + /* + * The pgoutput plugin only sends the tables of a publication. Create the + * publication before the slot, so that the slot starts to decode with the + * publication already in place. + */ + if (slot->plugin == STREAM_PLUGIN_PGOUTPUT) + { + if (!snapshot_prepare_publication(copySpecs, slot)) + { + /* errors have already been logged */ + return false; + } + } + if (!pgsql_create_logical_replication_slot(stream, slot)) { log_error("Failed to create a logical replication slot " @@ -624,6 +638,119 @@ copydb_create_logical_replication_slot(CopyDataSpec *copySpecs, } +/* + * snapshot_prepare_publication makes sure that a publication exists for the + * pgoutput plugin. + * + * When the user passed --publication, the named publication must already + * exist and pgcopydb never changes or drops it. Otherwise pgcopydb creates a + * publication named after the replication slot, applies the table filters to + * it, and drops it again in "pgcopydb stream cleanup". + */ +bool +snapshot_prepare_publication(CopyDataSpec *copySpecs, ReplicationSlot *slot) +{ + PGSQL src = { 0 }; + + if (IS_EMPTY_STRING_BUFFER(slot->publicationName)) + { + log_error("BUG: the pgoutput plugin requires a publication name"); + return false; + } + + if (!pgsql_init(&src, copySpecs->connStrings.source_pguri, + PGSQL_CONN_SOURCE)) + { + /* errors have already been logged */ + return false; + } + + if (!slot->publicationAutoManaged) + { + bool exists = false; + + if (!pgsql_publication_exists(&src, slot->publicationName, &exists)) + { + /* errors have already been logged */ + pgsql_finish(&src); + return false; + } + + if (!exists) + { + log_error("Publication \"%s\" does not exist on the source database", + slot->publicationName); + log_info("Create the publication first, or omit --publication to " + "let pgcopydb create and drop one"); + pgsql_finish(&src); + return false; + } + + log_info("Using the existing publication \"%s\"", + slot->publicationName); + + pgsql_finish(&src); + return true; + } + + /* + * CREATE PUBLICATION is a DDL statement, so it cannot run on a read-only + * standby. Ask the user for a publication that already exists on the + * primary instead of failing later with a less clear error. + */ + bool sourceIsReadOnly = false; + + if (!pgsql_is_in_recovery(&src, &sourceIsReadOnly)) + { + log_error("Failed to check if source is in recovery"); + pgsql_finish(&src); + return false; + } + + if (sourceIsReadOnly) + { + log_error("Failed to create publication \"%s\": " + "the source database is a read-only standby", + slot->publicationName); + log_info("Create the publication on the primary, then pass " + "--publication to use it"); + pgsql_finish(&src); + return false; + } + + bool exists = false; + + if (!pgsql_publication_exists(&src, slot->publicationName, &exists)) + { + /* errors have already been logged */ + pgsql_finish(&src); + return false; + } + + /* a --resume run finds the publication that a previous run created */ + if (exists) + { + log_info("Reusing the publication \"%s\" created by pgcopydb", + slot->publicationName); + pgsql_finish(&src); + return true; + } + + if (!pgsql_create_publication(&src, slot->publicationName, + &(copySpecs->filters))) + { + log_error("Failed to create publication \"%s\"", + slot->publicationName); + pgsql_finish(&src); + return false; + } + + pgsql_finish(&src); + + return true; +} + + /* * snapshot_write_slot writes a replication slot information to file. */ @@ -637,6 +764,8 @@ snapshot_write_slot(const char *filename, ReplicationSlot *slot) appendPQExpBuffer(contents, "%s\n", slot->snapshot); appendPQExpBuffer(contents, "%s\n", OutputPluginToString(slot->plugin)); appendPQExpBuffer(contents, "%s\n", boolToString(slot->wal2jsonNumericAsString)); + appendPQExpBuffer(contents, "%s\n", slot->publicationName); + appendPQExpBuffer(contents, "%s\n", boolToString(slot->publicationAutoManaged)); if (PQExpBufferBroken(contents)) { @@ -684,7 +813,11 @@ snapshot_read_slot(const char *filename, ReplicationSlot *slot) return false; } - if (lbuf.count != 5) + /* + * A slot file written before pgoutput support has 5 lines. A slot file + * with the publication information has 7 lines. + */ + if (lbuf.count != 5 && lbuf.count != 7) { log_error("Failed to parse replication slot file \"%s\"", filename); return false; @@ -750,6 +883,25 @@ snapshot_read_slot(const char *filename, ReplicationSlot *slot) filename); } + /* 6. publication name, 7. publication is managed by pgcopydb */ + if (lbuf.count == 7) + { + length = strlcpy(slot->publicationName, lbuf.lines[5], + sizeof(slot->publicationName)); + + if (length >= sizeof(slot->publicationName)) + { + log_error("Failed to read publication name \"%s\" from file \"%s\", " + "length is %lld bytes which exceeds maximum %lld bytes", + lbuf.lines[5], + filename, + (long long) strlen(lbuf.lines[5]), + (long long) sizeof(slot->publicationName)); + return false; + } + + parse_bool(lbuf.lines[6], &(slot->publicationAutoManaged)); + } log_notice("Read replication slot file \"%s\" with snapshot \"%s\", " "slot \"%s\", lsn %X/%X, and plugin \"%s\"", diff --git a/tests/Makefile b/tests/Makefile index 9b40a6f07..2820bfe6c 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -12,9 +12,10 @@ BUILD_ARGS = --build-arg PGVERSION=$(PGVERSION) all: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions \ cdc-wal2json cdc-test-decoding cdc-endpos-between-transaction cdc-low-level \ + cdc-pgoutput cdc-filtering-pgoutput \ cdc-filtering \ cdc-prune \ - follow-wal2json follow-standby follow-9.6 follow-data-only follow-target-reconnect \ + follow-pgoutput follow-wal2json follow-standby follow-9.6 follow-data-only follow-target-reconnect \ endpos-in-multi-wal-txn exclude-extension \ blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks \ fk-partition-clone follow-defer-validate-fks follow-sequence-reset; @@ -61,6 +62,15 @@ cdc-low-level: build cdc-filtering: build $(MAKE) -C $@ +cdc-pgoutput: build + $(MAKE) -C $@ + +cdc-filtering-pgoutput: build + $(MAKE) -C $@ + +follow-pgoutput: build + $(MAKE) -C $@ + follow-wal2json: build $(MAKE) -C $@ @@ -113,6 +123,7 @@ build: .PHONY: all build .PHONY: pagila pagila-multi-steps blobs unit filtering filtering-standby extensions .PHONY: cdc-wal2json cdc-test-decoding cdc-low-level cdc-filtering cdc-prune +.PHONY: cdc-pgoutput cdc-filtering-pgoutput follow-pgoutput .PHONY: follow-wal2json follow-standby follow-9.6 follow-target-reconnect .PHONY: endpos-in-multi-wal-txn exclude-extension .PHONY: blob-snapshot-release follow-defer-indexes fk-not-valid defer-validate-fks diff --git a/tests/cdc-filtering-pgoutput/Dockerfile b/tests/cdc-filtering-pgoutput/Dockerfile new file mode 100644 index 000000000..83ce4be34 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/Dockerfile @@ -0,0 +1,13 @@ +FROM pgcopydb + +USER docker +WORKDIR /usr/src/pgcopydb + +COPY --chmod=755 ./copydb.sh copydb.sh +COPY ./ddl.sql ddl.sql +COPY ./dml.sql dml.sql +COPY ./verify.sql verify.sql +COPY ./filters.ini filters.ini +COPY ./include-only.ini include-only.ini + +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/cdc-filtering-pgoutput/Makefile b/tests/cdc-filtering-pgoutput/Makefile new file mode 100644 index 000000000..5daf0cb69 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/Makefile @@ -0,0 +1,20 @@ +# Copyright (c) 2021 The PostgreSQL Global Development Group. +# Licensed under the PostgreSQL License. + +COMPOSE_EXIT = --exit-code-from=test --abort-on-container-exit + +test: down run down ; + +up: down build + $(DOCKER) compose up $(COMPOSE_EXIT) + +run: build + $(DOCKER) compose run test + +down: + $(DOCKER) compose down + +build: + $(DOCKER) compose build + +.PHONY: run down build test diff --git a/tests/cdc-filtering-pgoutput/compose.yaml b/tests/cdc-filtering-pgoutput/compose.yaml new file mode 100644 index 000000000..aad70bf3e --- /dev/null +++ b/tests/cdc-filtering-pgoutput/compose.yaml @@ -0,0 +1,40 @@ +services: + source: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c wal_level=logical + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + target: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + test: + build: + context: . + dockerfile: Dockerfile + environment: + PGSSLMODE: "require" + PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres + PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres + PGCOPYDB_TABLE_JOBS: 4 + PGCOPYDB_INDEX_JOBS: 2 + PGCOPYDB_OUTPUT_PLUGIN: pgoutput + depends_on: + - source + - target diff --git a/tests/cdc-filtering-pgoutput/copydb.sh b/tests/cdc-filtering-pgoutput/copydb.sh new file mode 100755 index 000000000..b1f6957e5 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/copydb.sh @@ -0,0 +1,164 @@ +#!/bin/bash + +set -x +set -e + +# Disable pager for psql to avoid hanging in non-interactive environments +export PAGER=cat + +# make sure source and target databases are ready +pgcopydb ping + +# Setup source database with multiple schemas and data +psql -o /tmp/ddl.out -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/ddl.sql + +# create the replication slot that captures all the changes +coproc ( pgcopydb snapshot --follow --plugin pgoutput \ + --filters /usr/src/pgcopydb/filters.ini ) + +sleep 1 + +# +# With pgoutput the filtering happens on the source server: an excluded table +# must not be in the publication at all, so its changes are never decoded. +# +pubtables() +{ + psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select schemaname || '.' || tablename + from pg_publication_tables + where pubname = 'pgcopydb' + order by 1" +} + +pubtables > /tmp/pubtables.txt +cat /tmp/pubtables.txt + +for t in public.users public.orders +do + if ! grep -qx "${t}" /tmp/pubtables.txt + then + echo "FAIL: ${t} is missing from the publication" + exit 1 + fi +done + +for t in cron.job_run_details cron.scheduled_jobs \ + excluded_schema.test_table public.filtered_events +do + if grep -qx "${t}" /tmp/pubtables.txt + then + echo "FAIL: excluded table ${t} is in the publication" + exit 1 + fi +done + +echo "PASS: publication matches the filters" + +# now setup the replication origin (target) and the pgcopydb.sentinel (source) +pgcopydb stream setup + +pgcopydb clone --filters /usr/src/pgcopydb/filters.ini + +kill -TERM ${COPROC_PID} +wait ${COPROC_PID} + +# inject CDC changes to BOTH included and excluded schemas +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + +# grab the current LSN, it's going to be our streaming end position +lsn=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c 'select pg_current_wal_lsn()'` + +pgcopydb stream prefetch --resume --endpos "${lsn}" -vv + +CDCDIR="${XDG_DATA_HOME:-$HOME/.local/share}/pgcopydb" +echo "Inspecting transformed CDC SQL under ${CDCDIR}" +ls -l "${CDCDIR}"/*.sql + +# fail closed if there is nothing to check (avoid a vacuous pass) +if ! ls "${CDCDIR}"/*.sql >/dev/null 2>&1; then + echo "FAIL: no CDC .sql files found under ${CDCDIR}" + exit 1 +fi + +# excluded schemas and tables must never reach the CDC files +if grep -nE 'cron|excluded_schema|filtered_events' "${CDCDIR}"/*.sql; then + echo "FAIL: excluded schema/table reference found in transformed CDC SQL" + exit 1 +fi +echo "PASS: no excluded schema/table references in transformed CDC SQL" + +# the same must hold for the raw JSON, since pgoutput filters server-side and +# the excluded rows should never have been decoded in the first place +if ls "${CDCDIR}"/*.json >/dev/null 2>&1; then + if grep -nE 'cron|excluded_schema|filtered_events' "${CDCDIR}"/*.json; then + echo "FAIL: excluded table reached the decoder, publication filter failed" + exit 1 + fi + echo "PASS: excluded tables never reached the decoder" +fi + +# transactions stay balanced even when a transaction was fully filtered +nbegin=$(cat "${CDCDIR}"/*.sql | grep -c '^BEGIN') +ncommit=$(cat "${CDCDIR}"/*.sql | grep -c '^COMMIT') +echo "BEGIN=${nbegin} COMMIT=${ncommit}" +if [ "${nbegin}" != "${ncommit}" ]; then + echo "FAIL: BEGIN/COMMIT counts differ (${nbegin}/${ncommit})" + exit 1 +fi + +# now allow for replaying/catching-up changes +pgcopydb stream sentinel set apply + +pgcopydb stream catchup --resume --endpos "${lsn}" -vv + +# Verify that excluded schemas do not exist and included data is correct +psql -d ${PGCOPYDB_TARGET_PGURI} -f /usr/src/pgcopydb/verify.sql + +users=`psql -At -d ${PGCOPYDB_TARGET_PGURI} -c 'select count(*) from public.users'` +orders=`psql -At -d ${PGCOPYDB_TARGET_PGURI} -c 'select count(*) from public.orders'` + +if [ "${users}" != "5" ] || [ "${orders}" != "2" ] +then + echo "FAIL: expected 5 users and 2 orders, found ${users} and ${orders}" + exit 1 +fi + +echo "PASS: included tables replayed correctly" + +pgcopydb stream cleanup + +# +# Second phase: the include-only filter takes a different branch when the +# publication table list is built, so cover it too. Only the publication is +# checked here, the replay path is already covered above. +# +pgcopydb snapshot --follow --plugin pgoutput --dir /tmp/pgo-include \ + --filters /usr/src/pgcopydb/include-only.ini & +SNAPSHOT_PID=$! + +sleep 3 + +pubtables > /tmp/pubtables2.txt +cat /tmp/pubtables2.txt + +if ! grep -qx "public.users" /tmp/pubtables2.txt +then + echo "FAIL: include-only filter dropped public.users" + exit 1 +fi + +if [ `wc -l < /tmp/pubtables2.txt` != "1" ] +then + echo "FAIL: include-only filter should publish exactly one table" + exit 1 +fi + +echo "PASS: include-only filter builds the right publication" + +kill -TERM ${SNAPSHOT_PID} || true +wait ${SNAPSHOT_PID} || true + +pgcopydb stream cleanup --dir /tmp/pgo-include + +echo "pgoutput filtering test passed" diff --git a/tests/cdc-filtering-pgoutput/ddl.sql b/tests/cdc-filtering-pgoutput/ddl.sql new file mode 100644 index 000000000..6cc46ae91 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/ddl.sql @@ -0,0 +1,67 @@ +-- Create multiple schemas to test filtering +CREATE SCHEMA IF NOT EXISTS public; +CREATE SCHEMA IF NOT EXISTS cron; +CREATE SCHEMA IF NOT EXISTS excluded_schema; + +-- Create tables in public schema (should be included) +CREATE TABLE public.users ( + id serial PRIMARY KEY, + username text NOT NULL, + email text +); + +CREATE TABLE public.orders ( + id serial PRIMARY KEY, + user_id int REFERENCES public.users(id), + amount numeric +); + +-- Create tables in cron schema (should be excluded) +CREATE TABLE cron.job_run_details ( + id serial PRIMARY KEY, + job_id int NOT NULL, + run_time timestamp DEFAULT now(), + status text +); + +CREATE TABLE cron.scheduled_jobs ( + id serial PRIMARY KEY, + job_name text NOT NULL, + schedule text +); + +-- Create tables in excluded_schema (should be excluded) +CREATE TABLE excluded_schema.test_table ( + id serial PRIMARY KEY, + data text +); + +-- Table in the included public schema but excluded by [exclude-table] +CREATE TABLE public.filtered_events ( + id serial PRIMARY KEY, + payload text +); + +-- Insert initial data in public schema +INSERT INTO public.users (username, email) VALUES + ('alice', 'alice@example.com'), + ('bob', 'bob@example.com'), + ('charlie', 'charlie@example.com'); + +INSERT INTO public.orders (user_id, amount) VALUES + (1, 100.50), + (2, 250.75); + +-- Insert initial data in excluded schemas +INSERT INTO cron.job_run_details (job_id, status) VALUES + (1, 'completed'), + (2, 'failed'); + +INSERT INTO cron.scheduled_jobs (job_name, schedule) VALUES + ('cleanup', '0 0 * * *'); + +INSERT INTO excluded_schema.test_table (data) VALUES + ('should not be copied'); + +INSERT INTO public.filtered_events (payload) VALUES + ('seed event, excluded by exclude-table'); diff --git a/tests/cdc-filtering-pgoutput/dml.sql b/tests/cdc-filtering-pgoutput/dml.sql new file mode 100644 index 000000000..340ec32ff --- /dev/null +++ b/tests/cdc-filtering-pgoutput/dml.sql @@ -0,0 +1,46 @@ +-- CDC changes to public schema (should be applied) +INSERT INTO public.users (username, email) VALUES + ('dave', 'dave@example.com'), + ('eve', 'eve@example.com'); + +UPDATE public.users SET email = 'alice.new@example.com' WHERE username = 'alice'; + +DELETE FROM public.orders WHERE id = 1; + +INSERT INTO public.orders (user_id, amount) VALUES + (3, 500.00); + +-- CDC changes to cron schema (should be FILTERED OUT) +INSERT INTO cron.job_run_details (job_id, status) VALUES + (3, 'running'), + (4, 'pending'); + +UPDATE cron.job_run_details SET status = 'completed' WHERE job_id = 2; + +DELETE FROM cron.scheduled_jobs WHERE id = 1; + +INSERT INTO cron.scheduled_jobs (job_name, schedule) VALUES + ('backup', '0 2 * * *'); + +-- CDC changes to excluded_schema (should be FILTERED OUT) +INSERT INTO excluded_schema.test_table (data) VALUES + ('this should not appear on target'); + +UPDATE excluded_schema.test_table SET data = 'updated but should not appear' WHERE id = 1; + +-- CDC changes to public.filtered_events (excluded by [exclude-table], should be FILTERED OUT) +INSERT INTO public.filtered_events (payload) VALUES + ('cdc event that must not appear on target'); + +UPDATE public.filtered_events SET payload = 'updated but excluded' WHERE id = 1; + +-- A transaction that touches ONLY excluded tables: after transform-time +-- filtering this becomes an empty transaction, which must still advance the +-- replication origin (BEGIN/COMMIT emitted) so the migration does not stall. +BEGIN; +INSERT INTO cron.job_run_details (job_id, status) VALUES (5, 'excluded-only-txn'); +INSERT INTO public.filtered_events (payload) VALUES ('excluded-only-txn'); +COMMIT; + +-- More public changes (should be applied) +UPDATE public.users SET username = 'robert' WHERE username = 'bob'; diff --git a/tests/cdc-filtering-pgoutput/filters.ini b/tests/cdc-filtering-pgoutput/filters.ini new file mode 100644 index 000000000..d2685495f --- /dev/null +++ b/tests/cdc-filtering-pgoutput/filters.ini @@ -0,0 +1,6 @@ +[exclude-schema] +cron +excluded_schema + +[exclude-table] +public.filtered_events diff --git a/tests/cdc-filtering-pgoutput/include-only.ini b/tests/cdc-filtering-pgoutput/include-only.ini new file mode 100644 index 000000000..1fb7208b8 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/include-only.ini @@ -0,0 +1,2 @@ +[include-only-table] +public.users diff --git a/tests/cdc-filtering-pgoutput/verify.sql b/tests/cdc-filtering-pgoutput/verify.sql new file mode 100644 index 000000000..498c8f253 --- /dev/null +++ b/tests/cdc-filtering-pgoutput/verify.sql @@ -0,0 +1,37 @@ +-- Verify public schema data was copied AND CDC changes were applied +SELECT 'Checking public.users' as test; +SELECT count(*) as user_count FROM public.users; +-- Should be 5 (3 initial + 2 from CDC) + +SELECT 'Checking alice email update' as test; +SELECT email FROM public.users WHERE username = 'alice'; +-- Should be 'alice.new@example.com' (updated via CDC) + +SELECT 'Checking bob username update' as test; +SELECT username FROM public.users WHERE username = 'robert'; +-- Should exist (bob was renamed to robert via CDC) + +SELECT 'Checking public.orders' as test; +SELECT count(*) as order_count FROM public.orders; +-- Should be 2 (2 initial - 1 deleted + 1 inserted via CDC) + +-- Verify cron schema was EXCLUDED (should not exist on target) +SELECT 'Checking cron schema exclusion' as test; +SELECT count(*) as cron_schema_exists +FROM information_schema.schemata +WHERE schema_name = 'cron'; +-- Should be 0 (schema should not exist) + +-- Verify excluded_schema was EXCLUDED (should not exist on target) +SELECT 'Checking excluded_schema exclusion' as test; +SELECT count(*) as excluded_schema_exists +FROM information_schema.schemata +WHERE schema_name = 'excluded_schema'; +-- Should be 0 (schema should not exist) + +-- Verify public.filtered_events was EXCLUDED by [exclude-table] +SELECT 'Checking public.filtered_events exclusion' as test; +SELECT count(*) as filtered_events_exists +FROM information_schema.tables +WHERE table_schema = 'public' AND table_name = 'filtered_events'; +-- Should be 0 (table excluded from the migration entirely) diff --git a/tests/cdc-pgoutput/Dockerfile b/tests/cdc-pgoutput/Dockerfile new file mode 100644 index 000000000..2085fd11d --- /dev/null +++ b/tests/cdc-pgoutput/Dockerfile @@ -0,0 +1,9 @@ +FROM pagila + +WORKDIR /usr/src/pgcopydb +COPY ./copydb.sh copydb.sh +COPY ./ddl.sql ddl.sql +COPY ./dml.sql dml.sql + +USER docker +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/cdc-pgoutput/Makefile b/tests/cdc-pgoutput/Makefile new file mode 100644 index 000000000..5daf0cb69 --- /dev/null +++ b/tests/cdc-pgoutput/Makefile @@ -0,0 +1,20 @@ +# Copyright (c) 2021 The PostgreSQL Global Development Group. +# Licensed under the PostgreSQL License. + +COMPOSE_EXIT = --exit-code-from=test --abort-on-container-exit + +test: down run down ; + +up: down build + $(DOCKER) compose up $(COMPOSE_EXIT) + +run: build + $(DOCKER) compose run test + +down: + $(DOCKER) compose down + +build: + $(DOCKER) compose build + +.PHONY: run down build test diff --git a/tests/cdc-pgoutput/compose.yaml b/tests/cdc-pgoutput/compose.yaml new file mode 100644 index 000000000..aad70bf3e --- /dev/null +++ b/tests/cdc-pgoutput/compose.yaml @@ -0,0 +1,40 @@ +services: + source: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c wal_level=logical + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + target: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + environment: + POSTGRES_USER: postgres + POSTGRES_PASSWORD: h4ckm3 + POSTGRES_HOST_AUTH_METHOD: trust + command: > + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + test: + build: + context: . + dockerfile: Dockerfile + environment: + PGSSLMODE: "require" + PGCOPYDB_SOURCE_PGURI: postgres://postgres:h4ckm3@source/postgres + PGCOPYDB_TARGET_PGURI: postgres://postgres:h4ckm3@target/postgres + PGCOPYDB_TABLE_JOBS: 4 + PGCOPYDB_INDEX_JOBS: 2 + PGCOPYDB_OUTPUT_PLUGIN: pgoutput + depends_on: + - source + - target diff --git a/tests/cdc-pgoutput/copydb.sh b/tests/cdc-pgoutput/copydb.sh new file mode 100755 index 000000000..c1a87b84b --- /dev/null +++ b/tests/cdc-pgoutput/copydb.sh @@ -0,0 +1,150 @@ +#! /bin/bash + +set -x +set -e + +# Disable pager for psql to avoid hanging in non-interactive environments +export PAGER=cat + +# This script expects the following environment variables to be set: +# +# - PGCOPYDB_SOURCE_PGURI +# - PGCOPYDB_TARGET_PGURI +# - PGCOPYDB_TABLE_JOBS +# - PGCOPYDB_INDEX_JOBS + +# make sure source and target databases are ready +pgcopydb ping + +psql -o /tmp/s.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-schema.sql +psql -o /tmp/d.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-data.sql + +# add the tables that exercise the pgoutput decoder +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/ddl.sql + +# the source database has no wal2json installed, that is the point of pgoutput +psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select count(*) from pg_available_extensions where name = 'wal2json'" + +# create the replication slot and export the snapshot +coproc ( pgcopydb snapshot --follow --plugin pgoutput ) + +sleep 1 + +# pgcopydb creates the publication itself, named after the replication slot +pubcount=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select count(*) from pg_publication where pubname = 'pgcopydb'"` + +if [ "${pubcount}" != "1" ] +then + echo "expected pgcopydb to create the publication \"pgcopydb\"" + psql -d ${PGCOPYDB_SOURCE_PGURI} -c 'select * from pg_publication' + exit 1 +fi + +# the publication must not contain the pgcopydb internal schema +badtables=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select count(*) from pg_publication_tables + where pubname = 'pgcopydb' and schemaname = 'pgcopydb'"` + +if [ "${badtables}" != "0" ] +then + echo "the publication must not contain the pgcopydb schema" + exit 1 +fi + +# now setup the replication origin (target) and the pgcopydb.sentinel (source) +pgcopydb stream setup + +pgcopydb clone --split-tables-larger-than 200kB + +kill -TERM ${COPROC_PID} +wait ${COPROC_PID} + +# now that the base copy is done, inject DML changes on the source +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + +# grab the current LSN, it is our streaming end position +lsn=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c 'select pg_current_wal_lsn()'` + +pgcopydb stream prefetch --resume --endpos "${lsn}" -vv + +# allow the changes to be replayed, then apply them +pgcopydb stream sentinel set apply +pgcopydb stream catchup --resume --endpos "${lsn}" -vv + +# +# Compare the source and the target. Every table below went through the +# pgoutput decoder, so any difference is a decoder bug. +# +compare() +{ + table=$1 + query=$2 + + src=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c "${query}"` + tgt=`psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "${query}"` + + if [ "${src}" != "${tgt}" ] + then + echo "MISMATCH on ${table}" + echo " source: ${src}" + echo " target: ${tgt}" + psql -d ${PGCOPYDB_SOURCE_PGURI} -c "select * from ${table} order by id" + psql -d ${PGCOPYDB_TARGET_PGURI} -c "select * from ${table} order by id" + exit 1 + fi + + echo "OK ${table}: ${src}" +} + +# row counts and a content checksum for each table +compare pgout_full \ + "select count(*), md5(string_agg(id || '|' || coalesce(label,'') || '|' + || coalesce(optional,'') || '|' || amount, + ',' order by id)) + from pgout_full" + +compare pgout_default \ + "select count(*), md5(string_agg(id || '|' || coalesce(label,'') || '|' + || coalesce(filler,''), ',' order by id)) + from pgout_default" + +# the TOASTed column must survive an UPDATE that never sent its value +compare pgout_toast \ + "select count(*), md5(string_agg(id || '|' || md5(big) || '|' + || coalesce(tag,''), ',' order by id)) + from pgout_toast" + +compare pgout_types \ + "select count(*), md5(string_agg(id || '|' || payload::text || '|' + || encode(blob, 'hex') || '|' || flag || '|' + || ts, ',' order by id)) + from pgout_types" + +# both relations of the multi-table TRUNCATE must be empty then refilled +compare pgout_trunc_a \ + "select count(*), md5(string_agg(id || '|' || label, ',' order by id)) + from pgout_trunc_a" + +compare pgout_trunc_b \ + "select count(*), md5(string_agg(id || '|' || label, ',' order by id)) + from pgout_trunc_b" + +# the base copy tables must match too +compare rental "select count(*) from rental" +compare payment "select count(*) from payment" + +# cleanup drops the publication that pgcopydb created +pgcopydb stream cleanup + +pubcount=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select count(*) from pg_publication where pubname = 'pgcopydb'"` + +if [ "${pubcount}" != "0" ] +then + echo "expected \"pgcopydb stream cleanup\" to drop the publication" + exit 1 +fi + +echo "pgoutput CDC test passed" diff --git a/tests/cdc-pgoutput/ddl.sql b/tests/cdc-pgoutput/ddl.sql new file mode 100644 index 000000000..79da6988d --- /dev/null +++ b/tests/cdc-pgoutput/ddl.sql @@ -0,0 +1,81 @@ +--- +--- pgcopydb test/cdc-pgoutput/ddl.sql +--- +--- Extra schema to exercise the pgoutput binary decoder. + +begin; + +--- REPLICA IDENTITY FULL sends the whole old tuple in an 'O' section. +--- A NULL column in that section must render as IS NULL in the WHERE clause. +create table pgout_full ( + id integer primary key, + label text, + optional text, + amount numeric(12,4) +); + +alter table pgout_full replica identity full; + +--- REPLICA IDENTITY DEFAULT sends a key-only 'K' section. The non-key +--- positions arrive with status 'n' as placeholders and must be skipped. +create table pgout_default ( + id integer primary key, + label text, + filler text +); + +--- An unchanged TOAST value arrives with status 'u' and must be left out of +--- the UPDATE statement. +create table pgout_toast ( + id integer primary key, + big text, + tag text +); + +alter table pgout_toast alter column big set storage external; + +--- json has no equality operator, so a REPLICA IDENTITY FULL comparison has +--- to cast both sides to text. +create table pgout_types ( + id integer primary key, + payload json, + blob bytea, + flag boolean, + ts timestamptz +); + +alter table pgout_types replica identity full; + +--- TRUNCATE may name several relations in a single pgoutput message. +create table pgout_trunc_a (id integer primary key, label text); +create table pgout_trunc_b (id integer primary key, label text); + +commit; + +--- seed rows that the base copy carries over to the target +insert into pgout_full + select g, 'label ' || g, case when g % 3 = 0 then null else 'set ' || g end, + (g * 1.2345)::numeric(12,4) + from generate_series(1, 50) g; + +insert into pgout_default + select g, 'label ' || g, repeat('f', 20) from generate_series(1, 50) g; + +--- incompressible payload so the value is stored out of line and TOASTed +insert into pgout_toast + select g, + (select string_agg(md5(random()::text), '') + from generate_series(1, 400)), + 'tag ' || g + from generate_series(1, 5) g; + +insert into pgout_types + select g, + ('{"n": ' || g || '}')::json, + decode(lpad(to_hex(g), 8, '0'), 'hex'), + g % 2 = 0, + '2024-01-01 00:00:00+00'::timestamptz + (g || ' hours')::interval + from generate_series(1, 20) g; + +insert into pgout_trunc_a select g, 'a' || g from generate_series(1, 10) g; +insert into pgout_trunc_b select g, 'b' || g from generate_series(1, 10) g; diff --git a/tests/cdc-pgoutput/dml.sql b/tests/cdc-pgoutput/dml.sql new file mode 100644 index 000000000..931469339 --- /dev/null +++ b/tests/cdc-pgoutput/dml.sql @@ -0,0 +1,74 @@ +--- +--- pgcopydb test/cdc-pgoutput/dml.sql +--- +--- DML that exercises every branch of the pgoutput binary decoder. + +begin; + +--- plain INSERT, 'N' tuple only +insert into pgout_full + select g, 'label ' || g, case when g % 3 = 0 then null else 'set ' || g end, + (g * 1.2345)::numeric(12,4) + from generate_series(51, 60) g; + +--- UPDATE that leaves the key alone: pgoutput sends no old tuple and the +--- decoder has to synthesize the key from the new tuple. +update pgout_full set label = 'renamed' where id between 1 and 5; + +--- UPDATE that changes the key: pgoutput sends an 'O' section (identity full) +update pgout_full set id = id + 1000 where id between 10 and 12; + +--- DELETE against REPLICA IDENTITY FULL where a column is NULL. The WHERE +--- clause must use IS NULL for that column, otherwise nothing is deleted. +delete from pgout_full where id in (3, 6, 9); + +commit; + +begin; + +--- REPLICA IDENTITY DEFAULT: only the key reaches the 'K' section +update pgout_default set label = 'changed' where id between 20 and 30; +delete from pgout_default where id between 40 and 45; + +insert into pgout_default + select g, 'label ' || g, repeat('g', 20) from generate_series(51, 55) g; + +commit; + +begin; + +--- UPDATE that does not touch the TOASTed column. pgoutput reports it with +--- status 'u' and the value must stay unchanged on the target. +update pgout_toast set tag = 'updated tag' where id <= 3; + +--- UPDATE that does replace the TOASTed value +update pgout_toast + set big = (select string_agg(md5(random()::text), '') + from generate_series(1, 400)) + where id = 4; + +commit; + +begin; + +--- json, bytea, boolean and timestamptz round trip +update pgout_types set payload = '{"n": -1}'::json where id <= 5; +update pgout_types set flag = not flag where id between 6 and 10; +delete from pgout_types where id between 15 and 17; + +insert into pgout_types + select g, + ('{"n": ' || g || '}')::json, + decode(lpad(to_hex(g), 8, '0'), 'hex'), + g % 2 = 0, + '2024-01-01 00:00:00+00'::timestamptz + (g || ' hours')::interval + from generate_series(21, 25) g; + +commit; + +--- a single TRUNCATE naming two relations produces one pgoutput message +--- carrying both relation OIDs +truncate table pgout_trunc_a, pgout_trunc_b; + +insert into pgout_trunc_a select g, 'a' || g from generate_series(100, 105) g; +insert into pgout_trunc_b select g, 'b' || g from generate_series(100, 105) g; diff --git a/tests/follow-pgoutput/Dockerfile b/tests/follow-pgoutput/Dockerfile new file mode 100644 index 000000000..2085fd11d --- /dev/null +++ b/tests/follow-pgoutput/Dockerfile @@ -0,0 +1,9 @@ +FROM pagila + +WORKDIR /usr/src/pgcopydb +COPY ./copydb.sh copydb.sh +COPY ./ddl.sql ddl.sql +COPY ./dml.sql dml.sql + +USER docker +CMD ["/usr/src/pgcopydb/copydb.sh"] diff --git a/tests/follow-pgoutput/Dockerfile.inject b/tests/follow-pgoutput/Dockerfile.inject new file mode 100644 index 000000000..1f4b69e3a --- /dev/null +++ b/tests/follow-pgoutput/Dockerfile.inject @@ -0,0 +1,14 @@ +FROM pgcopydb + +USER root + +RUN apt-get update \ + && apt-get install -y --no-install-recommends jq \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /usr/src/pgcopydb +COPY ./inject.sh inject.sh +COPY ./dml.sql dml.sql + +USER docker +CMD ["/usr/src/pgcopydb/inject.sh"] diff --git a/tests/follow-pgoutput/Makefile b/tests/follow-pgoutput/Makefile new file mode 100644 index 000000000..8bc9d3dce --- /dev/null +++ b/tests/follow-pgoutput/Makefile @@ -0,0 +1,33 @@ +# Copyright (c) 2021 The PostgreSQL Global Development Group. +# Licensed under the PostgreSQL License. + +COMPOSE_EXIT = --exit-code-from=test --abort-on-container-exit + +test: down run down ; + +up: down build + $(DOCKER) compose up $(COMPOSE_EXIT) + +run: build fix-volumes + $(DOCKER) compose run test + +down: + $(DOCKER) compose down + +build: + $(DOCKER) compose build + +VPATH = /var/run/pgcopydb +CNAME = follow-pgoutput +VNAME = follow-pgoutput +VOLUMES = -v $(VNAME):$(VPATH) +OPTS = --env-file=../paths.env $(VOLUMES) +CLEANUP = make -f /var/lib/postgres/cleanup.mk + +fix-volumes: + $(DOCKER) run --rm $(OPTS) $(CNAME) $(CLEANUP) + +attach: + $(DOCKER) run --rm -it $(OPTS) $(CNAME) bash + +.PHONY: run down build test fix-volumes diff --git a/tests/follow-pgoutput/compose.yaml b/tests/follow-pgoutput/compose.yaml new file mode 100644 index 000000000..25d5131e7 --- /dev/null +++ b/tests/follow-pgoutput/compose.yaml @@ -0,0 +1,56 @@ +services: + source: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + env_file: + - ../postgres.env + command: > + -c wal_level=logical + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + + target: + image: postgres:${PGVERSION:-16} + expose: + - 5432 + env_file: + - ../postgres.env + command: > + -c ssl=on + -c ssl_cert_file=/etc/ssl/certs/ssl-cert-snakeoil.pem + -c ssl_key_file=/etc/ssl/private/ssl-cert-snakeoil.key + + inject: + build: + context: . + dockerfile: Dockerfile.inject + env_file: + - ../uris.env + - ../paths.env + volumes: + - follow-pgoutput:/var/run/pgcopydb + + test: + image: follow-pgoutput + build: . + cap_add: + - SYS_ADMIN + - SYS_PTRACE + environment: + PGCOPYDB_TABLE_JOBS: 4 + PGCOPYDB_INDEX_JOBS: 2 + env_file: + - ../uris.env + - ../paths.env + volumes: + - follow-pgoutput:/var/run/pgcopydb + depends_on: + - source + - target + - inject + +volumes: + follow-pgoutput: + external: true diff --git a/tests/follow-pgoutput/copydb.sh b/tests/follow-pgoutput/copydb.sh new file mode 100755 index 000000000..8e5908888 --- /dev/null +++ b/tests/follow-pgoutput/copydb.sh @@ -0,0 +1,87 @@ +#! /bin/bash + +set -x +set -e + +export PAGER=cat + +# This script expects the following environment variables to be set: +# +# - PGCOPYDB_SOURCE_PGURI +# - PGCOPYDB_TARGET_PGURI +# - PGCOPYDB_TABLE_JOBS +# - PGCOPYDB_INDEX_JOBS + +pgcopydb ping + +psql -o /tmp/s.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-schema.sql +psql -o /tmp/d.out -d ${PGCOPYDB_SOURCE_PGURI} -1 -f /usr/src/pagila/pagila-data.sql + +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/ddl.sql + +# a single clone --follow run, which is what a migration actually uses +pgcopydb clone --follow --plugin pgoutput + +pgcopydb stream sentinel get + +# +# Compare source and target for every table that went through the decoder. +# +compare() +{ + table=$1 + query=$2 + + src=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c "${query}"` + tgt=`psql -At -d ${PGCOPYDB_TARGET_PGURI} -c "${query}"` + + if [ "${src}" != "${tgt}" ] + then + echo "MISMATCH on ${table}" + echo " source: ${src}" + echo " target: ${tgt}" + exit 1 + fi + + echo "OK ${table}: ${src}" +} + +compare pgout_full \ + "select count(*), md5(string_agg(id || '|' || coalesce(label,'') || '|' + || coalesce(optional,'') || '|' || amount, + ',' order by id)) + from pgout_full" + +compare pgout_default \ + "select count(*), md5(string_agg(id || '|' || coalesce(label,''), ',' order by id)) + from pgout_default" + +# the TOASTed value must survive updates that never sent it +compare pgout_toast \ + "select count(*), md5(string_agg(id || '|' || md5(big) || '|' + || coalesce(tag,''), ',' order by id)) + from pgout_toast" + +compare pgout_types \ + "select count(*), md5(string_agg(id || '|' || payload::text || '|' + || encode(blob, 'hex'), ',' order by id)) + from pgout_types" + +compare rental "select count(*) from rental" +compare payment "select count(*) from payment" + +# make sure the inject service has had time to see the final sentinel values +sleep 2 + +pgcopydb stream cleanup + +pubcount=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} \ + -c "select count(*) from pg_publication where pubname = 'pgcopydb'"` + +if [ "${pubcount}" != "0" ] +then + echo "expected \"pgcopydb stream cleanup\" to drop the publication" + exit 1 +fi + +echo "follow pgoutput test passed" diff --git a/tests/follow-pgoutput/ddl.sql b/tests/follow-pgoutput/ddl.sql new file mode 100644 index 000000000..79da6988d --- /dev/null +++ b/tests/follow-pgoutput/ddl.sql @@ -0,0 +1,81 @@ +--- +--- pgcopydb test/cdc-pgoutput/ddl.sql +--- +--- Extra schema to exercise the pgoutput binary decoder. + +begin; + +--- REPLICA IDENTITY FULL sends the whole old tuple in an 'O' section. +--- A NULL column in that section must render as IS NULL in the WHERE clause. +create table pgout_full ( + id integer primary key, + label text, + optional text, + amount numeric(12,4) +); + +alter table pgout_full replica identity full; + +--- REPLICA IDENTITY DEFAULT sends a key-only 'K' section. The non-key +--- positions arrive with status 'n' as placeholders and must be skipped. +create table pgout_default ( + id integer primary key, + label text, + filler text +); + +--- An unchanged TOAST value arrives with status 'u' and must be left out of +--- the UPDATE statement. +create table pgout_toast ( + id integer primary key, + big text, + tag text +); + +alter table pgout_toast alter column big set storage external; + +--- json has no equality operator, so a REPLICA IDENTITY FULL comparison has +--- to cast both sides to text. +create table pgout_types ( + id integer primary key, + payload json, + blob bytea, + flag boolean, + ts timestamptz +); + +alter table pgout_types replica identity full; + +--- TRUNCATE may name several relations in a single pgoutput message. +create table pgout_trunc_a (id integer primary key, label text); +create table pgout_trunc_b (id integer primary key, label text); + +commit; + +--- seed rows that the base copy carries over to the target +insert into pgout_full + select g, 'label ' || g, case when g % 3 = 0 then null else 'set ' || g end, + (g * 1.2345)::numeric(12,4) + from generate_series(1, 50) g; + +insert into pgout_default + select g, 'label ' || g, repeat('f', 20) from generate_series(1, 50) g; + +--- incompressible payload so the value is stored out of line and TOASTed +insert into pgout_toast + select g, + (select string_agg(md5(random()::text), '') + from generate_series(1, 400)), + 'tag ' || g + from generate_series(1, 5) g; + +insert into pgout_types + select g, + ('{"n": ' || g || '}')::json, + decode(lpad(to_hex(g), 8, '0'), 'hex'), + g % 2 = 0, + '2024-01-01 00:00:00+00'::timestamptz + (g || ' hours')::interval + from generate_series(1, 20) g; + +insert into pgout_trunc_a select g, 'a' || g from generate_series(1, 10) g; +insert into pgout_trunc_b select g, 'b' || g from generate_series(1, 10) g; diff --git a/tests/follow-pgoutput/dml.sql b/tests/follow-pgoutput/dml.sql new file mode 100644 index 000000000..57886b16a --- /dev/null +++ b/tests/follow-pgoutput/dml.sql @@ -0,0 +1,34 @@ +--- +--- pgcopydb test/follow-pgoutput/dml.sql +--- +--- This file runs many times during the follow test, so every statement has +--- to be safe to repeat. + +begin; + +update pgout_full set label = 'renamed ' || clock_timestamp() where id <= 5; + +update pgout_default set label = 'changed ' || clock_timestamp() + where id between 20 and 25; + +--- leave the TOASTed column alone, it arrives with status 'u' +update pgout_toast set tag = 'tag ' || clock_timestamp() where id <= 3; + +update pgout_types set payload = '{"n": -1}'::json where id <= 5; + +commit; + +begin; + +--- pagila traffic so the base tables move as well +with r as + ( + insert into rental(rental_date, inventory_id, customer_id, staff_id, last_update) + select '2022-06-01', 371, 291, 1, '2022-06-01' + returning rental_id, customer_id, staff_id + ) + insert into payment(customer_id, staff_id, rental_id, amount, payment_date) + select customer_id, staff_id, rental_id, 5.99, '2022-06-01' + from r; + +commit; diff --git a/tests/follow-pgoutput/inject.sh b/tests/follow-pgoutput/inject.sh new file mode 100755 index 000000000..dbae917de --- /dev/null +++ b/tests/follow-pgoutput/inject.sh @@ -0,0 +1,78 @@ +#! /bin/bash + +set -x +set -e + +# This script expects the following environment variables to be set: +# +# - PGCOPYDB_SOURCE_PGURI +# - PGCOPYDB_TARGET_PGURI +# - PGCOPYDB_TABLE_JOBS +# - PGCOPYDB_INDEX_JOBS + +pgcopydb ping + +# +# Only start injecting DML traffic on the source database when the pagila +# schema and base data set has been deployed already. Our proxy to know that +# that's the case is the existence of the pgcopydb.sentinel table on the +# source database. +# +dbfile=${TMPDIR}/pgcopydb/schema/source.db + +until [ -s ${dbfile} ] +do + sleep 1 +done + +# +# Inject changes from our DML file in a loop, again and again. +# +# Every other round of DML changes, we also force the source server to +# switch to another WAL file, to test that our streaming solution can follow +# WAL file changes. +# +for i in `seq 5` +do + psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + sleep 1 + + psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + sleep 1 + + psql -d ${PGCOPYDB_SOURCE_PGURI} -c 'select pg_switch_wal()' + sleep 1 +done + +# grab the current LSN, it's going to be our streaming end position +lsn=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c 'select pg_current_wal_flush_lsn()'` + +pgcopydb stream sentinel set endpos --current --debug +pgcopydb stream sentinel get + +endpos=`pgcopydb stream sentinel get --endpos 2>/dev/null` + +if [ ${endpos} = "0/0" ] +then + echo "expected ${lsn} endpos, found ${endpos}" + exit 1 +fi + +# +# Because we're using docker-compose --abort-on-container-exit make sure +# that the other process in the pgcopydb service is done before exiting +# here. +# +flushlsn="0/0" + +while [ ${flushlsn} \< ${endpos} ] +do + flushlsn=`pgcopydb stream sentinel get --flush-lsn 2>/dev/null` + sleep 1 +done + +# +# Still give some time to the pgcopydb service to finish its processing, +# with the cleanup and all. +# +sleep 10