From 4e553c228b1930cef5a7825d2db045504967f972 Mon Sep 17 00:00:00 2001 From: Vigneshwaran C Date: Thu, 28 Apr 2022 15:40:50 +0530 Subject: [PATCH v12 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 pub1 for the employee table b) Node2 - SUBSCRIPTION from pub1 with local_only=true c) Node2 - PUBLICATION pub2 for the employee table d) Node1 - SUBSCRIPTION from pub2 with local_only=true 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 will have to create another subscription subscribing from Node2 to Node3 using local_only option and copy_data as true. In the scenario above the user has specified local_only 'on' (which indicates that the publisher should only replicate the changes that are generated locally), but in this case, the publisher node is also subscribing data from other nodes, so the publisher node may have remotely originated data. We throw an error, in this case, to draw attention to there being possible remote data. If the user still wishes to continue with the operation user can specify copy_data as 'force' and proceed. Handling of initial data copying in this case is detailed in the documentation section of the patch. --- doc/src/sgml/logical-replication.sgml | 275 +++++++++++++++++ doc/src/sgml/ref/alter_subscription.sgml | 16 +- doc/src/sgml/ref/create_subscription.sgml | 33 ++- src/backend/commands/subscriptioncmds.c | 137 +++++++-- src/test/regress/expected/subscription.out | 18 +- src/test/regress/sql/subscription.sql | 12 + src/test/subscription/t/032_localonly.pl | 327 ++++++++++++++++++--- src/tools/pgindent/typedefs.list | 1 + 8 files changed, 747 insertions(+), 72 deletions(-) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 145ea71d61..f163b0cced 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -1267,4 +1267,279 @@ CREATE SUBSCRIPTION mysub CONNECTION 'dbname=foo host=bar user=repuser' PUBLICAT incremental changes to those tables. + + + Bidirectional logical replication + + + Setting bidirectional replication between two nodes + + Bidirectional replication is useful in creating a multi-master database + which helps in performing read/write operations from any of the nodes. + Setting up bidirectional logical replication between two nodes requires + creation of a publication in all the nodes, creating subscriptions in + each of the nodes that subscribes to data from all the nodes. The steps + to create a two-node bidirectional replication are given below: + + + + Create a publication in node1: + +node1=# CREATE PUBLICATION pub_node1 FOR TABLE t1; +CREATE PUBLICATION + + + + Create a subscription in node2 to subscribe the + changes from node1: + +node2=# CREATE SUBSCRIPTION sub_node2_node1 +node2-# CONNECTION 'dbname=foo host=node1 user=repuser' +node2-# PUBLICATION pub_node1 +node2-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a publication in node2: + +node2=# CREATE PUBLICATION pub_node2 FOR TABLE t1; +CREATE PUBLICATION + + + + Create a subscription in node1 to subscribe the changes + from node2: + +node1=# CREATE SUBSCRIPTION sub_node1_node2 +node1-# CONNECTION 'dbname=foo host=node2 user=repuser' +node1-# PUBLICATION pub_node2 +node1-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + 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. + + + + + Adding new node when there is no data in any of the nodes + + Adding a new node node3 to the existing + node1 and node2 requires setting + up subscription in node1 and node2 + to replicate the data from node3 and setting up + subscription in node3 to replicate data from + node1 and node2. + + + + Create a publication in node3: + +node3=# CREATE PUBLICATION pub_node3 FOR TABLE t1; +CREATE PUBLICATION + + + + Create a subscription in node1 to subscribe the changes + from node3: + +node1=# CREATE SUBSCRIPTION sub_node1_node3 +node1-# CONNECTION 'dbname=foo host=node3 user=repuser' +node1-# PUBLICATION pub_node3 +node1-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node2 to subscribe the changes + from node3: + +node2=# CREATE SUBSCRIPTION sub_node2_node3 +node2-# CONNECTION 'dbname=foo host=node3 user=repuser' +node2-# PUBLICATION pub_node3 +node2-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node3 to subscribe the changes + from node1: + +node3=# CREATE SUBSCRIPTION sub_node3_node1 +node3-# CONNECTION 'dbname=foo host=node1 user=repuser' +node3-# PUBLICATION pub_node1 +node3-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node3 to subscribe the changes + from node2: + +node3=# CREATE SUBSCRIPTION sub_node3_node2 +node3-# CONNECTION 'dbname=foo host=node2 user=repuser' +node3-# PUBLICATION pub_node2 +node3-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + + Adding new node when data is present in the existing nodes + + Adding a new node node3 to the existing + node1 and node2 when data is present + in existing nodes node1 and node2 + needs similar steps. The only change required here is that + node3 should create a subscription with + copy_data = force to one of the existing nodes to + receive the existing data during initial data synchronization. + + + + Create a publication in node3: + +node3=# CREATE PUBLICATION pub_node3 FOR TABLE t1; +CREATE PUBLICATION + + + + Create a subscription in node1 to subscribe the changes + from node3: + +node1=# CREATE SUBSCRIPTION sub_node1_node3 +node1-# CONNECTION 'dbname=foo host=node3 user=repuser' +node1-# PUBLICATION pub_node3 +node1-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node2 to subscribe the changes + from node3: + +node2=# CREATE SUBSCRIPTION sub_node2_node3 +node2-# CONNECTION 'dbname=foo host=node3 user=repuser' +node2-# PUBLICATION pub_node3 +node2-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + Create a 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: + +node3=# CREATE SUBSCRIPTION sub_node3_node1 +node3-# CONNECTION 'dbname=foo host=node1 user=repuser' +node3-# PUBLICATION pub_node1 +node3-# WITH (copy_data = force, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node3 to subscribe the changes + from node2: + +node3=# CREATE SUBSCRIPTION sub_node3_node2 +node3-# CONNECTION 'dbname=foo host=node2 user=repuser' +node3-# PUBLICATION pub_node2 +node3-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + + Adding new node when data is present in the new node + + Adding a new node node3 to the existing + node1 and node2 when data is present + in the new node node3 needs similar steps. A few changes + are required here to get the existing data from node3 + to node1 and node2 and later + cleaning up of data in node3 before synchronization of + all the data from the existing nodes. + + + + Create a publication in node3: + +node3=# CREATE PUBLICATION pub_node3 FOR TABLE t1; +CREATE PUBLICATION + + + + Create a 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: + +node1=# CREATE SUBSCRIPTION sub_node1_node3 +node1-# CONNECTION 'dbname=foo host=node3 user=repuser' +node1-# PUBLICATION pub_node3 +node1-# WITH (copy_data = force, local_only = on); +CREATE SUBSCRIPTION + + + + Create a 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: + +node2=# CREATE SUBSCRIPTION sub_node2_node3 +node2-# CONNECTION 'dbname=foo host=node3 user=repuser' +node2-# PUBLICATION pub_node3 +node2-# WITH (copy_data = force, local_only = on); +CREATE SUBSCRIPTION + + + + Adjust the publication publish settings so that truncate is not published + to the subscribers and truncate the table data in node3: + +node3=# ALTER PUBLICATION pub_node3 SET (publish='insert,update,delete'); +ALTER PUBLICATION +node3=# TRUNCATE t1; +TRUNCATE TABLE +node3=# ALTER PUBLICATION pub_node3 SET (publish='insert,update,delete,truncate'); +ALTER PUBLICATION + + + + Create a subscription in node3 to subscribe the changes + from node1, here copy_data is + specified as force when creating a subscription to + node1 so that the existing table data is copied during + initial sync: + +node3=# CREATE SUBSCRIPTION +node3-# sub_node3_node1 CONNECTION 'dbname=foo host=node1 user=repuser' +node3-# PUBLICATION pub_node1 +node3-# WITH (copy_data = force, local_only = on); +CREATE SUBSCRIPTION + + + + Create a subscription in node3 to subscribe the changes + from node2, here copy_data is + specified as off as the initial table data would have + been copied in the earlier step: + +node3=# CREATE SUBSCRIPTION +node3-# sub_node3_node2 CONNECTION 'dbname=foo host=node2 user=repuser' +node3-# PUBLICATION pub_node2 +node3-# WITH (copy_data = off, local_only = on); +CREATE SUBSCRIPTION + + + + diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index c5ebcf5500..2bc6e730ee 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -161,12 +161,22 @@ ALTER SUBSCRIPTION name RENAME TO < - copy_data (boolean) + copy_data (enum) Specifies whether to copy pre-existing data in the publications - that are being subscribed to when the replication starts. - The default is true. + that are being subscribed to when the replication starts. This + parameter may be either true, + false or force. The default is + true. + + + There is some interaction between the local_only + option and copy_data option. Refer to the + CREATE SUBSCRIPTION + for interaction + details and usage of force for + copy_data option. 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..dd733a8c1e 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -161,6 +161,11 @@ CREATE SUBSCRIPTION subscription_namefalse. + + There is some interaction between the local_only + option and copy_data option. Refer to the + for details. + @@ -213,18 +218,28 @@ CREATE SUBSCRIPTION subscription_name - copy_data (boolean) + copy_data (enum) Specifies whether to copy pre-existing data in the publications - that are being subscribed to when the replication starts. - The default is true. + that are being subscribed to when the replication starts. This + parameter may be either true, + false or force. The default is + true. If the publications contain WHERE clauses, it will affect what data is copied. Refer to the for details. + + + There is some interaction between the local_only + option and copy_data option. Refer to the + for interaction + details and usage of force for + copy_data option. + @@ -374,6 +389,18 @@ CREATE SUBSCRIPTION subscription_name + + If subscription is created with local_only = on and + copy_data = on, it will check if the publisher tables are + being subscribed to any other publisher and throw an error to prevent + inconsistent data in the subscription. The user can continue with the copy + operation without throwing any error in this case by specifying + copy_data = force. Refer to the + on how + copy_data and local_only can be used + in bidirectional replication. + + diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 21fd805a81..c80463c1cc 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 off, on 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, "false") == 0 || + pg_strcasecmp(sval, "off") == 0) + return COPY_DATA_OFF; + if (pg_strcasecmp(sval, "true") == 0 || + pg_strcasecmp(sval, "on") == 0) + return COPY_DATA_ON; + 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 && opts.copy_data == COPY_DATA_OFF && + tables != NIL) twophase_enabled = true; walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled, @@ -761,7 +830,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, } static void -AlterSubscription_refresh(Subscription *sub, bool copy_data, +AlterSubscription_refresh(Subscription *sub, CopyData copy_data, List *validate_publications) { char *err; @@ -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, + sub->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\"", @@ -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"), @@ -1255,7 +1325,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"), @@ -1778,22 +1849,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 +1895,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..91ccbc7ebe 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 = 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, 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, 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 copy_data options +CREATE SUBSCRIPTION regress_testsub3 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_testsub4 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_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..33722edc1b 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 = 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, copy_data = force); 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 copy_data options +CREATE SUBSCRIPTION regress_testsub3 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = false); +CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (slot_name = NONE, connect = false, copy_data = off); +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 40b9c626ac..96f6727d61 100644 --- a/src/test/subscription/t/032_localonly.pl +++ b/src/test/subscription/t/032_localonly.pl @@ -8,6 +8,116 @@ use PostgreSQL::Test::Cluster; use PostgreSQL::Test::Utils; use Test::More; +my $result; +my $stdout; +my $stderr; + +my $subname_AB = 'tap_sub_A_B'; +my $subname_AC = 'tap_sub_A_C'; +my $subname_BA = 'tap_sub_B_A'; +my $subname_BC = 'tap_sub_B_C'; +my $subname_CA = 'tap_sub_C_A'; +my $subname_CB = 'tap_sub_C_B'; + +# Detach node C from the node-group of (A, B, C) and clean the table contents +# from all nodes. +sub detach_node_clean_table_data +{ + 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 to verify the data is replicated successfully. +sub verify_data +{ + my ($node_A, $node_B, $node_C, $expect) = @_; + + $node_A->wait_for_catchup($subname_BA); + $node_A->wait_for_catchup($subname_CA); + $node_B->wait_for_catchup($subname_AB); + $node_B->wait_for_catchup($subname_CB); + $node_C->wait_for_catchup($subname_AC); + $node_C->wait_for_catchup($subname_BC); + + # 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($expect), + 'Data is replicated as expected' + ); + + $result = + $node_B->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); + is($result, qq($expect), + 'Data is replicated as expected' + ); + + $result = + $node_C->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); + is($result, qq($expect), + 'Data is replicated as expected' + ); +} + +my $synced_query = + "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r', 's');"; + +# Subroutine to create subscription and wait till the initial sync is completed. +# Subroutine expects subscriber node, publisher node, subscription name, +# destination connection string, publication name and the subscription with +# options to be passed as input parameters. +sub create_subscription +{ + my ($node_subscriber, $node_publisher, $sub_name, $node_connstr, + $pub_name, $with_options) + = @_; + + # Application_name is always assigned the same value as the subscription + # name. + $node_subscriber->safe_psql( + 'postgres', " + CREATE SUBSCRIPTION $sub_name + CONNECTION '$node_connstr application_name=$sub_name' + PUBLICATION $pub_name + WITH ($with_options)"); + $node_publisher->wait_for_catchup($sub_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 +153,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, $subname_BA, $node_A_connstr, + 'tap_pub_A', 'copy_data = on, 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"; +create_subscription($node_A, $node_B, $subname_AB, $node_B_connstr, + 'tap_pub_B', 'copy_data = off, local_only = on'); is(1, 1, "Circular replication setup is complete"); -my $result; - ############################################################################### # check that bidirectional logical replication setup does not cause infinite # recursive insertion. @@ -88,8 +175,8 @@ my $result; $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_A->wait_for_catchup($subname_BA); +$node_B->wait_for_catchup($subname_AB); # check that transaction was committed on subscriber(s) $result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); @@ -123,24 +210,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, $subname_BC, $node_C_connstr, + 'tap_pub_C', 'copy_data = on, local_only = 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_C->wait_for_catchup($subname_BC); +$node_B->wait_for_catchup($subname_AB); $result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); is($result, qq(11 @@ -154,6 +231,168 @@ is($result, qq(11 12), 'Remote data originated from other node is not replicated when local_only option is ON' ); +# clear the operations done by this test +$node_B->safe_psql( + 'postgres', " + DROP SUBSCRIPTION $subname_BC"); +$node_C->safe_psql( + 'postgres', " + DELETE FROM tab_full"); +$node_B->safe_psql( + 'postgres', " + DELETE FROM tab_full where a = 13"); + +############################################################################### +# Specifying local_only 'on' which indicates that the publisher should only +# replicate the changes that are generated locally from node_B, but in +# this case since the node_B is also subscribing data from node_A, node_B can +# have remotely originated data from node_A. We throw an error, in this case, +# to draw attention to there being possible remote data. +############################################################################### +($result, $stdout, $stderr) = $node_A->psql( + 'postgres', " + CREATE SUBSCRIPTION tap_sub_A2 + CONNECTION '$node_B_connstr application_name=$subname_AB' + 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=$subname_AC' + PUBLICATION tap_pub_B + WITH (local_only = on, copy_data = force)"); + +$node_A->safe_psql( + 'postgres', " + DROP SUBSCRIPTION tap_sub_A2"); + +############################################################################### +# Join 3rd node (node_C) to the existing 2 nodes(node_A & node_B) bidirectional +# replication setup when the existing nodes (node_A & node_B) has pre-existing +# data and the new node (node_C) does not have any data. +############################################################################### +$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(11 +12), 'Check existing data'); + +$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(11 +12), 'Check existing data'); + +$result = + $node_C->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +create_subscription($node_A, $node_C, $subname_AC, $node_C_connstr, + 'tap_pub_C', 'copy_data = off, local_only = on'); +create_subscription($node_B, $node_C, $subname_BC, $node_C_connstr, + 'tap_pub_C', 'copy_data = off, local_only = on'); +create_subscription($node_C, $node_A, $subname_CA, $node_A_connstr, + 'tap_pub_A', 'copy_data = force, local_only = on'); +create_subscription($node_C, $node_B, $subname_CB, $node_B_connstr, + 'tap_pub_B', 'copy_data = off, local_only = on'); + +# 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 (23);"); +$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (33);"); + +verify_data($node_A, $node_B, $node_C, '11 +12 +13 +23 +33'); + +detach_node_clean_table_data($node_A, $node_B, $node_C); + +############################################################################### +# Join 3rd node (node_C) to the existing 2 nodes(node_A & node_B) bidirectional +# replication setup when the existing nodes (node_A & node_B) and the new node +# (node_C) does not have any data. +############################################################################### +$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +$result = $node_C->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +create_subscription($node_A, $node_C, $subname_AC, $node_C_connstr, + 'tap_pub_C', 'copy_data = off, local_only = on'); +create_subscription($node_B, $node_C, $subname_BC, $node_C_connstr, + 'tap_pub_C', 'copy_data = off, local_only = on'); +create_subscription($node_C, $node_A, $subname_CA, $node_A_connstr, + 'tap_pub_A', 'copy_data = off, local_only = on'); +create_subscription($node_C, $node_B, $subname_CB, $node_B_connstr, + 'tap_pub_B', 'copy_data = off, local_only = on'); + +# insert some data in all the nodes +$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (11);"); +$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, '11 +21 +31'); + +detach_node_clean_table_data($node_A, $node_B, $node_C); + +############################################################################### +# Join 3rd node (node_C) to the existing 2 nodes(node_A & node_B) bidirectional +# replication setup when the existing nodes (node_A & node_B) has no data and +# the new node (node_C) some pre-existing data. +############################################################################### +$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (31);"); + +$result = $node_A->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +$result = $node_B->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is( $result, qq(), 'Check existing data'); + +$result = $node_C->safe_psql('postgres', "SELECT * FROM tab_full ORDER BY 1;"); +is($result, qq(31), 'Check existing data'); + +create_subscription($node_A, $node_C, $subname_AC, $node_C_connstr, + 'tap_pub_C', 'copy_data = on, local_only = on'); +create_subscription($node_B, $node_C, $subname_BC, $node_C_connstr, + 'tap_pub_C', 'copy_data = on, local_only = on'); + +$node_C->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_C SET (publish='insert,update,delete');"); + +$node_C->safe_psql('postgres', "TRUNCATE tab_full"); + +# include truncates now +$node_C->safe_psql('postgres', + "ALTER PUBLICATION tap_pub_C SET (publish='insert,update,delete,truncate');" +); + +create_subscription($node_C, $node_A, $subname_CA, $node_A_connstr, + 'tap_pub_A', 'copy_data = force, local_only = on'); +create_subscription($node_C, $node_B, $subname_CB, $node_B_connstr, + 'tap_pub_B', 'copy_data = off, local_only = on'); + +# insert some data in all the nodes +$node_A->safe_psql('postgres', "INSERT INTO tab_full VALUES (12);"); +$node_B->safe_psql('postgres', "INSERT INTO tab_full VALUES (22);"); +$node_C->safe_psql('postgres', "INSERT INTO tab_full VALUES (32);"); + +verify_data($node_A, $node_B, $node_C, '12 +22 +31 +32'); + # 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