On Mon, Jul 4, 2022 at 3:48 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> On Sun, Jul 3, 2022 at 8:29 PM vignesh C <vignesh21@gmail.com> wrote:
> >
>
> Review comments
> ===============
> 1.
> @@ -530,7 +557,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_ORIGIN);
> parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
>
> /*
> @@ -606,6 +633,8 @@ CreateSubscription(ParseState *pstate,
> CreateSubscriptionStmt *stmt,
> LOGICALREP_TWOPHASE_STATE_PENDING :
> LOGICALREP_TWOPHASE_STATE_DISABLED);
> values[Anum_pg_subscription_subdisableonerr - 1] =
> BoolGetDatum(opts.disableonerr);
> + values[Anum_pg_subscription_suborigin - 1] =
> + CStringGetTextDatum(opts.origin);
> values[Anum_pg_subscription_subconninfo - 1] =
> CStringGetTextDatum(conninfo);
> if (opts.slot_name)
> ...
> ...
>
> /* List of publications subscribed to */
> text subpublications[1] BKI_FORCE_NOT_NULL;
> +
> + /* Only publish data originating from the specified origin */
> + text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY);
> #endif
> } FormData_pg_subscription;
>
> The order of declaration and assignment for 'suborigin' should match
> in above usage.
Modified
> 2. Similarly the changes in GetSubscription() should also match the
> declaration of the origin column.
Modified
> 3.
> GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
> - subbinary, substream, subtwophasestate,
> subdisableonerr, subslotname,
> - subsynccommit, subpublications)
> + subbinary, substream, subtwophasestate, subdisableonerr,
> + suborigin, subslotname, subsynccommit, subpublications)
> ON pg_subscription TO public;
>
> This should also match the order of columns as in pg_subscription.h
> unless there is a reason for not doing so.
Modified
> 4.
> + /*
> + * Even though "origin" parameter allows only "local" and "any"
> + * values, it is implemented as a string type so that the parameter
> + * can be extended in future versions to support filtering using
> + * origin names specified by the user.
>
> /Even though "origin" .../Even though the "origin" parameter ...
Modified
> 5.
> +
> +# 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)");
>
> In both the above comments, you should use table instead of tables as
> the test creates only one table.
Modified
> 6.
> +$node_A->safe_psql('postgres', "DELETE FROM tab_full;");
>
> After this, the test didn't ensure that this operation is replicated.
> Can't that lead to unpredictable results for the other tests after
> this test?
Modified to include wait_for_catchup and verified the data at the
beginning of the next test
> 7.
> +# 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 (origin = local)");
> +
> +$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";
>
> This test allows the publisher (node_C) to poll for sync but it should
> be the subscriber (node_B) that needs to poll to allow the initial
> sync to finish.
Modified
> 8. Do you think it makes sense to see if this new option can also be
> supported by pg_recvlogical? I see that previously we have not
> extended pg_recvlogical for all the newly added options but I feel we
> should keep pg_recvlogical up to date w.r.t new options. We can do
> this as a separate patch if we agree?
I will analyze this and post my analysis soon.
Thanks for the comments, the v28 patch attached has the changes for the same.
Regards,
Vignesh