Re: Handle infinite recursion in logical replication setup

On Mon, Mar 7, 2022 at 2:28 PM Peter Smith <smithpb2250(at)gmail(dot)com> wrote:
>
> Hi Vignesh,
>
> Here are some review comments for patch v2.
>
> ======
>
> 1. Question about syntax
>
> I already posted some questions about why the syntax is on the CREATE
> SUBSCRCRIBER side.
> IMO "local_only" is a publisher option, so it seemed more natural to
> me for it to be specified as a "publish" option.
>
> Ref [1] my original question + suggestion for Option 2
> Ref [2] some other examples of subscribing to multiple-publishers
>
> Anyway, +1 to see what other people think.
>

I feel we can support it in the subscriber side first and then extend
it to the publisher side as being discussed in [1]. I have retained it
as it is.

> ~~~
>
> 2. ALTER
>
> (related also to the question about syntax)
>
> If subscribing to multiple publications then ALTER is going to change
> the 'local_only' for all of them, which might not be what you want
> (??)
>

I feel we can support it in the subscriber side first and then extend
it to the publisher side as being discussed in [1]. When it is
extended to the publisher side, it will get handled. I have retained
it as it is.

> ~~~
>
> 3. subscription_parameter
>
> (related also to the question about syntax)
>
> CREATE SUBSCRIPTION subscription_name
> CONNECTION 'conninfo'
> PUBLICATION publication_name [, ...]
> [ WITH ( subscription_parameter [= value] [, ... ] ) ]
>
> ~
>
> That WITH is for *subscription* options, not the publication options.
>
> So IMO 'local_only' intuitively seems like "local" means local where
> the subscriber is.
>
> So, if the Option 1 syntax is chosen (see comment #1) then I think the
> option name maybe should change to be something more like
> 'publish_local_only' or something similar to be more clear what local
> actually means.
>

Changed it to publish_local_only

> ~~~
>
> 4. contrib/test_decoding/test_decoding.c
>
> @@ -484,6 +487,16 @@ pg_decode_filter(LogicalDecodingContext *ctx,
> return false;
> }
>
> +static bool
> +pg_decode_filter_remotedata(LogicalDecodingContext *ctx,
> + RepOriginId origin_id)
> +{
> + TestDecodingData *data = ctx->output_plugin_private;
> +
> + if (data->only_local && origin_id != InvalidRepOriginId)
> + return true;
> + return false;
> +}
>
> 4a. Maybe needs function comment.

Modified

> 4b. Missing blank line following this function
>

Modified

> ~~~
>
> 5. General - please check all of the patch.
>
> There seems inconsistency with the member names, local variable names,
> parameter names etc. There are all variations of:
>
> - only_local
> - onlylocaldata
> - onlylocal_data
> - etc
>
> Please try using the same name everywhere for everything if possible.
>

I have changed it to only_local wherever possible.

> ~~~
>
> 6. src/backend/replication/logical/decode.c - FilterRemoteOriginData
>
> @@ -585,7 +594,8 @@ logicalmsg_decode(LogicalDecodingContext *ctx,
> XLogRecordBuffer *buf)
> message = (xl_logical_message *) XLogRecGetData(r);
>
> if (message->dbId != ctx->slot->data.database ||
> - FilterByOrigin(ctx, origin_id))
> + FilterByOrigin(ctx, origin_id) ||
> + FilterRemoteOriginData(ctx, origin_id))
> return;
>
> I noticed that every call to FilterRemoteOriginData has an associated
> preceding call to FilterByOrigin. It might be worth just combining the
> logic into FilterByOrigin. Then none of that calling code (9 x places)
> would need to change at all.

Modified

