diff options
| author | Amit Kapila <akapila@postgresql.org> | 2023-09-27 14:32:51 +0530 |
|---|---|---|
| committer | Amit Kapila <akapila@postgresql.org> | 2023-09-27 14:32:51 +0530 |
| commit | 54ccfd65868c013a8c6906bc894bc5ea3640740a (patch) | |
| tree | a36b9678a875324f596ff97af123e383d597ddde /src | |
| parent | 6fc3a138b1e8049a73e48d4b0f1031a1e6e7b8c6 (diff) | |
| download | postgresql-54ccfd65868c013a8c6906bc894bc5ea3640740a.tar.gz postgresql-54ccfd65868c013a8c6906bc894bc5ea3640740a.zip | |
Fix the misuse of origin filter across multiple pg_logical_slot_get_changes() calls.
The pgoutput module uses a global variable (publish_no_origin) to cache
the action for the origin filter, but we didn't reset the flag when
shutting down the output plugin, so subsequent retries may access the
previous publish_no_origin value.
We fix this by storing the flag in the output plugin's private data.
Additionally, the patch removes the currently unused origin string from the
structure.
For the back branch, to avoid changing the exposed structure, we eliminated the
global variable and instead directly used the origin string for change
filtering.
Author: Hou Zhijie
Reviewed-by: Amit Kapila, Michael Paquier
Backpatch-through: 16
Discussion: http://postgr.es/m/OS0PR01MB571690EF24F51F51EFFCBB0E94FAA@OS0PR01MB5716.jpnprd01.prod.outlook.com
Diffstat (limited to 'src')
| -rw-r--r-- | src/backend/replication/pgoutput/pgoutput.c | 19 | ||||
| -rw-r--r-- | src/include/replication/pgoutput.h | 2 |
2 files changed, 12 insertions, 9 deletions
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 3d2becb45cf..251ba46da5e 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -82,7 +82,6 @@ static void pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx, static bool publications_valid; static bool in_streaming; -static bool publish_no_origin; static List *LoadPublications(List *pubnames); static void publication_invalidation_cb(Datum arg, int cacheid, @@ -381,21 +380,23 @@ parse_output_parameters(List *options, PGOutputData *data) } else if (strcmp(defel->defname, "origin") == 0) { + char *origin; + if (origin_option_given) ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options")); origin_option_given = true; - data->origin = defGetString(defel); - if (pg_strcasecmp(data->origin, LOGICALREP_ORIGIN_NONE) == 0) - publish_no_origin = true; - else if (pg_strcasecmp(data->origin, LOGICALREP_ORIGIN_ANY) == 0) - publish_no_origin = false; + origin = defGetString(defel); + if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_NONE) == 0) + data->publish_no_origin = true; + else if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_ANY) == 0) + data->publish_no_origin = false; else ereport(ERROR, errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("unrecognized origin value: \"%s\"", data->origin)); + errmsg("unrecognized origin value: \"%s\"", origin)); } else elog(ERROR, "unrecognized pgoutput option: %s", defel->defname); @@ -1673,7 +1674,9 @@ static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId origin_id) { - if (publish_no_origin && origin_id != InvalidRepOriginId) + PGOutputData *data = (PGOutputData *) ctx->output_plugin_private; + + if (data->publish_no_origin && origin_id != InvalidRepOriginId) return true; return false; diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h index b4a8015403b..b3f9a016293 100644 --- a/src/include/replication/pgoutput.h +++ b/src/include/replication/pgoutput.h @@ -29,7 +29,7 @@ typedef struct PGOutputData char streaming; bool messages; bool two_phase; - char *origin; + bool publish_no_origin; } PGOutputData; #endif /* PGOUTPUT_H */ |