Add pgoutput as an optional logical decoding plugin

.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

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

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

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 <filename>          Use the filters defined in <filename>
      --wal2json-numeric-as-string  Print numeric data type as string when using wal2json output plugin
      --slot-name                   Use this Postgres replication slot name
 

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

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
 

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
 

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
 

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
 

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"

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:
 			{

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 <filename>          Use the filters defined in <filename>\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:
 	 *

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

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