On Thu, Apr 14, 2022 at 1:53 PM Peter Smith <smithpb2...@gmail.com> wrote:
>
> Here are my review comments for v8-0001.
>
> ======
>
> 1. Commit message
>
> CREATE SUBSCRIPTION sub1 CONNECTION 'dbname =postgres port=9999'
> PUBLICATION pub1 with (local_only = true);
>
> The spaces in the CONNECTION string are a bit strange.
Modified
> ~~~
>
> 2. src/backend/catalog/pg_subscription.
>
> @@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
> sub->stream = subform->substream;
> sub->twophasestate = subform->subtwophasestate;
> sub->disableonerr = subform->subdisableonerr;
> + sub->local_only = subform->sublocal;
>
> Maybe call it subform->sublocalonly;
>
Modified
> ~~~
>
> 3. src/backend/catalog/system_views.sql
>
> @@ -1290,8 +1290,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
> -- All columns of pg_subscription except subconninfo are publicly readable.
> REVOKE ALL ON pg_subscription FROM public;
> GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
> - subbinary, substream, subtwophasestate,
> subdisableonerr, subslotname,
> - subsynccommit, subpublications)
> + subbinary, substream, subtwophasestate, subdisableonerr,
> + sublocal, subslotname, subsynccommit, subpublications)
>
> Maybe call the column sublocalonly
>
Modified
> ~~~
>
> 4. .../libpqwalreceiver/libpqwalreceiver.c
>
> @@ -453,6 +453,10 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
> PQserverVersion(conn->streamConn) >= 150000)
> appendStringInfoString(&cmd, ", two_phase 'on'");
>
> + if (options->proto.logical.local_only &&
> + PQserverVersion(conn->streamConn) >= 150000)
> + appendStringInfoString(&cmd, ", local_only 'on'");
>
> Add a FIXME comment here as a reminder that this condition needs to
> change for PG16.
>
Modified
> ~~~
>
> 5. src/bin/pg_dump/pg_dump.c
>
> @@ -4361,6 +4361,7 @@ getSubscriptions(Archive *fout)
> int i_subsynccommit;
> int i_subpublications;
> int i_subbinary;
> + int i_sublocal;
> int i,
> ntups;
>
> Maybe call this member i_sublocalonly;
>
Modified
> ~~~
>
> 6. src/bin/pg_dump/pg_dump.c
> + if (fout->remoteVersion >= 150000)
> + appendPQExpBufferStr(query, " s.sublocal\n");
> + else
> + appendPQExpBufferStr(query, " false AS sublocal\n");
> +
>
> 6a. Add a FIXME comment as a reminder that this condition needs to
> change for PG16.
>
Modified
> 6b. Change the column name to sublocalonly.
>
Modified
> ~~~
>
> 7. src/bin/pg_dump/pg_dump.h
>
> @@ -661,6 +661,7 @@ typedef struct _SubscriptionInfo
> char *subdisableonerr;
> char *subsynccommit;
> char *subpublications;
> + char *sublocal;
> } SubscriptionInfo;
>
> Change the member name to sublocalonly
>
Modified
> ~~~
>
> 8. src/bin/psql/describe.c
>
> @@ -6241,6 +6241,12 @@ describeSubscriptions(const char *pattern, bool
> verbose)
> gettext_noop("Two phase commit"),
> gettext_noop("Disable on error"));
>
> + /* local_only is supported only in v15 and higher */
> + if (pset.sversion >= 150000)
> + appendPQExpBuffer(&buf,
> + ", sublocal AS \"%s\"\n",
> + gettext_noop("Local only"));
>
> 7a. The comment is wrong to mention v15.
>
I have removed this comment and added a FIXME. I will add it in once
version change is done to avoid confusion.
> 7b. Add a FIXME comment as a reminder that this condition needs to
> change for PG16.
>
Modified
> 7c. Change the column name to sublocalonly.
>
Modified
> ~~~
>
> 9. src/include/catalog/pg_subscription.h
>
> @@ -70,6 +70,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId)
> BKI_SHARED_RELATION BKI_ROW
>
> bool substream; /* Stream in-progress transactions. */
>
> + bool sublocal; /* skip copying of remote origin data */
>
> Change the member name to sublocalonly
>
Modified
> ~~~
>
> 10. src/include/replication/logicalproto.h
>
> @@ -32,12 +32,17 @@
> *
> * LOGICALREP_PROTO_TWOPHASE_VERSION_NUM is the minimum protocol version with
> * support for two-phase commit decoding (at prepare time). Introduced in
> PG15.
> + *
> + * LOGICALREP_PROTO_LOCALONLY_VERSION_NUM is the minimum protocol version
> with
> + * support for sending only locally originated data from the publisher.
> + * Introduced in PG16.
>
> Add a FIXME comment here as a reminder that the proto version number
> needs to be bumped to 4 in PG16.
>
Modified
> ~~~
>
> 11. src/test/subscription/t/032_circular.pl
>
> Perhaps there should be another test using a third "Node_C" which
> publishes some data to Node_B. Then you can ensure that by using
> local_only (when Node_A is subscribing to Node_A) that nothing from
> the Node_C can find its way onto Node_A. But then the test name
> "circular" is a bit misleading. Maybe it should be "032_localonly"?
Added the test and changed the file.
Thanks for the comments, attached v9 patch has the changes for the same.
Regards,
Vignesh
From 12aac489a07e0093eb2e6d13aada4470e2bd4fe8 Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignes...@gmail.com>
Date: Sat, 16 Apr 2022 19:30:44 +0530
Subject: [PATCH v9 2/2] Support force option for copy_data, check and throw an
error if publisher tables were also subscribing data in the publisher from
other publishers.
This patch does couple of things:
1) Added force option for copy_data.
2) Check and throw an error if the publication tables were also subscribing
data in the publisher from other publishers.
Let's consider an existing Multi master logical replication setup between
Node1 and Node2 that is created using the following steps:
a) Node1 - Publication publishing employee table.
b) Node2 - Subscription subscribing from publication pub1 with
local_only.
c) Node2 - Publication publishing employee table.
d) Node1 - Subscription subscribing from publication pub2 with
local_only.
Now when user is trying to add another node Node3 to the
above Multi master logical replication setup:
a) user will have to create one subscription subscribing from Node1 to
Node3
b) user wil have to create another subscription subscribing from
Node2 to Node3 using local_only option and copy_data as true.
While the subscription is created, server will identify that Node2 is
subscribing from Node1 and Node1 is subscribing from Node2 and throw an
error so that user can handle the initial copy data.
Handling of initial data copying in this case is detailed in
the documentation section of the patch.
---
doc/src/sgml/logical-replication.sgml | 215 ++++++++++++++++
doc/src/sgml/ref/alter_subscription.sgml | 8 +-
doc/src/sgml/ref/create_subscription.sgml | 19 +-
src/backend/commands/subscriptioncmds.c | 155 +++++++++---
src/test/regress/expected/subscription.out | 18 +-
src/test/regress/sql/subscription.sql | 12 +
src/test/subscription/t/032_localonly.pl | 279 +++++++++++++++++----
src/tools/pgindent/typedefs.list | 1 +
8 files changed, 630 insertions(+), 77 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 1126ce4ccf..05ce154032 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -700,4 +700,219 @@ CREATE SUBSCRIPTION mysub CONNECTION 'dbname=foo host=bar user=repuser' PUBLICAT
incremental changes to those tables.
</para>
</sect1>
+
+<sect1 id="bidirectional-logical-replication-quick-setup">
+ <title>Setting Bidirection logical replication between two nodes:</title>
+
+ <itemizedlist>
+ <listitem>
+ <para>
+ Create the publication in node1:
+<programlisting>
+CREATE PUBLICATION pub_node1 FOR TABLE t1;
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create the subscription in node2 to subscribe the changes from node1:
+<programlisting>
+CREATE SUBSCRIPTION sub_node1_node2 CONNECTION 'dbname=foo host=node1 user=repuser' PUBLICATION pub_node1 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create publication in node2:
+<programlisting>
+CREATE PUBLICATION pub_node2 FOR TABLE t1;
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node1 to subscribe the changes from node2:
+<programlisting>
+CREATE SUBSCRIPTION sub_node2_node1 CONNECTION 'dbname=foo host=node2 user=repuser' PUBLICATION pub_node2 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+ </itemizedlist>
+
+ <para>
+ Now the BiDirectional logical replication setup is complete between node1
+ and node2. Any incremental changes from node1 will be replicated to node2 and
+ the incremental changes from node2 will be replicated to node1.
+ </para>
+
+ <sect2 id="add-new-node">
+ <title>Adding new node to the two node bidirectional logical replication setup</title>
+ <para>
+ Let's see a slightly complex setup where we try to add node3 to the above
+ setup in various different scenarios:
+ </para>
+
+ <sect3 id="data-not-present-in-any-node">
+ <title>When there is no data present in any of the nodes node1, node2 or node3</title>
+ <itemizedlist>
+ <listitem>
+ <para>
+ Create publication in node3:
+<programlisting>
+CREATE PUBLICATION pub_node3 FOR TABLE t1;
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node1 to subscribe the changes from node3:
+<programlisting>
+CREATE SUBSCRIPTION sub_node1_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node2 to subscribe the changes from node3:
+<programlisting>
+CREATE SUBSCRIPTION sub_node2_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node3 to subscribe the changes from node1:
+<programlisting>
+CREATE SUBSCRIPTION sub_node3_node1 CONNECTION 'dbname=foo host=node1 user=repuser' PUBLICATION pub_node1 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node3 to subscribe the changes from node2:
+<programlisting>
+CREATE SUBSCRIPTION sub_node3_node2 CONNECTION 'dbname=foo host=node2 user=repuser' PUBLICATION pub_node2 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+ </itemizedlist>
+ </sect3>
+
+ <sect3 id="data-present-in-existing-node">
+ <title>When data is present in the exiting nodes node1 and node2</title>
+ <itemizedlist>
+ <listitem>
+ <para>
+ Create publication in node3:
+<programlisting>
+CREATE PUBLICATION pub_node3 FOR TABLE t1;
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node1 to subscribe the changes from node3:
+<programlisting>
+CREATE SUBSCRIPTION sub_node1_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node2 to subscribe the changes from node3:
+<programlisting>
+CREATE SUBSCRIPTION sub_node2_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node3 to subscribe the changes from node1, here
+ copy_data is specified as force so that the existing table data is
+ copied during initial sync:
+<programlisting>
+CREATE SUBSCRIPTION sub_node3_node1 CONNECTION 'dbname=foo host=node1 user=repuser' PUBLICATION pub_node1 WITH (copy_data = force, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node3 to subscribe the changes from node2:
+<programlisting>
+CREATE SUBSCRIPTION sub_node3_node2 CONNECTION 'dbname=foo host=node2 user=repuser' PUBLICATION pub_node2 WITH (copy_data = off, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+ </itemizedlist>
+ </sect3>
+
+ <sect3 id="data-present-in-new-node">
+ <title>When data is present in the new node node3</title>
+ <itemizedlist>
+ <listitem>
+ <para>
+ Create publication in node3:
+<programlisting>
+CREATE PUBLICATION pub_node3 FOR TABLE t1;
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node1 to subscribe the changes from node3, here
+ copy_data is specified as force so that the existing table data is
+ copied during initial sync:
+<programlisting>
+CREATE SUBSCRIPTION sub_node1_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = force, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node2 to subscribe the changes from node3, here
+ copy_data is specified as force so that the existing table data is
+ copied during initial sync:
+<programlisting>
+CREATE SUBSCRIPTION sub_node2_node3 CONNECTION 'dbname=foo host=node3 user=repuser' PUBLICATION pub_node3 WITH (copy_data = force, local_only = on);
+</programlisting>
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ Create subscription in node3 to subscribe the changes from node1 and
+ node2, here copy_data is specified as force when creating subscription
+ to node1 so that the existing table data is copied during initial sync:
+<programlisting>
+# Truncate the table data but do not replicate the truncate.
+ALTER PUBLICATION pub_node3 SET (publish='insert,update,delete');
+TRUNCATE t1;
+
+CREATE SUBSCRIPTION sub_node3_node1 CONNECTION 'dbname=foo host=node1 user=repuser' PUBLICATION pub_node1 WITH (copy_data = force, local_only = on);
+
+CREATE SUBSCRIPTION sub_node3_node2 CONNECTION 'dbname=foo host=node2 user=repuser' PUBLICATION pub_node2 WITH (copy_data = off, local_only = on);
+
+# Include truncate operations from now
+ALTER PUBLICATION pub_node3 SET (publish='insert,update,delete,truncate');
+</programlisting>
+ </para>
+ </listitem>
+ </itemizedlist>
+ </sect3>
+ </sect2>
+ </sect1>
+
</chapter>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index c5ebcf5500..cccac2132f 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -161,12 +161,14 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<variablelist>
<varlistentry>
- <term><literal>copy_data</literal> (<type>boolean</type>)</term>
+ <term><literal>copy_data</literal> (<type>enum</type>)</term>
<listitem>
<para>
Specifies whether to copy pre-existing data in the publications
- that are being subscribed to when the replication starts.
- The default is <literal>true</literal>.
+ that are being subscribed to when the replication starts. This
+ parameter may be either <literal>true</literal>,
+ <literal>false</literal> or <literal>force</literal>. The default is
+ <literal>true</literal>.
</para>
<para>
Previously subscribed tables are not copied, even if a table's row
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index c09f7b0600..907331015f 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -161,6 +161,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
publisher node changes regardless of their origin. The default is
<literal>false</literal>.
</para>
+ <para>
+ There is some interation between the "local_only" option and
+ "copy_data" option. Refer to the
+ <xref linkend="sql-createsubscription-notes" /> for details.
+ </para>
</listitem>
</varlistentry>
@@ -213,18 +218,26 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</varlistentry>
<varlistentry>
- <term><literal>copy_data</literal> (<type>boolean</type>)</term>
+ <term><literal>copy_data</literal> (<type>enum</type>)</term>
<listitem>
<para>
Specifies whether to copy pre-existing data in the publications
- that are being subscribed to when the replication starts.
- The default is <literal>true</literal>.
+ that are being subscribed to when the replication starts. This
+ parameter may be either <literal>true</literal>,
+ <literal>false</literal> or <literal>force</literal>. The default is
+ <literal>true</literal>.
</para>
<para>
If the publications contain <literal>WHERE</literal> clauses, it
will affect what data is copied. Refer to the
<xref linkend="sql-createsubscription-notes" /> for details.
</para>
+
+ <para>
+ There is some interation between the "local_only" option and
+ "copy_data" option. Refer to the
+ <xref linkend="sql-createsubscription-notes" /> for details.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 21fd805a81..11ab6d3512 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -69,6 +69,18 @@
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
+#define IS_COPY_DATA_ON_OR_FORCE(copy_data) (copy_data != COPY_DATA_OFF)
+
+/*
+ * Represents whether copy_data option is specified with on, off or force.
+ */
+typedef enum CopyData
+{
+ COPY_DATA_OFF,
+ COPY_DATA_ON,
+ COPY_DATA_FORCE
+} CopyData;
+
/*
* Structure to hold a bitmap representing the user-provided CREATE/ALTER
* SUBSCRIPTION command options and the parsed/default values of each of them.
@@ -81,7 +93,7 @@ typedef struct SubOpts
bool connect;
bool enabled;
bool create_slot;
- bool copy_data;
+ CopyData copy_data;
bool refresh;
bool binary;
bool streaming;
@@ -91,11 +103,66 @@ typedef struct SubOpts
XLogRecPtr lsn;
} SubOpts;
-static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *fetch_table_list(WalReceiverConn *wrconn, List *publications,
+ CopyData copydata, bool local_only);
static void check_duplicates_in_publist(List *publist, Datum *datums);
static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
+/*
+ * Validate the value specified for copy_data option.
+ */
+static CopyData
+DefGetCopyData(DefElem *def)
+{
+ /*
+ * If no parameter given, assume "true" is meant.
+ */
+ if (def->arg == NULL)
+ return COPY_DATA_ON;
+
+ /*
+ * Allow 0, 1, "true", "false", "on", "off" or "force".
+ */
+ switch (nodeTag(def->arg))
+ {
+ case T_Integer:
+ switch (intVal(def->arg))
+ {
+ case 0:
+ return COPY_DATA_OFF;
+ case 1:
+ return COPY_DATA_ON;
+ default:
+ /* otherwise, error out below */
+ break;
+ }
+ break;
+ default:
+ {
+ char *sval = defGetString(def);
+
+ /*
+ * The set of strings accepted here should match up with
+ * the grammar's opt_boolean_or_string production.
+ */
+ if (pg_strcasecmp(sval, "true") == 0 ||
+ pg_strcasecmp(sval, "on") == 0)
+ return COPY_DATA_ON;
+ if (pg_strcasecmp(sval, "false") == 0 ||
+ pg_strcasecmp(sval, "off") == 0)
+ return COPY_DATA_OFF;
+ if (pg_strcasecmp(sval, "force") == 0)
+ return COPY_DATA_FORCE;
+ }
+ break;
+ }
+
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s requires a boolean or \"force\"", def->defname));
+ return COPY_DATA_OFF; /* keep compiler quiet */
+}
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -128,7 +195,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
if (IsSet(supported_opts, SUBOPT_CREATE_SLOT))
opts->create_slot = true;
if (IsSet(supported_opts, SUBOPT_COPY_DATA))
- opts->copy_data = true;
+ opts->copy_data = COPY_DATA_ON;
if (IsSet(supported_opts, SUBOPT_REFRESH))
opts->refresh = true;
if (IsSet(supported_opts, SUBOPT_BINARY))
@@ -196,7 +263,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
errorConflictingDefElem(defel, pstate);
opts->specified_opts |= SUBOPT_COPY_DATA;
- opts->copy_data = defGetBoolean(defel);
+ opts->copy_data = DefGetCopyData(defel);
}
else if (IsSet(supported_opts, SUBOPT_SYNCHRONOUS_COMMIT) &&
strcmp(defel->defname, "synchronous_commit") == 0)
@@ -333,17 +400,17 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
errmsg("%s and %s are mutually exclusive options",
"connect = false", "create_slot = true")));
- if (opts->copy_data &&
+ if (IS_COPY_DATA_ON_OR_FORCE(opts->copy_data) &&
IsSet(opts->specified_opts, SUBOPT_COPY_DATA))
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("%s and %s are mutually exclusive options",
- "connect = false", "copy_data = true")));
+ "connect = false", "copy_data = true/force")));
/* Change the defaults of other options. */
opts->enabled = false;
opts->create_slot = false;
- opts->copy_data = false;
+ opts->copy_data = COPY_DATA_OFF;
}
/*
@@ -671,13 +738,14 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
* Set sync state based on if we were asked to do data copy or
* not.
*/
- table_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
+ table_state = IS_COPY_DATA_ON_OR_FORCE(opts.copy_data) ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
/*
* Get the table list from publisher and build local table status
* info.
*/
- tables = fetch_table_list(wrconn, publications);
+ tables = fetch_table_list(wrconn, publications, opts.copy_data,
+ opts.local_only);
foreach(lc, tables)
{
RangeVar *rv = (RangeVar *) lfirst(lc);
@@ -720,7 +788,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
* PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
* PUBLICATION to work.
*/
- if (opts.twophase && !opts.copy_data && tables != NIL)
+ if (opts.twophase && IS_COPY_DATA_ON_OR_FORCE(opts.copy_data)
+ && tables != NIL)
twophase_enabled = true;
walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
@@ -761,8 +830,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
}
static void
-AlterSubscription_refresh(Subscription *sub, bool copy_data,
- List *validate_publications)
+AlterSubscription_refresh(Subscription *sub, CopyData copy_data,
+ List *validate_publications, bool local_only)
{
char *err;
List *pubrel_names;
@@ -797,7 +866,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
check_publications(wrconn, validate_publications);
/* Get the table list from publisher. */
- pubrel_names = fetch_table_list(wrconn, sub->publications);
+ pubrel_names = fetch_table_list(wrconn, sub->publications, copy_data,
+ local_only);
/* Get local table list. */
subrel_states = GetSubscriptionRelations(sub->oid);
@@ -851,7 +921,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
list_length(subrel_states), sizeof(Oid), oid_cmp))
{
AddSubscriptionRelState(sub->oid, relid,
- copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
+ IS_COPY_DATA_ON_OR_FORCE(copy_data) ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
InvalidXLogRecPtr);
ereport(DEBUG1,
(errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
@@ -1134,7 +1204,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
case ALTER_SUBSCRIPTION_SET_PUBLICATION:
{
- supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
+ supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH | SUBOPT_LOCAL_ONLY;
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1157,7 +1227,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
* See ALTER_SUBSCRIPTION_REFRESH for details why this is
* not allowed.
*/
- if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && IS_COPY_DATA_ON_OR_FORCE(opts.copy_data))
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
@@ -1170,7 +1240,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
sub->publications = stmt->publication;
AlterSubscription_refresh(sub, opts.copy_data,
- stmt->publication);
+ stmt->publication,
+ opts.local_only);
}
break;
@@ -1182,7 +1253,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
List *publist;
bool isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
- supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
+ supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA | SUBOPT_LOCAL_ONLY;
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1222,7 +1293,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
sub->publications = publist;
AlterSubscription_refresh(sub, opts.copy_data,
- validate_publications);
+ validate_publications,
+ opts.local_only);
}
break;
@@ -1236,7 +1308,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions")));
parse_subscription_options(pstate, stmt->options,
- SUBOPT_COPY_DATA, &opts);
+ SUBOPT_COPY_DATA | SUBOPT_LOCAL_ONLY,
+ &opts);
/*
* The subscription option "two_phase" requires that
@@ -1255,7 +1328,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
*
* For more details see comments atop worker.c.
*/
- if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
+ if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED &&
+ IS_COPY_DATA_ON_OR_FORCE(opts.copy_data))
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
@@ -1264,7 +1338,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
- AlterSubscription_refresh(sub, opts.copy_data, NULL);
+ AlterSubscription_refresh(sub, opts.copy_data, NULL,
+ opts.local_only);
break;
}
@@ -1778,22 +1853,27 @@ AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId)
* publisher connection.
*/
static List *
-fetch_table_list(WalReceiverConn *wrconn, List *publications)
+fetch_table_list(WalReceiverConn *wrconn, List *publications, CopyData copydata,
+ bool local_only)
{
WalRcvExecResult *res;
StringInfoData cmd;
TupleTableSlot *slot;
- Oid tableRow[2] = {TEXTOID, TEXTOID};
+ Oid tableRow[3] = {TEXTOID, TEXTOID, CHAROID};
List *tablelist = NIL;
initStringInfo(&cmd);
- appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"
- " FROM pg_catalog.pg_publication_tables t\n"
- " WHERE t.pubname IN (");
+ appendStringInfoString(&cmd,
+ "SELECT DISTINCT N.nspname AS schemaname, C.relname AS tablename, PS.srrelid as replicated\n"
+ "FROM pg_publication P,\n"
+ "LATERAL pg_get_publication_tables(P.pubname) GPT\n"
+ "LEFT JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid),\n"
+ "pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
+ "WHERE C.oid = GPT.relid AND P.pubname in (");
get_publications_str(publications, &cmd, true);
appendStringInfoChar(&cmd, ')');
- res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
+ res = walrcv_exec(wrconn, cmd.data, 3, tableRow);
pfree(cmd.data);
if (res->status != WALRCV_OK_TUPLES)
@@ -1819,6 +1899,25 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
rv = makeRangeVar(nspname, relname, -1);
tablelist = lappend(tablelist, rv);
+ /*
+ * XXX: During initial table sync we cannot differentiate between the
+ * local and non-local data that is present in the HEAP. Identification
+ * of local data can be done only from the WAL by using the origin id.
+ * Throw an error so that the user can take care of the initial data
+ * copying and then create subscription with copy_data off.
+ *
+ * It is quite possible that subscriber has not yet pulled data to
+ * the tables, but in ideal cases the table data will be subscribed.
+ * To keep the code simple it is not checked if the subscriber table
+ * has pulled the data or not.
+ */
+ if (copydata == COPY_DATA_ON && local_only && !slot_attisnull(slot, 3))
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("CREATE/ALTER SUBSCRIPTION with local_only and copy_data as true is not allowed when the publisher might have replicated data, table:%s.%s might have replicated data in the publisher",
+ nspname, relname),
+ errhint("Use CREATE/ALTER SUBSCRIPTION with copy_data = off or force"));
+
ExecClearTuple(slot);
}
ExecDropSingleTupleTableSlot(slot);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 8bf7c810a5..fd8c49e05f 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -47,7 +47,13 @@ ERROR: must be superuser to create subscriptions
SET SESSION AUTHORIZATION 'regress_subscription_user';
-- fail - invalid option combinations
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
-ERROR: connect = false and copy_data = true are mutually exclusive options
+ERROR: connect = false and copy_data = true/force are mutually exclusive options
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = force);
+ERROR: connect = false and copy_data = true/force are mutually exclusive options
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = on);
+ERROR: connect = false and copy_data = true/force are mutually exclusive options
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = 1);
+ERROR: connect = false and copy_data = true/force are mutually exclusive options
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, enabled = true);
ERROR: connect = false and enabled = true are mutually exclusive options
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, create_slot = true);
@@ -93,6 +99,16 @@ ALTER SUBSCRIPTION regress_testsub4 SET (local_only = false);
DROP SUBSCRIPTION regress_testsub3;
DROP SUBSCRIPTION regress_testsub4;
+-- ok - valid coy_data options
+CREATE SUBSCRIPTION regress_testsub3 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = off);
+WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = false);
+WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
+CREATE SUBSCRIPTION regress_testsub5 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = 0);
+WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
+DROP SUBSCRIPTION regress_testsub3;
+DROP SUBSCRIPTION regress_testsub4;
+DROP SUBSCRIPTION regress_testsub5;
-- fail - invalid connection string
ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 327a1e2500..11b2c0a7c2 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -40,6 +40,9 @@ SET SESSION AUTHORIZATION 'regress_subscription_user';
-- fail - invalid option combinations
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = force);
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = on);
+CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = 1);
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, enabled = true);
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, create_slot = true);
CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, enabled = true);
@@ -66,6 +69,15 @@ ALTER SUBSCRIPTION regress_testsub4 SET (local_only = false);
DROP SUBSCRIPTION regress_testsub3;
DROP SUBSCRIPTION regress_testsub4;
+-- ok - valid coy_data options
+CREATE SUBSCRIPTION regress_testsub3 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = off);
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = false);
+CREATE SUBSCRIPTION regress_testsub5 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = 0);
+
+DROP SUBSCRIPTION regress_testsub3;
+DROP SUBSCRIPTION regress_testsub4;
+DROP SUBSCRIPTION regress_testsub5;
+
-- fail - invalid connection string
ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
diff --git a/src/test/subscription/t/032_localonly.pl b/src/test/subscription/t/032_localonly.pl
index 5c63eece1b..7f21d00894 100644
--- a/src/test/subscription/t/032_localonly.pl
+++ b/src/test/subscription/t/032_localonly.pl
@@ -8,6 +8,120 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+my $result;
+my $stdout;
+my $stderr;
+
+my $appname_A1 = 'tap_sub_A_B';
+my $appname_A2 = 'tap_sub_A_C';
+my $appname_B1 = 'tap_sub_B_A';
+my $appname_B2 = 'tap_sub_B_C';
+my $appname_C1 = 'tap_sub_C_A';
+my $appname_C2 = 'tap_sub_C_B';
+
+# Subroutine for cleaning up the subscriber contents.
+sub clean_subscriber_contents
+{
+ my ($node_A, $node_B, $node_C) = @_;
+ $node_A->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_A_C");
+ $node_B->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_B_C");
+ $node_C->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_C_A");
+ $node_C->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_C_B");
+
+ $result =
+ $node_A->safe_psql('postgres', "SELECT count(*) FROM pg_subscription");
+ is($result, qq(1), 'check subscription was dropped on subscriber');
+
+ $result =
+ $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_subscription");
+ is($result, qq(1), 'check subscription was dropped on subscriber');
+
+ $result =
+ $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_subscription");
+ is($result, qq(0), 'check subscription was dropped on subscriber');
+
+ $result = $node_A->safe_psql('postgres',
+ "SELECT count(*) FROM pg_replication_slots");
+ is($result, qq(1), 'check replication slot was dropped on publisher');
+
+ $result = $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM pg_replication_slots");
+ is($result, qq(1), 'check replication slot was dropped on publisher');
+
+ $result = $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM pg_replication_slots");
+ is($result, qq(0), 'check replication slot was dropped on publisher');
+
+ $node_A->safe_psql('postgres', "TRUNCATE tab_full");
+ $node_B->safe_psql('postgres', "TRUNCATE tab_full");
+ $node_C->safe_psql('postgres', "TRUNCATE tab_full");
+}
+
+# Subroutine for verify the data is replicated successfully.
+sub verify_data
+{
+ my ($node_A, $node_B, $node_C) = @_;
+
+ $node_A->wait_for_catchup($appname_B1);
+ $node_A->wait_for_catchup($appname_C1);
+ $node_B->wait_for_catchup($appname_A1);
+ $node_B->wait_for_catchup($appname_C2);
+ $node_C->wait_for_catchup($appname_A2);
+ $node_C->wait_for_catchup($appname_B2);
+
+ # check that data is replicated to all the nodes
+ $result =
+ $node_A->safe_psql('postgres', "SELECT * FROM tab_full order by 1;");
+ is( $result, qq(11
+12
+13
+21
+31),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+ );
+
+ $result =
+ $node_B->safe_psql('postgres', "SELECT * FROM tab_full order by 1;");
+ is( $result, qq(11
+12
+13
+21
+31),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+ );
+
+ $result =
+ $node_C->safe_psql('postgres', "SELECT * FROM tab_full order by 1;");
+ is( $result, qq(11
+12
+13
+21
+31),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+ );
+}
+
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+
+sub create_subscription
+{
+ my ($node_subscriber, $node_publisher, $sub_name, $node_connstr,
+ $application_name, $pub_name, $copy_data_val)
+ = @_;
+ $node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION $sub_name
+ CONNECTION '$node_connstr application_name=$application_name'
+ PUBLICATION $pub_name
+ WITH (copy_data = $copy_data_val, local_only = on)");
+ $node_publisher->wait_for_catchup($application_name);
+
+ # Also wait for initial table sync to finish
+ $node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+}
+
#####################################################################
# Setup a bidirectional logical replication between Node_A & Node_B
#####################################################################
@@ -43,42 +157,19 @@ $node_B->safe_psql('postgres', "CREATE TABLE tab_full (a int PRIMARY KEY)");
my $node_A_connstr = $node_A->connstr . ' dbname=postgres';
$node_A->safe_psql('postgres',
"CREATE PUBLICATION tap_pub_A FOR TABLE tab_full");
-my $appname_B1 = 'tap_sub_B1';
-$node_B->safe_psql(
- 'postgres', "
- CREATE SUBSCRIPTION tap_sub_B1
- CONNECTION '$node_A_connstr application_name=$appname_B1'
- PUBLICATION tap_pub_A
- WITH (local_only = on)");
+
+create_subscription($node_B, $node_A, $appname_B1, $node_A_connstr,
+ $appname_B1, 'tap_pub_A', 'on');
# node_B (pub) -> node_A (sub)
my $node_B_connstr = $node_B->connstr . ' dbname=postgres';
$node_B->safe_psql('postgres',
"CREATE PUBLICATION tap_pub_B FOR TABLE tab_full");
-my $appname_A = 'tap_sub_A';
-$node_A->safe_psql(
- 'postgres', "
- CREATE SUBSCRIPTION tap_sub_A
- CONNECTION '$node_B_connstr application_name=$appname_A'
- PUBLICATION tap_pub_B
- WITH (local_only = on, copy_data = off)");
-
-# Wait for subscribers to finish initialization
-$node_A->wait_for_catchup($appname_B1);
-$node_B->wait_for_catchup($appname_A);
-
-# Also wait for initial table sync to finish
-my $synced_query =
- "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
-$node_A->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
-$node_B->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+create_subscription($node_A, $node_B, $appname_A1, $node_B_connstr,
+ $appname_A1, 'tap_pub_B', 'off');
is(1, 1, "Circular replication setup is complete");
-my $result;
-
#############################################################################
# check that bidirectional logical replication setup does not cause infinite
# recursive insertion.
@@ -89,7 +180,7 @@ $node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (11);");
$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (12);");
$node_A->wait_for_catchup($appname_B1);
-$node_B->wait_for_catchup($appname_A);
+$node_B->wait_for_catchup($appname_A1);
# check that transaction was committed on subscriber(s)
$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full;");
@@ -125,24 +216,14 @@ my $node_C_connstr = $node_C->connstr . ' dbname=postgres';
$node_C->safe_psql('postgres',
"CREATE PUBLICATION tap_pub_C FOR TABLE tab_full");
-my $appname_B2 = 'tap_sub_B2';
-$node_B->safe_psql(
- 'postgres', "
- CREATE SUBSCRIPTION tap_sub_B2
- CONNECTION '$node_C_connstr application_name=$appname_B2'
- PUBLICATION tap_pub_C
- WITH (local_only = on)");
-
-$node_C->wait_for_catchup($appname_B2);
-
-$node_C->poll_query_until('postgres', $synced_query)
- or die "Timed out while waiting for subscriber to synchronize data";
+create_subscription($node_B, $node_C, $appname_B2, $node_C_connstr,
+ $appname_B2, 'tap_pub_C', 'on');
# insert a record
$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (13);");
$node_C->wait_for_catchup($appname_B2);
-$node_B->wait_for_catchup($appname_A);
+$node_B->wait_for_catchup($appname_A1);
$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full;");
is( $result, qq(11
@@ -158,6 +239,120 @@ is( $result, qq(11
'Inserted successfully without leading to infinite recursion in circular replication setup'
);
+# cleanup
+$node_B->safe_psql(
+ 'postgres', "
+ DROP SUBSCRIPTION $appname_B2");
+$node_C->safe_psql(
+ 'postgres', "
+ DELETE FROM tab_full");
+$node_B->safe_psql(
+ 'postgres', "
+ DELETE FROM tab_full where a = 13");
+
+# Error when creating susbcription with local_only and copy_data as true when
+# the publisher has replicated data
+($result, $stdout, $stderr) = $node_A->psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub_A3
+ CONNECTION '$node_B_connstr application_name=$appname_A1'
+ PUBLICATION tap_pub_B
+ WITH (local_only = on, copy_data = on)");
+like(
+ $stderr,
+ qr/ERROR: CREATE\/ALTER SUBSCRIPTION with local_only and copy_data as true is not allowed when the publisher might have replicated data/,
+ "Create subscription with local_only and copy_data having replicated table in publisher"
+);
+
+# Creating subscription with local_only and copy_data as force should be
+# successful when the publisher has replicated data
+$node_A->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub_A2
+ CONNECTION '$node_B_connstr application_name=$appname_A2'
+ PUBLICATION tap_pub_B
+ WITH (local_only = on, copy_data = force)");
+
+$node_A->safe_psql(
+ 'postgres', "
+ DROP SUBSCRIPTION tap_sub_A2");
+
+##########################################################################
+# Add 3rd node when the existing node has some data
+##########################################################################
+create_subscription($node_A, $node_C, $appname_A2, $node_C_connstr,
+ $appname_A2, 'tap_pub_C', 'off');
+create_subscription($node_B, $node_C, $appname_B2, $node_C_connstr,
+ $appname_B2, 'tap_pub_C', 'off');
+create_subscription($node_C, $node_A, $appname_C1, $node_A_connstr,
+ $appname_C1, 'tap_pub_A', 'force');
+create_subscription($node_C, $node_B, $appname_C2, $node_B_connstr,
+ $appname_C2, 'tap_pub_B', 'off');
+
+# insert some data in all the nodes
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (13);");
+$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (21);");
+$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (31);");
+
+verify_data($node_A, $node_B, $node_C);
+
+clean_subscriber_contents($node_A, $node_B, $node_C);
+
+##########################################################################
+# Add 3rd node when the existing node has no data
+##########################################################################
+create_subscription($node_A, $node_C, $appname_A2, $node_C_connstr,
+ $appname_A2, 'tap_pub_C', 'off');
+create_subscription($node_B, $node_C, $appname_B2, $node_C_connstr,
+ $appname_B2, 'tap_pub_C', 'off');
+create_subscription($node_C, $node_A, $appname_C1, $node_A_connstr,
+ $appname_C1, 'tap_pub_A', 'off');
+create_subscription($node_C, $node_B, $appname_C2, $node_B_connstr,
+ $appname_C2, 'tap_pub_B', 'off');
+
+# insert some data in all the nodes
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (11);");
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (12);");
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (13);");
+$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (21);");
+$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (31);");
+
+verify_data($node_A, $node_B, $node_C);
+
+clean_subscriber_contents($node_A, $node_B, $node_C);
+
+##########################################################################
+# Add 3rd node when the new node has some data
+##########################################################################
+create_subscription($node_A, $node_C, $appname_A2, $node_C_connstr,
+ $appname_A2, 'tap_pub_C', 'force');
+create_subscription($node_B, $node_C, $appname_B2, $node_C_connstr,
+ $appname_B2, 'tap_pub_C', 'force');
+
+$node_C->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_C SET (publish='insert,update,delete');");
+
+$node_C->safe_psql('postgres', "TRUNCATE tab_full");
+
+create_subscription($node_C, $node_A, $appname_C1, $node_A_connstr,
+ $appname_C1, 'tap_pub_A', 'force');
+create_subscription($node_C, $node_B, $appname_C2, $node_B_connstr,
+ $appname_C2, 'tap_pub_B', 'off');
+
+#include truncates now
+$node_C->safe_psql('postgres',
+ "ALTER PUBLICATION tap_pub_C SET (publish='insert,update,delete,truncate');"
+);
+
+# insert some data in all the nodes
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (11);");
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (12);");
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (13);");
+$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (21);");
+$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (31);");
+
+verify_data($node_A, $node_B, $node_C);
+
# shutdown
$node_B->stop('fast');
$node_A->stop('fast');
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 87ee7bf866..3edea910de 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -442,6 +442,7 @@ ConvProcInfo
ConversionLocation
ConvertRowtypeExpr
CookedConstraint
+CopyData
CopyDest
CopyFormatOptions
CopyFromState
--
2.32.0
From 7c4f0895ec5a34da72670245ff2ad9e004bdc3cc Mon Sep 17 00:00:00 2001
From: Vigneshwaran C <vignes...@gmail.com>
Date: Fri, 8 Apr 2022 11:10:05 +0530
Subject: [PATCH v9 1/2] Skip replication of non local data.
This patch adds a new SUBSCRIPTION boolean option
"local_only". The default is false. When a SUBSCRIPTION is
created with this option enabled, the publisher will only publish data
that originated at the publisher node.
Usage:
CREATE SUBSCRIPTION sub1 CONNECTION 'dbname=postgres port=9999'
PUBLICATION pub1 with (local_only = true);
---
doc/src/sgml/ref/alter_subscription.sgml | 5 +-
doc/src/sgml/ref/create_subscription.sgml | 12 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 4 +-
src/backend/commands/subscriptioncmds.c | 26 ++-
.../libpqwalreceiver/libpqwalreceiver.c | 5 +
src/backend/replication/logical/worker.c | 2 +
src/backend/replication/pgoutput/pgoutput.c | 22 +++
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 8 +-
src/bin/psql/tab-complete.c | 4 +-
src/include/catalog/pg_subscription.h | 3 +
src/include/replication/logicalproto.h | 10 +-
src/include/replication/pgoutput.h | 1 +
src/include/replication/walreceiver.h | 1 +
src/test/regress/expected/subscription.out | 142 ++++++++-------
src/test/regress/sql/subscription.sql | 10 ++
src/test/subscription/t/032_localonly.pl | 166 ++++++++++++++++++
19 files changed, 368 insertions(+), 72 deletions(-)
create mode 100644 src/test/subscription/t/032_localonly.pl
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 353ea5def2..c5ebcf5500 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -207,8 +207,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
information. The parameters that can be altered
are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>,
- <literal>binary</literal>, <literal>streaming</literal>, and
- <literal>disable_on_error</literal>.
+ <literal>binary</literal>, <literal>streaming</literal>,
+ <literal>disable_on_error</literal>, and
+ <literal>local_only</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 203bb41844..c09f7b0600 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -152,6 +152,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><literal>local_only</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether the subscription will request the publisher to send
+ locally originated changes at the publisher node, or send any
+ publisher node changes regardless of their origin. The default is
+ <literal>false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><literal>slot_name</literal> (<type>string</type>)</term>
<listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index add51caadf..e3d13ffb1c 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->local_only = subform->sublocalonly;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 0fc614e32c..4cc4a60005 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1290,8 +1290,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
-- All columns of pg_subscription except subconninfo are publicly readable.
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
- subbinary, substream, subtwophasestate, subdisableonerr, subslotname,
- subsynccommit, subpublications)
+ subbinary, substream, subtwophasestate, subdisableonerr,
+ sublocalonly, subslotname, subsynccommit, subpublications)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index b94236f74d..21fd805a81 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -64,6 +64,7 @@
#define SUBOPT_TWOPHASE_COMMIT 0x00000200
#define SUBOPT_DISABLE_ON_ERR 0x00000400
#define SUBOPT_LSN 0x00000800
+#define SUBOPT_LOCAL_ONLY 0x00001000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -86,6 +87,7 @@ typedef struct SubOpts
bool streaming;
bool twophase;
bool disableonerr;
+ bool local_only;
XLogRecPtr lsn;
} SubOpts;
@@ -137,6 +139,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->twophase = false;
if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
opts->disableonerr = false;
+ if (IsSet(supported_opts, SUBOPT_LOCAL_ONLY))
+ opts->local_only = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -235,6 +239,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_STREAMING;
opts->streaming = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_LOCAL_ONLY) &&
+ strcmp(defel->defname, "local_only") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_LOCAL_ONLY))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_LOCAL_ONLY;
+ opts->local_only = defGetBoolean(defel);
+ }
else if (strcmp(defel->defname, "two_phase") == 0)
{
/*
@@ -531,7 +544,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
- SUBOPT_DISABLE_ON_ERR);
+ SUBOPT_DISABLE_ON_ERR | SUBOPT_LOCAL_ONLY);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -602,6 +615,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
values[Anum_pg_subscription_substream - 1] = BoolGetDatum(opts.streaming);
+ values[Anum_pg_subscription_sublocalonly - 1] = BoolGetDatum(opts.local_only);
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
LOGICALREP_TWOPHASE_STATE_PENDING :
@@ -1015,7 +1029,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR);
+ SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
+ SUBOPT_LOCAL_ONLY);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1072,6 +1087,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
= true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_LOCAL_ONLY))
+ {
+ values[Anum_pg_subscription_sublocalonly - 1] =
+ BoolGetDatum(opts.local_only);
+ replaces[Anum_pg_subscription_sublocalonly - 1] = true;
+ }
+
update_tuple = true;
break;
}
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 0d89db4e6a..0072f5f101 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -453,6 +453,11 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
PQserverVersion(conn->streamConn) >= 150000)
appendStringInfoString(&cmd, ", two_phase 'on'");
+ /* FIXME: 150000 should be changed to 160000 later for PG16. */
+ if (options->proto.logical.local_only &&
+ PQserverVersion(conn->streamConn) >= 150000)
+ appendStringInfoString(&cmd, ", local_only 'on'");
+
pubnames = options->proto.logical.publication_names;
pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
if (!pubnames_str)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 4171371296..9ccaa255c0 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3055,6 +3055,7 @@ maybe_reread_subscription(void)
newsub->binary != MySubscription->binary ||
newsub->stream != MySubscription->stream ||
newsub->owner != MySubscription->owner ||
+ newsub->local_only != MySubscription->local_only ||
!equal(newsub->publications, MySubscription->publications))
{
ereport(LOG,
@@ -3735,6 +3736,7 @@ ApplyWorkerMain(Datum main_arg)
options.proto.logical.binary = MySubscription->binary;
options.proto.logical.streaming = MySubscription->stream;
options.proto.logical.twophase = false;
+ options.proto.logical.local_only = MySubscription->local_only;
if (!am_tablesync_worker())
{
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index b197bfd565..78b3c3bf5f 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -286,11 +286,13 @@ parse_output_parameters(List *options, PGOutputData *data)
bool messages_option_given = false;
bool streaming_given = false;
bool two_phase_option_given = false;
+ bool local_only_option_given = false;
data->binary = false;
data->streaming = false;
data->messages = false;
data->two_phase = false;
+ data->local_only = false;
foreach(lc, options)
{
@@ -379,6 +381,16 @@ parse_output_parameters(List *options, PGOutputData *data)
data->two_phase = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "local_only") == 0)
+ {
+ if (local_only_option_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ local_only_option_given = true;
+
+ data->local_only = defGetBoolean(defel);
+ }
else
elog(ERROR, "unrecognized pgoutput option: %s", defel->defname);
}
@@ -477,6 +489,12 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
else
ctx->twophase_opt_given = true;
+ if (data->local_only && data->protocol_version < LOGICALREP_PROTO_LOCALONLY_VERSION_NUM)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("requested proto_version=%d does not support local_only, need %d or higher",
+ data->protocol_version, LOGICALREP_PROTO_LOCALONLY_VERSION_NUM)));
+
/* Init publication state. */
data->publications = NIL;
publications_valid = false;
@@ -1696,6 +1714,10 @@ static bool
pgoutput_origin_filter(LogicalDecodingContext *ctx,
RepOriginId origin_id)
{
+ PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+
+ if (data->local_only && origin_id != InvalidRepOriginId)
+ return true;
return false;
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 969e2a7a46..fca5c87096 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4352,6 +4352,7 @@ getSubscriptions(Archive *fout)
int i_subsynccommit;
int i_subpublications;
int i_subbinary;
+ int i_sublocalonly;
int i,
ntups;
@@ -4396,13 +4397,19 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 150000)
appendPQExpBufferStr(query,
" s.subtwophasestate,\n"
- " s.subdisableonerr\n");
+ " s.subdisableonerr,\n");
else
appendPQExpBuffer(query,
" '%c' AS subtwophasestate,\n"
- " false AS subdisableonerr\n",
+ " false AS subdisableonerr,\n",
LOGICALREP_TWOPHASE_STATE_DISABLED);
+ /* FIXME: 150000 should be changed to 160000 later for PG16. */
+ if (fout->remoteVersion >= 150000)
+ appendPQExpBufferStr(query, " s.sublocalonly\n");
+ else
+ appendPQExpBufferStr(query, " false AS sublocalonly\n");
+
appendPQExpBufferStr(query,
"FROM pg_subscription s\n"
"WHERE s.subdbid = (SELECT oid FROM pg_database\n"
@@ -4428,6 +4435,7 @@ getSubscriptions(Archive *fout)
i_substream = PQfnumber(res, "substream");
i_subtwophasestate = PQfnumber(res, "subtwophasestate");
i_subdisableonerr = PQfnumber(res, "subdisableonerr");
+ i_sublocalonly = PQfnumber(res, "sublocalonly");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4457,6 +4465,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subtwophasestate));
subinfo[i].subdisableonerr =
pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
+ subinfo[i].sublocalonly =
+ pg_strdup(PQgetvalue(res, i, i_sublocalonly));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4530,6 +4540,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subdisableonerr, "t") == 0)
appendPQExpBufferStr(query, ", disable_on_error = true");
+ if (strcmp(subinfo->sublocalonly, "t") == 0)
+ appendPQExpBufferStr(query, ", local_only = true");
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 1d21c2906f..545b76ba5e 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -661,6 +661,7 @@ typedef struct _SubscriptionInfo
char *subdisableonerr;
char *subsynccommit;
char *subpublications;
+ char *sublocalonly;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 583817b0cc..0741588e45 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6204,7 +6204,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false};
+ false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6246,6 +6246,12 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Two phase commit"),
gettext_noop("Disable on error"));
+ /* FIXME: 150000 should be changed to 160000 later for PG16 */
+ if (pset.sversion >= 150000)
+ appendPQExpBuffer(&buf,
+ ", sublocalonly AS \"%s\"\n",
+ gettext_noop("Local only"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 588c0841fe..173e9c1a17 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1874,7 +1874,7 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "slot_name", "streaming", "synchronous_commit", "disable_on_error");
+ COMPLETE_WITH("binary", "slot_name", "streaming", "local_only", "synchronous_commit", "disable_on_error");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -3156,7 +3156,7 @@ psql_completion(const char *text, int start, int end)
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "enabled", "slot_name", "streaming",
+ "enabled", "slot_name", "streaming", "local_only",
"synchronous_commit", "two_phase", "disable_on_error");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index f006a92612..92cd82c0a3 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -70,6 +70,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool substream; /* Stream in-progress transactions. */
+ bool sublocalonly; /* skip copying of remote origin data */
+
char subtwophasestate; /* Stream two-phase transactions */
bool subdisableonerr; /* True if a worker error should cause the
@@ -112,6 +114,7 @@ typedef struct Subscription
bool binary; /* Indicates if the subscription wants data in
* binary format */
bool stream; /* Allow streaming in-progress transactions. */
+ bool local_only; /* Skip copying of remote orgin data */
char twophasestate; /* Allow streaming two-phase transactions */
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index a771ab8ff3..ed6c5de23c 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -32,12 +32,20 @@
*
* LOGICALREP_PROTO_TWOPHASE_VERSION_NUM is the minimum protocol version with
* support for two-phase commit decoding (at prepare time). Introduced in PG15.
+ *
+ * LOGICALREP_PROTO_LOCALONLY_VERSION_NUM is the minimum protocol version with
+ * support for sending only locally originated data from the publisher.
+ * Introduced in PG16.
+ *
+ * FIXME: LOGICALREP_PROTO_LOCALONLY_VERSION_NUM needs to be bumped to 4 in
+ * PG16.
*/
#define LOGICALREP_PROTO_MIN_VERSION_NUM 1
#define LOGICALREP_PROTO_VERSION_NUM 1
#define LOGICALREP_PROTO_STREAM_VERSION_NUM 2
#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM 3
-#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
+#define LOGICALREP_PROTO_LOCALONLY_VERSION_NUM 3
+#define LOGICALREP_PROTO_MAX_VERSION_NUM LOGICALREP_PROTO_LOCALONLY_VERSION_NUM
/*
* Logical message types
diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
index eafedd610a..eb7859c445 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -29,6 +29,7 @@ typedef struct PGOutputData
bool streaming;
bool messages;
bool two_phase;
+ bool local_only;
} PGOutputData;
#endif /* PGOUTPUT_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 81184aa92f..10b860f74e 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -183,6 +183,7 @@ typedef struct
bool streaming; /* Streaming of large transactions */
bool twophase; /* Streaming of two-phase transactions at
* prepare time */
+ bool local_only; /* publish only locally originated data */
} logical;
} proto;
} WalRcvStreamOptions;
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 7fcfad1591..8bf7c810a5 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -70,16 +70,38 @@ ALTER SUBSCRIPTION regress_testsub3 ENABLE;
ERROR: cannot enable subscription that does not have a slot name
ALTER SUBSCRIPTION regress_testsub3 REFRESH PUBLICATION;
ERROR: ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions
+-- fail - local_only must be boolean
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, local_only = foo);
+ERROR: local_only requires a Boolean value
+-- now it works
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, local_only = true);
+WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
+\dRs+ regress_testsub4
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | f | d | f | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub4 SET (local_only = false);
+\dRs+ regress_testsub4
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
DROP SUBSCRIPTION regress_testsub3;
+DROP SUBSCRIPTION regress_testsub4;
-- fail - invalid connection string
ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -96,10 +118,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | f | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -108,10 +130,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | f | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -143,10 +165,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | f | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -179,19 +201,19 @@ ERROR: binary requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, binary = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -202,19 +224,19 @@ ERROR: streaming requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -229,10 +251,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more then once
@@ -247,10 +269,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -284,10 +306,10 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -296,10 +318,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -308,10 +330,10 @@ DROP SUBSCRIPTION regress_testsub;
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -323,18 +345,18 @@ ERROR: disable_on_error requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = false);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | t | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Local only | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | t | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 74c38ead5d..327a1e2500 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -54,7 +54,17 @@ CREATE SUBSCRIPTION regress_testsub3 CONNECTION 'dbname=regress_doesnotexist' PU
ALTER SUBSCRIPTION regress_testsub3 ENABLE;
ALTER SUBSCRIPTION regress_testsub3 REFRESH PUBLICATION;
+-- fail - local_only must be boolean
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, local_only = foo);
+
+-- now it works
+CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, local_only = true);
+\dRs+ regress_testsub4
+ALTER SUBSCRIPTION regress_testsub4 SET (local_only = false);
+\dRs+ regress_testsub4
+
DROP SUBSCRIPTION regress_testsub3;
+DROP SUBSCRIPTION regress_testsub4;
-- fail - invalid connection string
ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
diff --git a/src/test/subscription/t/032_localonly.pl b/src/test/subscription/t/032_localonly.pl
new file mode 100644
index 0000000000..5c63eece1b
--- /dev/null
+++ b/src/test/subscription/t/032_localonly.pl
@@ -0,0 +1,166 @@
+
+# Copyright (c) 2021-2022, PostgreSQL Global Development Group
+
+# Test logical replication using local_only option.
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+#####################################################################
+# Setup a bidirectional logical replication between Node_A & Node_B
+#####################################################################
+
+# Initialize nodes
+# node_A
+my $node_A = PostgreSQL::Test::Cluster->new('node_A');
+$node_A->init(allows_streaming => 'logical');
+$node_A->append_conf(
+ 'postgresql.conf', qq(
+max_prepared_transactions = 10
+logical_decoding_work_mem = 64kB
+));
+$node_A->start;
+# node_B
+my $node_B = PostgreSQL::Test::Cluster->new('node_B');
+$node_B->init(allows_streaming => 'logical');
+$node_B->append_conf(
+ 'postgresql.conf', qq(
+max_prepared_transactions = 10
+logical_decoding_work_mem = 64kB
+));
+$node_B->start;
+
+# Create tables on node_A
+$node_A->safe_psql('postgres', "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Create the same tables on node_B
+$node_B->safe_psql('postgres', "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication
+# node_A (pub) -> node_B (sub)
+my $node_A_connstr = $node_A->connstr . ' dbname=postgres';
+$node_A->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_A FOR TABLE tab_full");
+my $appname_B1 = 'tap_sub_B1';
+$node_B->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub_B1
+ CONNECTION '$node_A_connstr application_name=$appname_B1'
+ PUBLICATION tap_pub_A
+ WITH (local_only = on)");
+
+# node_B (pub) -> node_A (sub)
+my $node_B_connstr = $node_B->connstr . ' dbname=postgres';
+$node_B->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_B FOR TABLE tab_full");
+my $appname_A = 'tap_sub_A';
+$node_A->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub_A
+ CONNECTION '$node_B_connstr application_name=$appname_A'
+ PUBLICATION tap_pub_B
+ WITH (local_only = on, copy_data = off)");
+
+# Wait for subscribers to finish initialization
+$node_A->wait_for_catchup($appname_B1);
+$node_B->wait_for_catchup($appname_A);
+
+# Also wait for initial table sync to finish
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');";
+$node_A->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+$node_B->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+is(1, 1, "Circular replication setup is complete");
+
+my $result;
+
+#############################################################################
+# check that bidirectional logical replication setup does not cause infinite
+# recursive insertion.
+#############################################################################
+
+# insert a record
+$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (11);");
+$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (12);");
+
+$node_A->wait_for_catchup($appname_B1);
+$node_B->wait_for_catchup($appname_A);
+
+# check that transaction was committed on subscriber(s)
+$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full;");
+is( $result, qq(11
+12),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+);
+$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full;");
+is( $result, qq(11
+12),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+);
+
+#############################################################################
+# check that remote data that is origined from node_C to node_B is not
+# published to node_A
+#############################################################################
+# Initialize node node_C
+my $node_C = PostgreSQL::Test::Cluster->new('node_C');
+$node_C->init(allows_streaming => 'logical');
+$node_C->append_conf(
+ 'postgresql.conf', qq(
+max_prepared_transactions = 10
+logical_decoding_work_mem = 64kB
+));
+$node_C->start;
+
+$node_C->safe_psql('postgres', "CREATE TABLE tab_full (a int PRIMARY KEY)");
+
+# Setup logical replication
+# node_C (pub) -> node_B (sub)
+my $node_C_connstr = $node_C->connstr . ' dbname=postgres';
+$node_C->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_C FOR TABLE tab_full");
+
+my $appname_B2 = 'tap_sub_B2';
+$node_B->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub_B2
+ CONNECTION '$node_C_connstr application_name=$appname_B2'
+ PUBLICATION tap_pub_C
+ WITH (local_only = on)");
+
+$node_C->wait_for_catchup($appname_B2);
+
+$node_C->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# insert a record
+$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (13);");
+
+$node_C->wait_for_catchup($appname_B2);
+$node_B->wait_for_catchup($appname_A);
+
+$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full;");
+is( $result, qq(11
+12
+13),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+);
+
+# check that the data published from node_C to node_B is not sent to node_A
+$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full;");
+is( $result, qq(11
+12),
+ 'Inserted successfully without leading to infinite recursion in circular replication setup'
+);
+
+# shutdown
+$node_B->stop('fast');
+$node_A->stop('fast');
+$node_C->stop('fast');
+
+done_testing();
--
2.32.0