> ~~~
>
> 7. src/backend/replication/logical/logical.c - CreateInitDecodingContext
>
> @@ -451,6 +453,8 @@ CreateInitDecodingContext(const char *plugin,
> */
> ctx->twophase &= slot->data.two_phase;
>
> + ctx->onlylocal_data &= slot->data.onlylocal_data;
>
> The equivalent 'twophase' option had a big comment. Probably this new
> option should also have a similar comment?

These change is not required anymore, the comment no more applies. I
have not made any change for this.

> ~~~
>
> 8. src/backend/replication/logical/logical.c - filter_remotedata_cb_wrapper
>
> +bool
> +filter_remotedata_cb_wrapper(LogicalDecodingContext *ctx,
> + RepOriginId origin_id)
> +{
> + LogicalErrorCallbackState state;
> + ErrorContextCallback errcallback;
> + bool ret;
> +
> + Assert(!ctx->fast_forward);
> +
> + /* Push callback + info on the error context stack */
> + state.ctx = ctx;
> + state.callback_name = "filter_remoteorigin";
>
> There is no consistency between the function and the name:
>
> "filter_remoteorigin" versus filter_remotedata_cb.
>
> A similar inconsistency for this is elsewhere. See review comment #9

Modified it to filter_remote_origin_cb_wrapper and changed to
*_remotedata_* to *_remote_origin_*

> ~~~
>
> 9. src/backend/replication/pgoutput/pgoutput.c
>
> @@ -215,6 +217,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
> cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
> cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn;
> cb->filter_by_origin_cb = pgoutput_origin_filter;
> + cb->filter_remotedata_cb = pgoutput_remoteorigin_filter;
>
> Inconsistent names for the member and function.
>
> filter_remotedata_cb VS pgoutput_remoteorigin_filter.

Modified it to filter_remote_origin_cb and changed to *_remotedata_*
to *_remote_origin_*

> ~~~
>
> 10. src/backend/replication/pgoutput/pgoutput.c
>
> @@ -1450,6 +1465,16 @@ pgoutput_origin_filter(LogicalDecodingContext *ctx,
> return false;
> }
>
> +static bool
> +pgoutput_remoteorigin_filter(LogicalDecodingContext *ctx,
> + RepOriginId origin_id)
> +{
> + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
> +
> + if (data->onlylocal_data && origin_id != InvalidRepOriginId)
> + return true;
> + return false;
> +}
> /*
> * Shutdown the output plugin.
> *
>
> 10a. Add a function comment.

Modified

> 10b. Missing blank line after the function

Modified

> ~~~
>
> 11. src/backend/replication/slotfuncs.c - pg_create_logical_replication_slot
>
> @@ -171,6 +174,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
> Name plugin = PG_GETARG_NAME(1);
> bool temporary = PG_GETARG_BOOL(2);
> bool two_phase = PG_GETARG_BOOL(3);
> + bool onlylocal_data = PG_GETARG_BOOL(4);
> Datum result;
> TupleDesc tupdesc;
> HeapTuple tuple;
>
>
> Won't there be some PG Docs needing to be updated now there is another
> parameter?

This change is not required anymore, the comment no longer applies. I
have not made any changes for this.

> ~~~
>
> 12. src/include/catalog/pg_proc.dat - pg_get_replication_slots
>
> I did not see any update for pg_get_replication_slots, but you added
> the 4th parameter elsewhere. Is something missing here?

This change is not required anymore, the comment no longer applies. I
have not made any changes for this.

> ~~~
>
> 13. src/include/replication/logical.h
>
> @@ -99,6 +99,8 @@ typedef struct LogicalDecodingContext
> */
> bool twophase_opt_given;
>
> + bool onlylocal_data;
> +
>
> I think the new member needs some comment.

Modified

> ~~~
>
> 14. 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 onlylocal_data;
> } logical;
> } proto;
> } WalRcvStreamOptions;
>
> I think the new member needs some comment.

Modified

> ~~~
>
> 15. src/test/regress/sql/subscription.sql
>
> ALTER SUBSCRIPTION test missing?

Added

Thanks for the comments, the attached v3 patch has the fixes for the same.
[1] - https://www.postgresql.org/message-id/CAA4eK1%2BMtz%2BStvNNtTg9%3D9BTq8%3DpMu-V5i4yWqs%3DKJUh0Z_L4g%40mail.gmail.com

Regards,
Vignesh