Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * pgoutput.c
4 : : * Logical Replication output plugin
5 : : *
6 : : * Copyright (c) 2012-2026, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * src/backend/replication/pgoutput/pgoutput.c
10 : : *
11 : : *-------------------------------------------------------------------------
12 : : */
13 : : #include "postgres.h"
14 : :
15 : : #include "access/tupconvert.h"
16 : : #include "catalog/partition.h"
17 : : #include "catalog/pg_publication.h"
18 : : #include "catalog/pg_publication_rel.h"
19 : : #include "catalog/pg_subscription.h"
20 : : #include "commands/defrem.h"
21 : : #include "commands/subscriptioncmds.h"
22 : : #include "executor/executor.h"
23 : : #include "fmgr.h"
24 : : #include "nodes/makefuncs.h"
25 : : #include "parser/parse_relation.h"
26 : : #include "replication/logical.h"
27 : : #include "replication/logicalproto.h"
28 : : #include "replication/origin.h"
29 : : #include "replication/pgoutput.h"
30 : : #include "rewrite/rewriteHandler.h"
31 : : #include "utils/builtins.h"
32 : : #include "utils/inval.h"
33 : : #include "utils/lsyscache.h"
34 : : #include "utils/memutils.h"
35 : : #include "utils/rel.h"
36 : : #include "utils/syscache.h"
37 : : #include "utils/varlena.h"
38 : :
39 : 621 : PG_MODULE_MAGIC_EXT(
40 : : .name = "pgoutput",
41 : : .version = PG_VERSION
42 : : );
43 : :
44 : : static void pgoutput_startup(LogicalDecodingContext *ctx,
45 : : OutputPluginOptions *opt, bool is_init);
46 : : static void pgoutput_shutdown(LogicalDecodingContext *ctx);
47 : : static void pgoutput_begin_txn(LogicalDecodingContext *ctx,
48 : : ReorderBufferTXN *txn);
49 : : static void pgoutput_commit_txn(LogicalDecodingContext *ctx,
50 : : ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
51 : : static void pgoutput_change(LogicalDecodingContext *ctx,
52 : : ReorderBufferTXN *txn, Relation relation,
53 : : ReorderBufferChange *change);
54 : : static void pgoutput_truncate(LogicalDecodingContext *ctx,
55 : : ReorderBufferTXN *txn, int nrelations, Relation relations[],
56 : : ReorderBufferChange *change);
57 : : static void pgoutput_message(LogicalDecodingContext *ctx,
58 : : ReorderBufferTXN *txn, XLogRecPtr message_lsn,
59 : : bool transactional, const char *prefix,
60 : : Size sz, const char *message);
61 : : static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
62 : : ReplOriginId origin_id);
63 : : static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
64 : : ReorderBufferTXN *txn);
65 : : static void pgoutput_prepare_txn(LogicalDecodingContext *ctx,
66 : : ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);
67 : : static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx,
68 : : ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
69 : : static void pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
70 : : ReorderBufferTXN *txn,
71 : : XLogRecPtr prepare_end_lsn,
72 : : TimestampTz prepare_time);
73 : : static void pgoutput_stream_start(struct LogicalDecodingContext *ctx,
74 : : ReorderBufferTXN *txn);
75 : : static void pgoutput_stream_stop(struct LogicalDecodingContext *ctx,
76 : : ReorderBufferTXN *txn);
77 : : static void pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
78 : : ReorderBufferTXN *txn,
79 : : XLogRecPtr abort_lsn);
80 : : static void pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
81 : : ReorderBufferTXN *txn,
82 : : XLogRecPtr commit_lsn);
83 : : static void pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
84 : : ReorderBufferTXN *txn, XLogRecPtr prepare_lsn);
85 : :
86 : : static bool publications_valid;
87 : :
88 : : static List *LoadPublications(List *pubnames);
89 : : static void publication_invalidation_cb(Datum arg, SysCacheIdentifier cacheid,
90 : : uint32 hashvalue);
91 : : static void send_repl_origin(LogicalDecodingContext *ctx,
92 : : ReplOriginId origin_id, XLogRecPtr origin_lsn,
93 : : bool send_origin);
94 : :
95 : : /*
96 : : * Only 3 publication actions are used for row filtering ("insert", "update",
97 : : * "delete"). See RelationSyncEntry.exprstate[].
98 : : */
99 : : enum RowFilterPubAction
100 : : {
101 : : PUBACTION_INSERT,
102 : : PUBACTION_UPDATE,
103 : : PUBACTION_DELETE,
104 : : };
105 : :
106 : : #define NUM_ROWFILTER_PUBACTIONS (PUBACTION_DELETE+1)
107 : :
108 : : /*
109 : : * Entry in the map used to remember which relation schemas we sent.
110 : : *
111 : : * The schema_sent flag determines if the current schema record for the
112 : : * relation (and for its ancestor if publish_as_relid is set) was already
113 : : * sent to the subscriber (in which case we don't need to send it again).
114 : : *
115 : : * The schema cache on downstream is however updated only at commit time,
116 : : * and with streamed transactions the commit order may be different from
117 : : * the order the transactions are sent in. Also, the (sub) transactions
118 : : * might get aborted so we need to send the schema for each (sub) transaction
119 : : * so that we don't lose the schema information on abort. For handling this,
120 : : * we maintain the list of xids (streamed_txns) for those we have already sent
121 : : * the schema.
122 : : *
123 : : * For partitions, 'pubactions' considers not only the table's own
124 : : * publications, but also those of all of its ancestors.
125 : : */
126 : : typedef struct RelationSyncEntry
127 : : {
128 : : Oid relid; /* relation oid */
129 : :
130 : : bool replicate_valid; /* overall validity flag for entry */
131 : :
132 : : bool schema_sent;
133 : :
134 : : /*
135 : : * This will be PUBLISH_GENCOLS_STORED if the relation contains generated
136 : : * columns and the 'publish_generated_columns' parameter is set to
137 : : * PUBLISH_GENCOLS_STORED. Otherwise, it will be PUBLISH_GENCOLS_NONE,
138 : : * indicating that no generated columns should be published, unless
139 : : * explicitly specified in the column list.
140 : : */
141 : : PublishGencolsType include_gencols_type;
142 : : List *streamed_txns; /* streamed toplevel transactions with this
143 : : * schema */
144 : :
145 : : /* are we publishing this rel? */
146 : : PublicationActions pubactions;
147 : :
148 : : /*
149 : : * ExprState array for row filter. Different publication actions don't
150 : : * allow multiple expressions to always be combined into one, because
151 : : * updates or deletes restrict the column in expression to be part of the
152 : : * replica identity index whereas inserts do not have this restriction, so
153 : : * there is one ExprState per publication action.
154 : : */
155 : : ExprState *exprstate[NUM_ROWFILTER_PUBACTIONS];
156 : : EState *estate; /* executor state used for row filter */
157 : : TupleTableSlot *new_slot; /* slot for storing new tuple */
158 : : TupleTableSlot *old_slot; /* slot for storing old tuple */
159 : :
160 : : /*
161 : : * OID of the relation to publish changes as. For a partition, this may
162 : : * be set to one of its ancestors whose schema will be used when
163 : : * replicating changes, if publish_via_partition_root is set for the
164 : : * publication.
165 : : */
166 : : Oid publish_as_relid;
167 : :
168 : : /*
169 : : * Map used when replicating using an ancestor's schema to convert tuples
170 : : * from partition's type to the ancestor's; NULL if publish_as_relid is
171 : : * same as 'relid' or if unnecessary due to partition and the ancestor
172 : : * having identical TupleDesc.
173 : : */
174 : : AttrMap *attrmap;
175 : :
176 : : /*
177 : : * Columns included in the publication, or NULL if all columns are
178 : : * included implicitly. Note that the attnums in this bitmap are not
179 : : * shifted by FirstLowInvalidHeapAttributeNumber.
180 : : */
181 : : Bitmapset *columns;
182 : :
183 : : /*
184 : : * Private context to store additional data for this entry - state for the
185 : : * row filter expressions, column list, etc.
186 : : */
187 : : MemoryContext entry_cxt;
188 : : } RelationSyncEntry;
189 : :
190 : : /*
191 : : * Maintain a per-transaction level variable to track whether the transaction
192 : : * has sent BEGIN. BEGIN is only sent when the first change in a transaction
193 : : * is processed. This makes it possible to skip sending a pair of BEGIN/COMMIT
194 : : * messages for empty transactions which saves network bandwidth.
195 : : *
196 : : * This optimization is not used for prepared transactions because if the
197 : : * WALSender restarts after prepare of a transaction and before commit prepared
198 : : * of the same transaction then we won't be able to figure out if we have
199 : : * skipped sending BEGIN/PREPARE of a transaction as it was empty. This is
200 : : * because we would have lost the in-memory txndata information that was
201 : : * present prior to the restart. This will result in sending a spurious
202 : : * COMMIT PREPARED without a corresponding prepared transaction at the
203 : : * downstream which would lead to an error when it tries to process it.
204 : : *
205 : : * XXX We could achieve this optimization by changing protocol to send
206 : : * additional information so that downstream can detect that the corresponding
207 : : * prepare has not been sent. However, adding such a check for every
208 : : * transaction in the downstream could be costly so we might want to do it
209 : : * optionally.
210 : : *
211 : : * We also don't have this optimization for streamed transactions because
212 : : * they can contain prepared transactions.
213 : : */
214 : : typedef struct PGOutputTxnData
215 : : {
216 : : bool sent_begin_txn; /* flag indicating whether BEGIN has been sent */
217 : : } PGOutputTxnData;
218 : :
219 : : /* Map used to remember which relation schemas we sent. */
220 : : static HTAB *RelationSyncCache = NULL;
221 : :
222 : : static void init_rel_sync_cache(MemoryContext cachectx);
223 : : static void cleanup_rel_sync_cache(TransactionId xid, bool is_commit);
224 : : static RelationSyncEntry *get_rel_sync_entry(PGOutputData *data,
225 : : Relation relation);
226 : : static void send_relation_and_attrs(Relation relation, TransactionId xid,
227 : : LogicalDecodingContext *ctx,
228 : : RelationSyncEntry *relentry);
229 : : static void rel_sync_cache_relation_cb(Datum arg, Oid relid);
230 : : static void rel_sync_cache_publication_cb(Datum arg, SysCacheIdentifier cacheid,
231 : : uint32 hashvalue);
232 : : static void set_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
233 : : TransactionId xid);
234 : : static bool get_schema_sent_in_streamed_txn(RelationSyncEntry *entry,
235 : : TransactionId xid);
236 : : static void init_tuple_slot(PGOutputData *data, Relation relation,
237 : : RelationSyncEntry *entry);
238 : : static void pgoutput_memory_context_reset(void *arg);
239 : :
240 : : /* row filter routines */
241 : : static EState *create_estate_for_relation(Relation rel);
242 : : static void pgoutput_row_filter_init(PGOutputData *data,
243 : : List *publications,
244 : : RelationSyncEntry *entry);
245 : : static bool pgoutput_row_filter_exec_expr(ExprState *state,
246 : : ExprContext *econtext);
247 : : static bool pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
248 : : TupleTableSlot **new_slot_ptr,
249 : : RelationSyncEntry *entry,
250 : : ReorderBufferChangeType *action);
251 : :
252 : : /* column list routines */
253 : : static void pgoutput_column_list_init(PGOutputData *data,
254 : : List *publications,
255 : : RelationSyncEntry *entry);
256 : :
257 : : /*
258 : : * Specify output plugin callbacks
259 : : */
260 : : void
261 : 826 : _PG_output_plugin_init(OutputPluginCallbacks *cb)
262 : : {
263 : 826 : cb->startup_cb = pgoutput_startup;
264 : 826 : cb->begin_cb = pgoutput_begin_txn;
265 : 826 : cb->change_cb = pgoutput_change;
266 : 826 : cb->truncate_cb = pgoutput_truncate;
267 : 826 : cb->message_cb = pgoutput_message;
268 : 826 : cb->commit_cb = pgoutput_commit_txn;
269 : :
270 : 826 : cb->begin_prepare_cb = pgoutput_begin_prepare_txn;
271 : 826 : cb->prepare_cb = pgoutput_prepare_txn;
272 : 826 : cb->commit_prepared_cb = pgoutput_commit_prepared_txn;
273 : 826 : cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn;
274 : 826 : cb->filter_by_origin_cb = pgoutput_origin_filter;
275 : 826 : cb->shutdown_cb = pgoutput_shutdown;
276 : :
277 : : /* transaction streaming */
278 : 826 : cb->stream_start_cb = pgoutput_stream_start;
279 : 826 : cb->stream_stop_cb = pgoutput_stream_stop;
280 : 826 : cb->stream_abort_cb = pgoutput_stream_abort;
281 : 826 : cb->stream_commit_cb = pgoutput_stream_commit;
282 : 826 : cb->stream_change_cb = pgoutput_change;
283 : 826 : cb->stream_message_cb = pgoutput_message;
284 : 826 : cb->stream_truncate_cb = pgoutput_truncate;
285 : : /* transaction streaming - two-phase commit */
286 : 826 : cb->stream_prepare_cb = pgoutput_stream_prepare_txn;
287 : 826 : }
288 : :
289 : : static void
290 : 467 : parse_output_parameters(List *options, PGOutputData *data)
291 : : {
292 : : ListCell *lc;
293 : 467 : bool protocol_version_given = false;
294 : 467 : bool publication_names_given = false;
295 : 467 : bool binary_option_given = false;
296 : 467 : bool messages_option_given = false;
297 : 467 : bool streaming_given = false;
298 : 467 : bool two_phase_option_given = false;
299 : 467 : bool origin_option_given = false;
300 : :
301 : : /* Initialize optional parameters to defaults */
302 : 467 : data->binary = false;
303 : 467 : data->streaming = LOGICALREP_STREAM_OFF;
304 : 467 : data->messages = false;
305 : 467 : data->two_phase = false;
306 : 467 : data->publish_no_origin = false;
307 : :
308 [ + - + + : 2319 : foreach(lc, options)
+ + ]
309 : : {
310 : 1853 : DefElem *defel = (DefElem *) lfirst(lc);
311 : :
312 : : Assert(defel->arg == NULL || IsA(defel->arg, String));
313 : :
314 : : /* Check each param, whether or not we recognize it */
315 [ + + ]: 1853 : if (strcmp(defel->defname, "proto_version") == 0)
316 : : {
317 : : unsigned long parsed;
318 : : char *endptr;
319 : 467 : const char *val = strVal(defel->arg);
320 : :
321 [ - + ]: 467 : if (protocol_version_given)
322 [ # # ]: 0 : ereport(ERROR,
323 : : (errcode(ERRCODE_SYNTAX_ERROR),
324 : : errmsg("conflicting or redundant options")));
325 : 467 : protocol_version_given = true;
326 : :
327 : 467 : errno = 0;
328 : 467 : parsed = strtoul(val, &endptr, 10);
329 [ + + + - : 467 : if (endptr == val || errno != 0 || *endptr != '\0')
- + ]
330 [ + - ]: 1 : ereport(ERROR,
331 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
332 : : errmsg("invalid proto_version")));
333 : :
334 [ - + ]: 466 : if (parsed > PG_UINT32_MAX)
335 [ # # ]: 0 : ereport(ERROR,
336 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
337 : : errmsg("proto_version \"%s\" out of range",
338 : : strVal(defel->arg))));
339 : :
340 : 466 : data->protocol_version = (uint32) parsed;
341 : : }
342 [ + + ]: 1386 : else if (strcmp(defel->defname, "publication_names") == 0)
343 : : {
344 [ - + ]: 466 : if (publication_names_given)
345 [ # # ]: 0 : ereport(ERROR,
346 : : (errcode(ERRCODE_SYNTAX_ERROR),
347 : : errmsg("conflicting or redundant options")));
348 : 466 : publication_names_given = true;
349 : :
350 : : /*
351 : : * Pass a copy of the DefElem->arg since SplitIdentifierString
352 : : * modifies its input.
353 : : */
354 [ - + ]: 466 : if (!SplitIdentifierString(pstrdup(strVal(defel->arg)), ',',
355 : : &data->publication_names))
356 [ # # ]: 0 : ereport(ERROR,
357 : : (errcode(ERRCODE_INVALID_NAME),
358 : : errmsg("invalid publication_names syntax")));
359 : : }
360 [ + + ]: 920 : else if (strcmp(defel->defname, "binary") == 0)
361 : : {
362 [ - + ]: 11 : if (binary_option_given)
363 [ # # ]: 0 : ereport(ERROR,
364 : : (errcode(ERRCODE_SYNTAX_ERROR),
365 : : errmsg("conflicting or redundant options")));
366 : 11 : binary_option_given = true;
367 : :
368 : 11 : data->binary = defGetBoolean(defel);
369 : : }
370 [ + + ]: 909 : else if (strcmp(defel->defname, "messages") == 0)
371 : : {
372 [ - + ]: 4 : if (messages_option_given)
373 [ # # ]: 0 : ereport(ERROR,
374 : : (errcode(ERRCODE_SYNTAX_ERROR),
375 : : errmsg("conflicting or redundant options")));
376 : 4 : messages_option_given = true;
377 : :
378 : 4 : data->messages = defGetBoolean(defel);
379 : : }
380 [ + + ]: 905 : else if (strcmp(defel->defname, "streaming") == 0)
381 : : {
382 [ - + ]: 444 : if (streaming_given)
383 [ # # ]: 0 : ereport(ERROR,
384 : : (errcode(ERRCODE_SYNTAX_ERROR),
385 : : errmsg("conflicting or redundant options")));
386 : 444 : streaming_given = true;
387 : :
388 : 444 : data->streaming = defGetStreamingMode(defel);
389 : : }
390 [ + + ]: 461 : else if (strcmp(defel->defname, "two_phase") == 0)
391 : : {
392 [ - + ]: 8 : if (two_phase_option_given)
393 [ # # ]: 0 : ereport(ERROR,
394 : : (errcode(ERRCODE_SYNTAX_ERROR),
395 : : errmsg("conflicting or redundant options")));
396 : 8 : two_phase_option_given = true;
397 : :
398 : 8 : data->two_phase = defGetBoolean(defel);
399 : : }
400 [ + - ]: 453 : else if (strcmp(defel->defname, "origin") == 0)
401 : : {
402 : : char *origin;
403 : :
404 [ - + ]: 453 : if (origin_option_given)
405 [ # # ]: 0 : ereport(ERROR,
406 : : errcode(ERRCODE_SYNTAX_ERROR),
407 : : errmsg("conflicting or redundant options"));
408 : 453 : origin_option_given = true;
409 : :
410 : 453 : origin = defGetString(defel);
411 [ + + ]: 453 : if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_NONE) == 0)
412 : 29 : data->publish_no_origin = true;
413 [ + - ]: 424 : else if (pg_strcasecmp(origin, LOGICALREP_ORIGIN_ANY) == 0)
414 : 424 : data->publish_no_origin = false;
415 : : else
416 [ # # ]: 0 : ereport(ERROR,
417 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
418 : : errmsg("unrecognized origin value: \"%s\"", origin));
419 : : }
420 : : else
421 [ # # ]: 0 : elog(ERROR, "unrecognized pgoutput option: %s", defel->defname);
422 : : }
423 : :
424 : : /* Check required options */
425 [ - + ]: 466 : if (!protocol_version_given)
426 [ # # ]: 0 : ereport(ERROR,
427 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
428 : : errmsg("option \"%s\" missing", "proto_version"));
429 [ - + ]: 466 : if (!publication_names_given)
430 [ # # ]: 0 : ereport(ERROR,
431 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
432 : : errmsg("option \"%s\" missing", "publication_names"));
433 : 466 : }
434 : :
435 : : /*
436 : : * Memory context reset callback of PGOutputData->context.
437 : : */
438 : : static void
439 : 1146 : pgoutput_memory_context_reset(void *arg)
440 : : {
441 [ + + ]: 1146 : if (RelationSyncCache)
442 : : {
443 : 214 : hash_destroy(RelationSyncCache);
444 : 214 : RelationSyncCache = NULL;
445 : : }
446 : 1146 : }
447 : :
448 : : /*
449 : : * Initialize this plugin
450 : : */
451 : : static void
452 : 826 : pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
453 : : bool is_init)
454 : : {
455 : 826 : PGOutputData *data = palloc0_object(PGOutputData);
456 : : static bool publication_callback_registered = false;
457 : : MemoryContextCallback *mcallback;
458 : :
459 : : /* Create our memory context for private allocations. */
460 : 826 : data->context = AllocSetContextCreate(ctx->context,
461 : : "logical replication output context",
462 : : ALLOCSET_DEFAULT_SIZES);
463 : :
464 : 826 : data->cachectx = AllocSetContextCreate(ctx->context,
465 : : "logical replication cache context",
466 : : ALLOCSET_DEFAULT_SIZES);
467 : :
468 : 826 : data->pubctx = AllocSetContextCreate(ctx->context,
469 : : "logical replication publication list context",
470 : : ALLOCSET_SMALL_SIZES);
471 : :
472 : : /*
473 : : * Ensure to cleanup RelationSyncCache even when logical decoding invoked
474 : : * via SQL interface ends up with an error.
475 : : */
476 : 826 : mcallback = palloc0_object(MemoryContextCallback);
477 : 826 : mcallback->func = pgoutput_memory_context_reset;
478 : 826 : MemoryContextRegisterResetCallback(ctx->context, mcallback);
479 : :
480 : 826 : ctx->output_plugin_private = data;
481 : :
482 : : /* This plugin uses binary protocol. */
483 : 826 : opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
484 : :
485 : : /*
486 : : * This is replication start and not slot initialization.
487 : : *
488 : : * Parse and validate options passed by the client.
489 : : */
490 [ + + ]: 826 : if (!is_init)
491 : : {
492 : : /* Parse the params and ERROR if we see any we don't recognize */
493 : 467 : parse_output_parameters(ctx->output_plugin_options, data);
494 : :
495 : : /* Check if we support requested protocol */
496 [ - + ]: 466 : if (data->protocol_version > LOGICALREP_PROTO_MAX_VERSION_NUM)
497 [ # # ]: 0 : ereport(ERROR,
498 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
499 : : errmsg("client sent proto_version=%d but server only supports protocol %d or lower",
500 : : data->protocol_version, LOGICALREP_PROTO_MAX_VERSION_NUM)));
501 : :
502 [ - + ]: 466 : if (data->protocol_version < LOGICALREP_PROTO_MIN_VERSION_NUM)
503 [ # # ]: 0 : ereport(ERROR,
504 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
505 : : errmsg("client sent proto_version=%d but server only supports protocol %d or higher",
506 : : data->protocol_version, LOGICALREP_PROTO_MIN_VERSION_NUM)));
507 : :
508 : : /*
509 : : * Decide whether to enable streaming. It is disabled by default, in
510 : : * which case we just update the flag in decoding context. Otherwise
511 : : * we only allow it with sufficient version of the protocol, and when
512 : : * the output plugin supports it.
513 : : */
514 [ + + ]: 466 : if (data->streaming == LOGICALREP_STREAM_OFF)
515 : 22 : ctx->streaming = false;
516 [ + + ]: 444 : else if (data->streaming == LOGICALREP_STREAM_ON &&
517 [ - + ]: 27 : data->protocol_version < LOGICALREP_PROTO_STREAM_VERSION_NUM)
518 [ # # ]: 0 : ereport(ERROR,
519 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
520 : : errmsg("requested proto_version=%d does not support streaming, need %d or higher",
521 : : data->protocol_version, LOGICALREP_PROTO_STREAM_VERSION_NUM)));
522 [ + + ]: 444 : else if (data->streaming == LOGICALREP_STREAM_PARALLEL &&
523 [ - + ]: 417 : data->protocol_version < LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM)
524 [ # # ]: 0 : ereport(ERROR,
525 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
526 : : errmsg("requested proto_version=%d does not support parallel streaming, need %d or higher",
527 : : data->protocol_version, LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM)));
528 [ - + ]: 444 : else if (!ctx->streaming)
529 [ # # ]: 0 : ereport(ERROR,
530 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
531 : : errmsg("streaming requested, but not supported by output plugin")));
532 : :
533 : : /*
534 : : * Here, we just check whether the two-phase option is passed by
535 : : * plugin and decide whether to enable it at later point of time. It
536 : : * remains enabled if the previous start-up has done so. But we only
537 : : * allow the option to be passed in with sufficient version of the
538 : : * protocol, and when the output plugin supports it.
539 : : */
540 [ + + ]: 466 : if (!data->two_phase)
541 : 458 : ctx->twophase_opt_given = false;
542 [ - + ]: 8 : else if (data->protocol_version < LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)
543 [ # # ]: 0 : ereport(ERROR,
544 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
545 : : errmsg("requested proto_version=%d does not support two-phase commit, need %d or higher",
546 : : data->protocol_version, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM)));
547 [ - + ]: 8 : else if (!ctx->twophase)
548 [ # # ]: 0 : ereport(ERROR,
549 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
550 : : errmsg("two-phase commit requested, but not supported by output plugin")));
551 : : else
552 : 8 : ctx->twophase_opt_given = true;
553 : :
554 : : /* Init publication state. */
555 : 466 : data->publications = NIL;
556 : 466 : publications_valid = false;
557 : :
558 : : /*
559 : : * Register callback for pg_publication if we didn't already do that
560 : : * during some previous call in this process.
561 : : */
562 [ + + ]: 466 : if (!publication_callback_registered)
563 : : {
564 : 464 : CacheRegisterSyscacheCallback(PUBLICATIONOID,
565 : : publication_invalidation_cb,
566 : : (Datum) 0);
567 : 464 : CacheRegisterRelSyncCallback(rel_sync_cache_relation_cb,
568 : : (Datum) 0);
569 : 464 : publication_callback_registered = true;
570 : : }
571 : :
572 : : /* Initialize relation schema cache. */
573 : 466 : init_rel_sync_cache(CacheMemoryContext);
574 : : }
575 : : else
576 : : {
577 : : /*
578 : : * Disable the streaming and prepared transactions during the slot
579 : : * initialization mode.
580 : : */
581 : 359 : ctx->streaming = false;
582 : 359 : ctx->twophase = false;
583 : : }
584 : 825 : }
585 : :
586 : : /*
587 : : * BEGIN callback.
588 : : *
589 : : * Don't send the BEGIN message here instead postpone it until the first
590 : : * change. In logical replication, a common scenario is to replicate a set of
591 : : * tables (instead of all tables) and transactions whose changes were on
592 : : * the table(s) that are not published will produce empty transactions. These
593 : : * empty transactions will send BEGIN and COMMIT messages to subscribers,
594 : : * using bandwidth on something with little/no use for logical replication.
595 : : */
596 : : static void
597 : 1400 : pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
598 : : {
599 : 1400 : PGOutputTxnData *txndata = MemoryContextAllocZero(ctx->context,
600 : : sizeof(PGOutputTxnData));
601 : :
602 : 1400 : txn->output_plugin_private = txndata;
603 : 1400 : }
604 : :
605 : : /*
606 : : * Send BEGIN.
607 : : *
608 : : * This is called while processing the first change of the transaction.
609 : : */
610 : : static void
611 : 515 : pgoutput_send_begin(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
612 : : {
613 : 515 : bool send_replication_origin = txn->origin_id != InvalidReplOriginId;
614 : 515 : PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
615 : :
616 : : Assert(txndata);
617 : : Assert(!txndata->sent_begin_txn);
618 : :
619 : 515 : OutputPluginPrepareWrite(ctx, !send_replication_origin);
620 : 515 : logicalrep_write_begin(ctx->out, txn);
621 : 515 : txndata->sent_begin_txn = true;
622 : :
623 : 515 : send_repl_origin(ctx, txn->origin_id, txn->origin_lsn,
624 : : send_replication_origin);
625 : :
626 : 515 : OutputPluginWrite(ctx, true);
627 : 515 : }
628 : :
629 : : /*
630 : : * COMMIT callback
631 : : */
632 : : static void
633 : 1397 : pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
634 : : XLogRecPtr commit_lsn)
635 : : {
636 : 1397 : PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
637 : : bool sent_begin_txn;
638 : :
639 : : Assert(txndata);
640 : :
641 : : /*
642 : : * We don't need to send the commit message unless some relevant change
643 : : * from this transaction has been sent to the downstream.
644 : : */
645 : 1397 : sent_begin_txn = txndata->sent_begin_txn;
646 : 1397 : OutputPluginUpdateProgress(ctx, !sent_begin_txn);
647 : 1397 : pfree(txndata);
648 : 1397 : txn->output_plugin_private = NULL;
649 : :
650 [ + + ]: 1397 : if (!sent_begin_txn)
651 : : {
652 [ + + ]: 883 : elog(DEBUG1, "skipped replication of an empty transaction with XID: %u", txn->xid);
653 : 883 : return;
654 : : }
655 : :
656 : 514 : OutputPluginPrepareWrite(ctx, true);
657 : 514 : logicalrep_write_commit(ctx->out, txn, commit_lsn);
658 : 514 : OutputPluginWrite(ctx, true);
659 : : }
660 : :
661 : : /*
662 : : * BEGIN PREPARE callback
663 : : */
664 : : static void
665 : 20 : pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
666 : : {
667 : 20 : bool send_replication_origin = txn->origin_id != InvalidReplOriginId;
668 : :
669 : 20 : OutputPluginPrepareWrite(ctx, !send_replication_origin);
670 : 20 : logicalrep_write_begin_prepare(ctx->out, txn);
671 : :
672 : 20 : send_repl_origin(ctx, txn->origin_id, txn->origin_lsn,
673 : : send_replication_origin);
674 : :
675 : 20 : OutputPluginWrite(ctx, true);
676 : 20 : }
677 : :
678 : : /*
679 : : * PREPARE callback
680 : : */
681 : : static void
682 : 20 : pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
683 : : XLogRecPtr prepare_lsn)
684 : : {
685 : 20 : OutputPluginUpdateProgress(ctx, false);
686 : :
687 : 20 : OutputPluginPrepareWrite(ctx, true);
688 : 20 : logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
689 : 20 : OutputPluginWrite(ctx, true);
690 : 20 : }
691 : :
692 : : /*
693 : : * COMMIT PREPARED callback
694 : : */
695 : : static void
696 : 25 : pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
697 : : XLogRecPtr commit_lsn)
698 : : {
699 : 25 : OutputPluginUpdateProgress(ctx, false);
700 : :
701 : 25 : OutputPluginPrepareWrite(ctx, true);
702 : 25 : logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn);
703 : 25 : OutputPluginWrite(ctx, true);
704 : 25 : }
705 : :
706 : : /*
707 : : * ROLLBACK PREPARED callback
708 : : */
709 : : static void
710 : 7 : pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx,
711 : : ReorderBufferTXN *txn,
712 : : XLogRecPtr prepare_end_lsn,
713 : : TimestampTz prepare_time)
714 : : {
715 : 7 : OutputPluginUpdateProgress(ctx, false);
716 : :
717 : 7 : OutputPluginPrepareWrite(ctx, true);
718 : 7 : logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn,
719 : : prepare_time);
720 : 7 : OutputPluginWrite(ctx, true);
721 : 7 : }
722 : :
723 : : /*
724 : : * Write the current schema of the relation and its ancestor (if any) if not
725 : : * done yet.
726 : : */
727 : : static void
728 : 202724 : maybe_send_schema(LogicalDecodingContext *ctx,
729 : : ReorderBufferChange *change,
730 : : Relation relation, RelationSyncEntry *relentry)
731 : : {
732 : 202724 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
733 : : bool schema_sent;
734 : 202724 : TransactionId xid = InvalidTransactionId;
735 : 202724 : TransactionId topxid = InvalidTransactionId;
736 : :
737 : : /*
738 : : * Remember XID of the (sub)transaction for the change. We don't care if
739 : : * it's top-level transaction or not (we have already sent that XID in
740 : : * start of the current streaming block).
741 : : *
742 : : * If we're not in a streaming block, just use InvalidTransactionId and
743 : : * the write methods will not include it.
744 : : */
745 [ + + ]: 202724 : if (data->in_streaming)
746 : 175923 : xid = change->txn->xid;
747 : :
748 [ + + ]: 202724 : if (rbtxn_is_subtxn(change->txn))
749 [ + - ]: 10169 : topxid = rbtxn_get_toptxn(change->txn)->xid;
750 : : else
751 : 192555 : topxid = xid;
752 : :
753 : : /*
754 : : * Do we need to send the schema? We do track streamed transactions
755 : : * separately, because those may be applied later (and the regular
756 : : * transactions won't see their effects until then) and in an order that
757 : : * we don't know at this point.
758 : : *
759 : : * XXX There is a scope of optimization here. Currently, we always send
760 : : * the schema first time in a streaming transaction but we can probably
761 : : * avoid that by checking 'relentry->schema_sent' flag. However, before
762 : : * doing that we need to study its impact on the case where we have a mix
763 : : * of streaming and non-streaming transactions.
764 : : */
765 [ + + ]: 202724 : if (data->in_streaming)
766 : 175923 : schema_sent = get_schema_sent_in_streamed_txn(relentry, topxid);
767 : : else
768 : 26801 : schema_sent = relentry->schema_sent;
769 : :
770 : : /* Nothing to do if we already sent the schema. */
771 [ + + ]: 202724 : if (schema_sent)
772 : 202333 : return;
773 : :
774 : : /*
775 : : * Send the schema. If the changes will be published using an ancestor's
776 : : * schema, not the relation's own, send that ancestor's schema before
777 : : * sending relation's own (XXX - maybe sending only the former suffices?).
778 : : */
779 [ + + ]: 391 : if (relentry->publish_as_relid != RelationGetRelid(relation))
780 : : {
781 : 39 : Relation ancestor = RelationIdGetRelation(relentry->publish_as_relid);
782 : :
783 : 39 : send_relation_and_attrs(ancestor, xid, ctx, relentry);
784 : 39 : RelationClose(ancestor);
785 : : }
786 : :
787 : 391 : send_relation_and_attrs(relation, xid, ctx, relentry);
788 : :
789 [ + + ]: 391 : if (data->in_streaming)
790 : 70 : set_schema_sent_in_streamed_txn(relentry, topxid);
791 : : else
792 : 321 : relentry->schema_sent = true;
793 : : }
794 : :
795 : : /*
796 : : * Sends a relation
797 : : */
798 : : static void
799 : 430 : send_relation_and_attrs(Relation relation, TransactionId xid,
800 : : LogicalDecodingContext *ctx,
801 : : RelationSyncEntry *relentry)
802 : : {
803 : 430 : TupleDesc desc = RelationGetDescr(relation);
804 : 430 : Bitmapset *columns = relentry->columns;
805 : 430 : PublishGencolsType include_gencols_type = relentry->include_gencols_type;
806 : : int i;
807 : :
808 : : /*
809 : : * Write out type info if needed. We do that only for user-created types.
810 : : * We use FirstGenbkiObjectId as the cutoff, so that we only consider
811 : : * objects with hand-assigned OIDs to be "built in", not for instance any
812 : : * function or type defined in the information_schema. This is important
813 : : * because only hand-assigned OIDs can be expected to remain stable across
814 : : * major versions.
815 : : */
816 [ + + ]: 1321 : for (i = 0; i < desc->natts; i++)
817 : : {
818 : 891 : Form_pg_attribute att = TupleDescAttr(desc, i);
819 : :
820 [ + + ]: 891 : if (!logicalrep_should_publish_column(att, columns,
821 : : include_gencols_type))
822 : 71 : continue;
823 : :
824 [ + + ]: 820 : if (att->atttypid < FirstGenbkiObjectId)
825 : 802 : continue;
826 : :
827 : 18 : OutputPluginPrepareWrite(ctx, false);
828 : 18 : logicalrep_write_typ(ctx->out, xid, att->atttypid);
829 : 18 : OutputPluginWrite(ctx, false);
830 : : }
831 : :
832 : 430 : OutputPluginPrepareWrite(ctx, false);
833 : 430 : logicalrep_write_rel(ctx->out, xid, relation, columns,
834 : : include_gencols_type);
835 : 430 : OutputPluginWrite(ctx, false);
836 : 430 : }
837 : :
838 : : /*
839 : : * Executor state preparation for evaluation of row filter expressions for the
840 : : * specified relation.
841 : : */
842 : : static EState *
843 : 18 : create_estate_for_relation(Relation rel)
844 : : {
845 : : EState *estate;
846 : : RangeTblEntry *rte;
847 : 18 : List *perminfos = NIL;
848 : :
849 : 18 : estate = CreateExecutorState();
850 : :
851 : 18 : rte = makeNode(RangeTblEntry);
852 : 18 : rte->rtekind = RTE_RELATION;
853 : 18 : rte->relid = RelationGetRelid(rel);
854 : 18 : rte->relkind = rel->rd_rel->relkind;
855 : 18 : rte->rellockmode = AccessShareLock;
856 : :
857 : 18 : addRTEPermissionInfo(&perminfos, rte);
858 : :
859 : 18 : ExecInitRangeTable(estate, list_make1(rte), perminfos,
860 : : bms_make_singleton(1));
861 : :
862 : 18 : estate->es_output_cid = GetCurrentCommandId(false);
863 : :
864 : 18 : return estate;
865 : : }
866 : :
867 : : /*
868 : : * Evaluates row filter.
869 : : *
870 : : * If the row filter evaluates to NULL, it is taken as false i.e. the change
871 : : * isn't replicated.
872 : : */
873 : : static bool
874 : 39 : pgoutput_row_filter_exec_expr(ExprState *state, ExprContext *econtext)
875 : : {
876 : : Datum ret;
877 : : bool isnull;
878 : :
879 : : Assert(state != NULL);
880 : :
881 : 39 : ret = ExecEvalExprSwitchContext(state, econtext, &isnull);
882 : :
883 [ - + - - : 38 : elog(DEBUG3, "row filter evaluates to %s (isnull: %s)",
- - - - ]
884 : : isnull ? "false" : DatumGetBool(ret) ? "true" : "false",
885 : : isnull ? "true" : "false");
886 : :
887 [ + + ]: 38 : if (isnull)
888 : 1 : return false;
889 : :
890 : 37 : return DatumGetBool(ret);
891 : : }
892 : :
893 : : /*
894 : : * Make sure the per-entry memory context exists.
895 : : */
896 : : static void
897 : 375 : pgoutput_ensure_entry_cxt(PGOutputData *data, RelationSyncEntry *entry)
898 : : {
899 : : Relation relation;
900 : :
901 : : /* The context may already exist, in which case bail out. */
902 [ + + ]: 375 : if (entry->entry_cxt)
903 : 18 : return;
904 : :
905 : 357 : relation = RelationIdGetRelation(entry->publish_as_relid);
906 : :
907 : 357 : entry->entry_cxt = AllocSetContextCreate(data->cachectx,
908 : : "entry private context",
909 : : ALLOCSET_SMALL_SIZES);
910 : :
911 : 357 : MemoryContextCopyAndSetIdentifier(entry->entry_cxt,
912 : : RelationGetRelationName(relation));
913 : : }
914 : :
915 : : /*
916 : : * Initialize the row filter.
917 : : */
918 : : static void
919 : 357 : pgoutput_row_filter_init(PGOutputData *data, List *publications,
920 : : RelationSyncEntry *entry)
921 : : {
922 : : ListCell *lc;
923 : 357 : List *rfnodes[] = {NIL, NIL, NIL}; /* One per pubaction */
924 : 357 : bool no_filter[] = {false, false, false}; /* One per pubaction */
925 : : MemoryContext oldctx;
926 : : int idx;
927 : 357 : bool has_filter = true;
928 : 357 : Oid schemaid = get_rel_namespace(entry->publish_as_relid);
929 : :
930 : : /*
931 : : * Find if there are any row filters for this relation. If there are, then
932 : : * prepare the necessary ExprState and cache it in entry->exprstate. To
933 : : * build an expression state, we need to ensure the following:
934 : : *
935 : : * All the given publication-table mappings must be checked.
936 : : *
937 : : * Multiple publications might have multiple row filters for this
938 : : * relation. Since row filter usage depends on the DML operation, there
939 : : * are multiple lists (one for each operation) to which row filters will
940 : : * be appended.
941 : : *
942 : : * FOR ALL TABLES and FOR TABLES IN SCHEMA implies "don't use row filter
943 : : * expression" so it takes precedence.
944 : : */
945 [ + - + + : 379 : foreach(lc, publications)
+ + ]
946 : : {
947 : 361 : Publication *pub = lfirst(lc);
948 : 361 : HeapTuple rftuple = NULL;
949 : 361 : Datum rfdatum = 0;
950 : 361 : bool pub_no_filter = true;
951 : :
952 : : /*
953 : : * If the publication is FOR ALL TABLES, or the publication includes a
954 : : * FOR TABLES IN SCHEMA where the table belongs to the referred
955 : : * schema, then it is treated the same as if there are no row filters
956 : : * (even if other publications have a row filter).
957 : : */
958 [ + + ]: 361 : if (!pub->alltables &&
959 [ + + ]: 265 : !SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
960 : : ObjectIdGetDatum(schemaid),
961 : : ObjectIdGetDatum(pub->oid)))
962 : : {
963 : : /*
964 : : * Check for the presence of a row filter in this publication.
965 : : */
966 : 258 : rftuple = SearchSysCache2(PUBLICATIONRELMAP,
967 : : ObjectIdGetDatum(entry->publish_as_relid),
968 : : ObjectIdGetDatum(pub->oid));
969 : :
970 [ + + ]: 258 : if (HeapTupleIsValid(rftuple))
971 : : {
972 : : /* Null indicates no filter. */
973 : 246 : rfdatum = SysCacheGetAttr(PUBLICATIONRELMAP, rftuple,
974 : : Anum_pg_publication_rel_prqual,
975 : : &pub_no_filter);
976 : : }
977 : : }
978 : :
979 [ + + ]: 361 : if (pub_no_filter)
980 : : {
981 [ + + ]: 346 : if (rftuple)
982 : 231 : ReleaseSysCache(rftuple);
983 : :
984 : 346 : no_filter[PUBACTION_INSERT] |= pub->pubactions.pubinsert;
985 : 346 : no_filter[PUBACTION_UPDATE] |= pub->pubactions.pubupdate;
986 : 346 : no_filter[PUBACTION_DELETE] |= pub->pubactions.pubdelete;
987 : :
988 : : /*
989 : : * Quick exit if all the DML actions are publicized via this
990 : : * publication.
991 : : */
992 [ + - ]: 346 : if (no_filter[PUBACTION_INSERT] &&
993 [ + + ]: 346 : no_filter[PUBACTION_UPDATE] &&
994 [ + - ]: 339 : no_filter[PUBACTION_DELETE])
995 : : {
996 : 339 : has_filter = false;
997 : 339 : break;
998 : : }
999 : :
1000 : : /* No additional work for this publication. Next one. */
1001 : 7 : continue;
1002 : : }
1003 : :
1004 : : /* Form the per pubaction row filter lists. */
1005 [ + - + - ]: 15 : if (pub->pubactions.pubinsert && !no_filter[PUBACTION_INSERT])
1006 : 15 : rfnodes[PUBACTION_INSERT] = lappend(rfnodes[PUBACTION_INSERT],
1007 : 15 : TextDatumGetCString(rfdatum));
1008 [ + - + - ]: 15 : if (pub->pubactions.pubupdate && !no_filter[PUBACTION_UPDATE])
1009 : 15 : rfnodes[PUBACTION_UPDATE] = lappend(rfnodes[PUBACTION_UPDATE],
1010 : 15 : TextDatumGetCString(rfdatum));
1011 [ + - + - ]: 15 : if (pub->pubactions.pubdelete && !no_filter[PUBACTION_DELETE])
1012 : 15 : rfnodes[PUBACTION_DELETE] = lappend(rfnodes[PUBACTION_DELETE],
1013 : 15 : TextDatumGetCString(rfdatum));
1014 : :
1015 : 15 : ReleaseSysCache(rftuple);
1016 : : } /* loop all subscribed publications */
1017 : :
1018 : : /* Clean the row filter */
1019 [ + + ]: 1428 : for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
1020 : : {
1021 [ + + ]: 1071 : if (no_filter[idx])
1022 : : {
1023 : 1026 : list_free_deep(rfnodes[idx]);
1024 : 1026 : rfnodes[idx] = NIL;
1025 : : }
1026 : : }
1027 : :
1028 [ + + ]: 357 : if (has_filter)
1029 : : {
1030 : 18 : Relation relation = RelationIdGetRelation(entry->publish_as_relid);
1031 : :
1032 : 18 : pgoutput_ensure_entry_cxt(data, entry);
1033 : :
1034 : : /*
1035 : : * Now all the filters for all pubactions are known. Combine them when
1036 : : * their pubactions are the same.
1037 : : */
1038 : 18 : oldctx = MemoryContextSwitchTo(entry->entry_cxt);
1039 : 18 : entry->estate = create_estate_for_relation(relation);
1040 [ + + ]: 72 : for (idx = 0; idx < NUM_ROWFILTER_PUBACTIONS; idx++)
1041 : : {
1042 : 54 : List *filters = NIL;
1043 : : Expr *rfnode;
1044 : :
1045 [ + + ]: 54 : if (rfnodes[idx] == NIL)
1046 : 21 : continue;
1047 : :
1048 [ + - + + : 69 : foreach(lc, rfnodes[idx])
+ + ]
1049 : 36 : filters = lappend(filters, expand_generated_columns_in_expr(stringToNode((char *) lfirst(lc)), relation, 1));
1050 : :
1051 : : /* combine the row filter and cache the ExprState */
1052 : 33 : rfnode = make_orclause(filters);
1053 : 33 : entry->exprstate[idx] = ExecPrepareExpr(rfnode, entry->estate);
1054 : : } /* for each pubaction */
1055 : 18 : MemoryContextSwitchTo(oldctx);
1056 : :
1057 : 18 : RelationClose(relation);
1058 : : }
1059 : 357 : }
1060 : :
1061 : : /*
1062 : : * If the table contains a generated column, check for any conflicting
1063 : : * values of 'publish_generated_columns' parameter in the publications.
1064 : : */
1065 : : static void
1066 : 357 : check_and_init_gencol(PGOutputData *data, List *publications,
1067 : : RelationSyncEntry *entry)
1068 : : {
1069 : 357 : Relation relation = RelationIdGetRelation(entry->publish_as_relid);
1070 : 357 : TupleDesc desc = RelationGetDescr(relation);
1071 : 357 : bool gencolpresent = false;
1072 : 357 : bool first = true;
1073 : :
1074 : : /* Check if there is any generated column present. */
1075 [ + + ]: 1090 : for (int i = 0; i < desc->natts; i++)
1076 : : {
1077 : 740 : CompactAttribute *att = TupleDescCompactAttr(desc, i);
1078 : :
1079 [ + + ]: 740 : if (att->attgenerated)
1080 : : {
1081 : 7 : gencolpresent = true;
1082 : 7 : break;
1083 : : }
1084 : : }
1085 : :
1086 : : /* There are no generated columns to be published. */
1087 [ + + ]: 357 : if (!gencolpresent)
1088 : : {
1089 : 350 : entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
1090 : 350 : return;
1091 : : }
1092 : :
1093 : : /*
1094 : : * There may be a conflicting value for 'publish_generated_columns'
1095 : : * parameter in the publications.
1096 : : */
1097 [ + - + + : 22 : foreach_ptr(Publication, pub, publications)
+ + ]
1098 : : {
1099 : : /*
1100 : : * The column list takes precedence over the
1101 : : * 'publish_generated_columns' parameter. Those will be checked later,
1102 : : * see pgoutput_column_list_init.
1103 : : */
1104 [ + + ]: 8 : if (check_and_fetch_column_list(pub, entry->publish_as_relid, NULL, NULL))
1105 : 3 : continue;
1106 : :
1107 [ + - ]: 5 : if (first)
1108 : : {
1109 : 5 : entry->include_gencols_type = pub->pubgencols_type;
1110 : 5 : first = false;
1111 : : }
1112 [ # # ]: 0 : else if (entry->include_gencols_type != pub->pubgencols_type)
1113 [ # # ]: 0 : ereport(ERROR,
1114 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1115 : : errmsg("cannot use different values of publish_generated_columns for table \"%s.%s\" in different publications",
1116 : : get_namespace_name(RelationGetNamespace(relation)),
1117 : : RelationGetRelationName(relation)));
1118 : : }
1119 : : }
1120 : :
1121 : : /*
1122 : : * Initialize the column list.
1123 : : */
1124 : : static void
1125 : 357 : pgoutput_column_list_init(PGOutputData *data, List *publications,
1126 : : RelationSyncEntry *entry)
1127 : : {
1128 : : ListCell *lc;
1129 : 357 : bool first = true;
1130 : 357 : Relation relation = RelationIdGetRelation(entry->publish_as_relid);
1131 : 357 : bool found_pub_collist = false;
1132 : 357 : Bitmapset *relcols = NULL;
1133 : :
1134 : 357 : pgoutput_ensure_entry_cxt(data, entry);
1135 : :
1136 : : /*
1137 : : * Find if there are any column lists for this relation. If there are,
1138 : : * build a bitmap using the column lists.
1139 : : *
1140 : : * Multiple publications might have multiple column lists for this
1141 : : * relation.
1142 : : *
1143 : : * Note that we don't support the case where the column list is different
1144 : : * for the same table when combining publications. See comments atop
1145 : : * fetch_relation_list. But one can later change the publication so we
1146 : : * still need to check all the given publication-table mappings and report
1147 : : * an error if any publications have a different column list.
1148 : : */
1149 [ + - + + : 724 : foreach(lc, publications)
+ + ]
1150 : : {
1151 : 368 : Publication *pub = lfirst(lc);
1152 : 368 : Bitmapset *cols = NULL;
1153 : :
1154 : : /* Retrieve the bitmap of columns for a column list publication. */
1155 : 368 : found_pub_collist |= check_and_fetch_column_list(pub,
1156 : : entry->publish_as_relid,
1157 : : entry->entry_cxt, &cols);
1158 : :
1159 : : /*
1160 : : * For non-column list publications — e.g. TABLE (without a column
1161 : : * list), ALL TABLES, or ALL TABLES IN SCHEMA, we consider all columns
1162 : : * of the table (including generated columns when
1163 : : * 'publish_generated_columns' parameter is true).
1164 : : */
1165 [ + + ]: 368 : if (!cols)
1166 : : {
1167 : : /*
1168 : : * Cache the table columns for the first publication with no
1169 : : * specified column list to detect publication with a different
1170 : : * column list.
1171 : : */
1172 [ + + + + ]: 329 : if (!relcols && (list_length(publications) > 1))
1173 : : {
1174 : 9 : MemoryContext oldcxt = MemoryContextSwitchTo(entry->entry_cxt);
1175 : :
1176 : 9 : relcols = pub_form_cols_map(relation,
1177 : : entry->include_gencols_type);
1178 : 9 : MemoryContextSwitchTo(oldcxt);
1179 : : }
1180 : :
1181 : 329 : cols = relcols;
1182 : : }
1183 : :
1184 [ + + ]: 368 : if (first)
1185 : : {
1186 : 357 : entry->columns = cols;
1187 : 357 : first = false;
1188 : : }
1189 [ + + ]: 11 : else if (!bms_equal(entry->columns, cols))
1190 [ + - ]: 1 : ereport(ERROR,
1191 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1192 : : errmsg("cannot use different column lists for table \"%s.%s\" in different publications",
1193 : : get_namespace_name(RelationGetNamespace(relation)),
1194 : : RelationGetRelationName(relation)));
1195 : : } /* loop all subscribed publications */
1196 : :
1197 : : /*
1198 : : * If no column list publications exist, columns to be published will be
1199 : : * computed later according to the 'publish_generated_columns' parameter.
1200 : : */
1201 [ + + ]: 356 : if (!found_pub_collist)
1202 : 320 : entry->columns = NULL;
1203 : :
1204 : 356 : RelationClose(relation);
1205 : 356 : }
1206 : :
1207 : : /*
1208 : : * Initialize the slot for storing new and old tuples, and build the map that
1209 : : * will be used to convert the relation's tuples into the ancestor's format.
1210 : : */
1211 : : static void
1212 : 357 : init_tuple_slot(PGOutputData *data, Relation relation,
1213 : : RelationSyncEntry *entry)
1214 : : {
1215 : : MemoryContext oldctx;
1216 : : TupleDesc oldtupdesc;
1217 : : TupleDesc newtupdesc;
1218 : :
1219 : 357 : oldctx = MemoryContextSwitchTo(data->cachectx);
1220 : :
1221 : : /*
1222 : : * Create tuple table slots. Create a copy of the TupleDesc as it needs to
1223 : : * live as long as the cache remains.
1224 : : */
1225 : 357 : oldtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
1226 : 357 : newtupdesc = CreateTupleDescCopyConstr(RelationGetDescr(relation));
1227 : :
1228 : 357 : entry->old_slot = MakeSingleTupleTableSlot(oldtupdesc, &TTSOpsHeapTuple);
1229 : 357 : entry->new_slot = MakeSingleTupleTableSlot(newtupdesc, &TTSOpsHeapTuple);
1230 : :
1231 : 357 : MemoryContextSwitchTo(oldctx);
1232 : :
1233 : : /*
1234 : : * Cache the map that will be used to convert the relation's tuples into
1235 : : * the ancestor's format, if needed.
1236 : : */
1237 [ + + ]: 357 : if (entry->publish_as_relid != RelationGetRelid(relation))
1238 : : {
1239 : 43 : Relation ancestor = RelationIdGetRelation(entry->publish_as_relid);
1240 : 43 : TupleDesc indesc = RelationGetDescr(relation);
1241 : 43 : TupleDesc outdesc = RelationGetDescr(ancestor);
1242 : :
1243 : : /* Map must live as long as the logical decoding context. */
1244 : 43 : oldctx = MemoryContextSwitchTo(data->cachectx);
1245 : :
1246 : 43 : entry->attrmap = build_attrmap_by_name_if_req(indesc, outdesc, false);
1247 : :
1248 : 43 : MemoryContextSwitchTo(oldctx);
1249 : 43 : RelationClose(ancestor);
1250 : : }
1251 : 357 : }
1252 : :
1253 : : /*
1254 : : * Change is checked against the row filter if any.
1255 : : *
1256 : : * Returns true if the change is to be replicated, else false.
1257 : : *
1258 : : * For inserts, evaluate the row filter for new tuple.
1259 : : * For deletes, evaluate the row filter for old tuple.
1260 : : * For updates, evaluate the row filter for old and new tuple.
1261 : : *
1262 : : * For updates, if both evaluations are true, we allow sending the UPDATE and
1263 : : * if both the evaluations are false, it doesn't replicate the UPDATE. Now, if
1264 : : * only one of the tuples matches the row filter expression, we transform
1265 : : * UPDATE to DELETE or INSERT to avoid any data inconsistency based on the
1266 : : * following rules:
1267 : : *
1268 : : * Case 1: old-row (no match) new-row (no match) -> (drop change)
1269 : : * Case 2: old-row (no match) new row (match) -> INSERT
1270 : : * Case 3: old-row (match) new-row (no match) -> DELETE
1271 : : * Case 4: old-row (match) new row (match) -> UPDATE
1272 : : *
1273 : : * The new action is updated in the action parameter.
1274 : : *
1275 : : * The new slot could be updated when transforming the UPDATE into INSERT,
1276 : : * because the original new tuple might not have column values from the replica
1277 : : * identity.
1278 : : *
1279 : : * Examples:
1280 : : * Let's say the old tuple satisfies the row filter but the new tuple doesn't.
1281 : : * Since the old tuple satisfies, the initial table synchronization copied this
1282 : : * row (or another method was used to guarantee that there is data
1283 : : * consistency). However, after the UPDATE the new tuple doesn't satisfy the
1284 : : * row filter, so from a data consistency perspective, that row should be
1285 : : * removed on the subscriber. The UPDATE should be transformed into a DELETE
1286 : : * statement and be sent to the subscriber. Keeping this row on the subscriber
1287 : : * is undesirable because it doesn't reflect what was defined in the row filter
1288 : : * expression on the publisher. This row on the subscriber would likely not be
1289 : : * modified by replication again. If someone inserted a new row with the same
1290 : : * old identifier, replication could stop due to a constraint violation.
1291 : : *
1292 : : * Let's say the old tuple doesn't match the row filter but the new tuple does.
1293 : : * Since the old tuple doesn't satisfy, the initial table synchronization
1294 : : * probably didn't copy this row. However, after the UPDATE the new tuple does
1295 : : * satisfy the row filter, so from a data consistency perspective, that row
1296 : : * should be inserted on the subscriber. Otherwise, subsequent UPDATE or DELETE
1297 : : * statements have no effect (it matches no row -- see
1298 : : * apply_handle_update_internal()). So, the UPDATE should be transformed into a
1299 : : * INSERT statement and be sent to the subscriber. However, this might surprise
1300 : : * someone who expects the data set to satisfy the row filter expression on the
1301 : : * provider.
1302 : : */
1303 : : static bool
1304 : 202719 : pgoutput_row_filter(Relation relation, TupleTableSlot *old_slot,
1305 : : TupleTableSlot **new_slot_ptr, RelationSyncEntry *entry,
1306 : : ReorderBufferChangeType *action)
1307 : : {
1308 : : TupleDesc desc;
1309 : : int i;
1310 : : bool old_matched,
1311 : : new_matched,
1312 : : result;
1313 : : TupleTableSlot *tmp_new_slot;
1314 : 202719 : TupleTableSlot *new_slot = *new_slot_ptr;
1315 : : ExprContext *ecxt;
1316 : : ExprState *filter_exprstate;
1317 : :
1318 : : /*
1319 : : * We need this map to avoid relying on ReorderBufferChangeType enums
1320 : : * having specific values.
1321 : : */
1322 : : static const int map_changetype_pubaction[] = {
1323 : : [REORDER_BUFFER_CHANGE_INSERT] = PUBACTION_INSERT,
1324 : : [REORDER_BUFFER_CHANGE_UPDATE] = PUBACTION_UPDATE,
1325 : : [REORDER_BUFFER_CHANGE_DELETE] = PUBACTION_DELETE
1326 : : };
1327 : :
1328 : : Assert(*action == REORDER_BUFFER_CHANGE_INSERT ||
1329 : : *action == REORDER_BUFFER_CHANGE_UPDATE ||
1330 : : *action == REORDER_BUFFER_CHANGE_DELETE);
1331 : :
1332 : : Assert(new_slot || old_slot);
1333 : :
1334 : : /* Get the corresponding row filter */
1335 : 202719 : filter_exprstate = entry->exprstate[map_changetype_pubaction[*action]];
1336 : :
1337 : : /* Bail out if there is no row filter */
1338 [ + + ]: 202719 : if (!filter_exprstate)
1339 : 202684 : return true;
1340 : :
1341 [ - + ]: 35 : elog(DEBUG3, "table \"%s.%s\" has row filter",
1342 : : get_namespace_name(RelationGetNamespace(relation)),
1343 : : RelationGetRelationName(relation));
1344 : :
1345 [ + + ]: 35 : ResetPerTupleExprContext(entry->estate);
1346 : :
1347 [ + + ]: 35 : ecxt = GetPerTupleExprContext(entry->estate);
1348 : :
1349 : : /*
1350 : : * For the following occasions where there is only one tuple, we can
1351 : : * evaluate the row filter for that tuple and return.
1352 : : *
1353 : : * For inserts, we only have the new tuple.
1354 : : *
1355 : : * For updates, we can have only a new tuple when none of the replica
1356 : : * identity columns changed and none of those columns have external data
1357 : : * but we still need to evaluate the row filter for the new tuple as the
1358 : : * existing values of those columns might not match the filter. Also,
1359 : : * users can use constant expressions in the row filter, so we anyway need
1360 : : * to evaluate it for the new tuple.
1361 : : *
1362 : : * For deletes, we only have the old tuple.
1363 : : */
1364 [ + + + + ]: 35 : if (!new_slot || !old_slot)
1365 : : {
1366 [ + + ]: 31 : ecxt->ecxt_scantuple = new_slot ? new_slot : old_slot;
1367 : 31 : result = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
1368 : :
1369 : 30 : return result;
1370 : : }
1371 : :
1372 : : /*
1373 : : * Both the old and new tuples must be valid only for updates and need to
1374 : : * be checked against the row filter.
1375 : : */
1376 : : Assert(map_changetype_pubaction[*action] == PUBACTION_UPDATE);
1377 : :
1378 : 4 : slot_getallattrs(new_slot);
1379 : 4 : slot_getallattrs(old_slot);
1380 : :
1381 : 4 : tmp_new_slot = NULL;
1382 : 4 : desc = RelationGetDescr(relation);
1383 : :
1384 : : /*
1385 : : * The new tuple might not have all the replica identity columns, in which
1386 : : * case it needs to be copied over from the old tuple.
1387 : : */
1388 [ + + ]: 12 : for (i = 0; i < desc->natts; i++)
1389 : : {
1390 : 8 : CompactAttribute *att = TupleDescCompactAttr(desc, i);
1391 : :
1392 : : /*
1393 : : * if the column in the new tuple or old tuple is null, nothing to do
1394 : : */
1395 [ + + - + ]: 8 : if (new_slot->tts_isnull[i] || old_slot->tts_isnull[i])
1396 : 1 : continue;
1397 : :
1398 : : /*
1399 : : * Unchanged toasted replica identity columns are only logged in the
1400 : : * old tuple. Copy this over to the new tuple. The changed (or WAL
1401 : : * Logged) toast values are always assembled in memory and set as
1402 : : * VARTAG_INDIRECT. See ReorderBufferToastReplace.
1403 : : */
1404 [ + + + + ]: 11 : if (att->attlen == -1 &&
1405 : 4 : VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(new_slot->tts_values[i])) &&
1406 [ + - ]: 1 : !VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(old_slot->tts_values[i])))
1407 : : {
1408 [ + - ]: 1 : if (!tmp_new_slot)
1409 : : {
1410 : 1 : tmp_new_slot = MakeSingleTupleTableSlot(desc, &TTSOpsVirtual);
1411 : 1 : ExecClearTuple(tmp_new_slot);
1412 : :
1413 : 1 : memcpy(tmp_new_slot->tts_values, new_slot->tts_values,
1414 : 1 : desc->natts * sizeof(Datum));
1415 : 1 : memcpy(tmp_new_slot->tts_isnull, new_slot->tts_isnull,
1416 : 1 : desc->natts * sizeof(bool));
1417 : : }
1418 : :
1419 : 1 : tmp_new_slot->tts_values[i] = old_slot->tts_values[i];
1420 : 1 : tmp_new_slot->tts_isnull[i] = old_slot->tts_isnull[i];
1421 : : }
1422 : : }
1423 : :
1424 : 4 : ecxt->ecxt_scantuple = old_slot;
1425 : 4 : old_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
1426 : :
1427 [ + + ]: 4 : if (tmp_new_slot)
1428 : : {
1429 : 1 : ExecStoreVirtualTuple(tmp_new_slot);
1430 : 1 : ecxt->ecxt_scantuple = tmp_new_slot;
1431 : : }
1432 : : else
1433 : 3 : ecxt->ecxt_scantuple = new_slot;
1434 : :
1435 : 4 : new_matched = pgoutput_row_filter_exec_expr(filter_exprstate, ecxt);
1436 : :
1437 : : /*
1438 : : * Case 1: if both tuples don't match the row filter, bailout. Send
1439 : : * nothing.
1440 : : */
1441 [ + + - + ]: 4 : if (!old_matched && !new_matched)
1442 : 0 : return false;
1443 : :
1444 : : /*
1445 : : * Case 2: if the old tuple doesn't satisfy the row filter but the new
1446 : : * tuple does, transform the UPDATE into INSERT.
1447 : : *
1448 : : * Use the newly transformed tuple that must contain the column values for
1449 : : * all the replica identity columns. This is required to ensure that the
1450 : : * while inserting the tuple in the downstream node, we have all the
1451 : : * required column values.
1452 : : */
1453 [ + + + - ]: 4 : if (!old_matched && new_matched)
1454 : : {
1455 : 2 : *action = REORDER_BUFFER_CHANGE_INSERT;
1456 : :
1457 [ + + ]: 2 : if (tmp_new_slot)
1458 : 1 : *new_slot_ptr = tmp_new_slot;
1459 : : }
1460 : :
1461 : : /*
1462 : : * Case 3: if the old tuple satisfies the row filter but the new tuple
1463 : : * doesn't, transform the UPDATE into DELETE.
1464 : : *
1465 : : * This transformation does not require another tuple. The Old tuple will
1466 : : * be used for DELETE.
1467 : : */
1468 [ + - + + ]: 2 : else if (old_matched && !new_matched)
1469 : 1 : *action = REORDER_BUFFER_CHANGE_DELETE;
1470 : :
1471 : : /*
1472 : : * Case 4: if both tuples match the row filter, transformation isn't
1473 : : * required. (*action is default UPDATE).
1474 : : */
1475 : :
1476 : 4 : return true;
1477 : : }
1478 : :
1479 : : /*
1480 : : * Sends the decoded DML over wire.
1481 : : *
1482 : : * This is called both in streaming and non-streaming modes.
1483 : : */
1484 : : static void
1485 : 204527 : pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
1486 : : Relation relation, ReorderBufferChange *change)
1487 : : {
1488 : 204527 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1489 : 204527 : PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
1490 : : MemoryContext old;
1491 : : RelationSyncEntry *relentry;
1492 : 204527 : TransactionId xid = InvalidTransactionId;
1493 : 204527 : Relation ancestor = NULL;
1494 : 204527 : Relation targetrel = relation;
1495 : 204527 : ReorderBufferChangeType action = change->action;
1496 : 204527 : TupleTableSlot *old_slot = NULL;
1497 : 204527 : TupleTableSlot *new_slot = NULL;
1498 : :
1499 [ - + ]: 204527 : if (!is_publishable_relation(relation))
1500 : 1807 : return;
1501 : :
1502 : : /*
1503 : : * Remember the xid for the change in streaming mode. We need to send xid
1504 : : * with each change in the streaming mode so that subscriber can make
1505 : : * their association and on aborts, it can discard the corresponding
1506 : : * changes.
1507 : : */
1508 [ + + ]: 204527 : if (data->in_streaming)
1509 : 175923 : xid = change->txn->xid;
1510 : :
1511 : 204527 : relentry = get_rel_sync_entry(data, relation);
1512 : :
1513 : : /* First check the table filter */
1514 [ + + + - ]: 204526 : switch (action)
1515 : : {
1516 : 127050 : case REORDER_BUFFER_CHANGE_INSERT:
1517 [ + + ]: 127050 : if (!relentry->pubactions.pubinsert)
1518 : 674 : return;
1519 : 126376 : break;
1520 : 34501 : case REORDER_BUFFER_CHANGE_UPDATE:
1521 [ + + ]: 34501 : if (!relentry->pubactions.pubupdate)
1522 : 44 : return;
1523 : 34457 : break;
1524 : 42975 : case REORDER_BUFFER_CHANGE_DELETE:
1525 [ + + ]: 42975 : if (!relentry->pubactions.pubdelete)
1526 : 1089 : return;
1527 : :
1528 : : /*
1529 : : * This is only possible if deletes are allowed even when replica
1530 : : * identity is not defined for a table. Since the DELETE action
1531 : : * can't be published, we simply return.
1532 : : */
1533 [ - + ]: 41886 : if (!change->data.tp.oldtuple)
1534 : : {
1535 [ # # ]: 0 : elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
1536 : 0 : return;
1537 : : }
1538 : 41886 : break;
1539 : 202719 : default:
1540 : : Assert(false);
1541 : : }
1542 : :
1543 : : /* Avoid leaking memory by using and resetting our own context */
1544 : 202719 : old = MemoryContextSwitchTo(data->context);
1545 : :
1546 : : /* Switch relation if publishing via root. */
1547 [ + + ]: 202719 : if (relentry->publish_as_relid != RelationGetRelid(relation))
1548 : : {
1549 : : Assert(relation->rd_rel->relispartition);
1550 : 74 : ancestor = RelationIdGetRelation(relentry->publish_as_relid);
1551 : 74 : targetrel = ancestor;
1552 : : }
1553 : :
1554 [ + + ]: 202719 : if (change->data.tp.oldtuple)
1555 : : {
1556 : 42036 : old_slot = relentry->old_slot;
1557 : 42036 : ExecStoreHeapTuple(change->data.tp.oldtuple, old_slot, false);
1558 : :
1559 : : /* Convert tuple if needed. */
1560 [ + + ]: 42036 : if (relentry->attrmap)
1561 : : {
1562 : 5 : TupleTableSlot *slot = MakeTupleTableSlot(RelationGetDescr(targetrel),
1563 : : &TTSOpsVirtual, 0);
1564 : :
1565 : 5 : old_slot = execute_attr_map_slot(relentry->attrmap, old_slot, slot);
1566 : : }
1567 : : }
1568 : :
1569 [ + + ]: 202719 : if (change->data.tp.newtuple)
1570 : : {
1571 : 160833 : new_slot = relentry->new_slot;
1572 : 160833 : ExecStoreHeapTuple(change->data.tp.newtuple, new_slot, false);
1573 : :
1574 : : /* Convert tuple if needed. */
1575 [ + + ]: 160833 : if (relentry->attrmap)
1576 : : {
1577 : 21 : TupleTableSlot *slot = MakeTupleTableSlot(RelationGetDescr(targetrel),
1578 : : &TTSOpsVirtual, 0);
1579 : :
1580 : 21 : new_slot = execute_attr_map_slot(relentry->attrmap, new_slot, slot);
1581 : : }
1582 : : }
1583 : :
1584 : : /*
1585 : : * Check row filter.
1586 : : *
1587 : : * Updates could be transformed to inserts or deletes based on the results
1588 : : * of the row filter for old and new tuple.
1589 : : */
1590 [ + + ]: 202719 : if (!pgoutput_row_filter(targetrel, old_slot, &new_slot, relentry, &action))
1591 : 12 : goto cleanup;
1592 : :
1593 : : /*
1594 : : * Send BEGIN if we haven't yet.
1595 : : *
1596 : : * We send the BEGIN message after ensuring that we will actually send the
1597 : : * change. This avoids sending a pair of BEGIN/COMMIT messages for empty
1598 : : * transactions.
1599 : : */
1600 [ + + + + ]: 202706 : if (txndata && !txndata->sent_begin_txn)
1601 : 501 : pgoutput_send_begin(ctx, txn);
1602 : :
1603 : : /*
1604 : : * Schema should be sent using the original relation because it also sends
1605 : : * the ancestor's relation.
1606 : : */
1607 : 202706 : maybe_send_schema(ctx, change, relation, relentry);
1608 : :
1609 : 202706 : OutputPluginPrepareWrite(ctx, true);
1610 : :
1611 : : /* Send the data */
1612 [ + + + - ]: 202706 : switch (action)
1613 : : {
1614 : 126365 : case REORDER_BUFFER_CHANGE_INSERT:
1615 : 126365 : logicalrep_write_insert(ctx->out, xid, targetrel, new_slot,
1616 : 126365 : data->binary, relentry->columns,
1617 : : relentry->include_gencols_type);
1618 : 126365 : break;
1619 : 34454 : case REORDER_BUFFER_CHANGE_UPDATE:
1620 : 34454 : logicalrep_write_update(ctx->out, xid, targetrel, old_slot,
1621 : 34454 : new_slot, data->binary, relentry->columns,
1622 : : relentry->include_gencols_type);
1623 : 34454 : break;
1624 : 41887 : case REORDER_BUFFER_CHANGE_DELETE:
1625 : 41887 : logicalrep_write_delete(ctx->out, xid, targetrel, old_slot,
1626 : 41887 : data->binary, relentry->columns,
1627 : : relentry->include_gencols_type);
1628 : 41887 : break;
1629 : 202706 : default:
1630 : : Assert(false);
1631 : : }
1632 : :
1633 : 202706 : OutputPluginWrite(ctx, true);
1634 : :
1635 : 202717 : cleanup:
1636 [ + + ]: 202717 : if (RelationIsValid(ancestor))
1637 : : {
1638 : 74 : RelationClose(ancestor);
1639 : 74 : ancestor = NULL;
1640 : : }
1641 : :
1642 : : /* Drop the new slots that were used to store the converted tuples. */
1643 [ + + ]: 202717 : if (relentry->attrmap)
1644 : : {
1645 [ + + ]: 26 : if (old_slot)
1646 : 5 : ExecDropSingleTupleTableSlot(old_slot);
1647 : :
1648 [ + + ]: 26 : if (new_slot)
1649 : 21 : ExecDropSingleTupleTableSlot(new_slot);
1650 : : }
1651 : :
1652 : 202717 : MemoryContextSwitchTo(old);
1653 : 202717 : MemoryContextReset(data->context);
1654 : : }
1655 : :
1656 : : static void
1657 : 28 : pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
1658 : : int nrelations, Relation relations[], ReorderBufferChange *change)
1659 : : {
1660 : 28 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1661 : 28 : PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
1662 : : MemoryContext old;
1663 : : RelationSyncEntry *relentry;
1664 : : int i;
1665 : : int nrelids;
1666 : : Oid *relids;
1667 : 28 : TransactionId xid = InvalidTransactionId;
1668 : :
1669 : : /* Remember the xid for the change in streaming mode. See pgoutput_change. */
1670 [ - + ]: 28 : if (data->in_streaming)
1671 : 0 : xid = change->txn->xid;
1672 : :
1673 : 28 : old = MemoryContextSwitchTo(data->context);
1674 : :
1675 : 28 : relids = palloc0(nrelations * sizeof(Oid));
1676 : 28 : nrelids = 0;
1677 : :
1678 [ + + ]: 75 : for (i = 0; i < nrelations; i++)
1679 : : {
1680 : 47 : Relation relation = relations[i];
1681 : 47 : Oid relid = RelationGetRelid(relation);
1682 : :
1683 [ - + ]: 47 : if (!is_publishable_relation(relation))
1684 : 0 : continue;
1685 : :
1686 : 47 : relentry = get_rel_sync_entry(data, relation);
1687 : :
1688 [ + + ]: 47 : if (!relentry->pubactions.pubtruncate)
1689 : 26 : continue;
1690 : :
1691 : : /*
1692 : : * Don't send partitions if the publication wants to send only the
1693 : : * root tables through it.
1694 : : */
1695 [ + + ]: 21 : if (relation->rd_rel->relispartition &&
1696 [ + + ]: 15 : relentry->publish_as_relid != relid)
1697 : 3 : continue;
1698 : :
1699 : 18 : relids[nrelids++] = relid;
1700 : :
1701 : : /* Send BEGIN if we haven't yet */
1702 [ + - + + ]: 18 : if (txndata && !txndata->sent_begin_txn)
1703 : 12 : pgoutput_send_begin(ctx, txn);
1704 : :
1705 : 18 : maybe_send_schema(ctx, change, relation, relentry);
1706 : : }
1707 : :
1708 [ + + ]: 28 : if (nrelids > 0)
1709 : : {
1710 : 12 : OutputPluginPrepareWrite(ctx, true);
1711 : 12 : logicalrep_write_truncate(ctx->out,
1712 : : xid,
1713 : : nrelids,
1714 : : relids,
1715 : 12 : change->data.truncate.cascade,
1716 : 12 : change->data.truncate.restart_seqs);
1717 : 12 : OutputPluginWrite(ctx, true);
1718 : : }
1719 : :
1720 : 28 : MemoryContextSwitchTo(old);
1721 : 28 : MemoryContextReset(data->context);
1722 : 28 : }
1723 : :
1724 : : static void
1725 : 7 : pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
1726 : : XLogRecPtr message_lsn, bool transactional, const char *prefix, Size sz,
1727 : : const char *message)
1728 : : {
1729 : 7 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1730 : 7 : TransactionId xid = InvalidTransactionId;
1731 : :
1732 [ + + ]: 7 : if (!data->messages)
1733 : 2 : return;
1734 : :
1735 : : /*
1736 : : * Remember the xid for the message in streaming mode. See
1737 : : * pgoutput_change.
1738 : : */
1739 [ - + ]: 5 : if (data->in_streaming)
1740 : 0 : xid = txn->xid;
1741 : :
1742 : : /*
1743 : : * Output BEGIN if we haven't yet. Avoid for non-transactional messages.
1744 : : */
1745 [ + + ]: 5 : if (transactional)
1746 : : {
1747 : 2 : PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
1748 : :
1749 : : /* Send BEGIN if we haven't yet */
1750 [ + - + - ]: 2 : if (txndata && !txndata->sent_begin_txn)
1751 : 2 : pgoutput_send_begin(ctx, txn);
1752 : : }
1753 : :
1754 : 5 : OutputPluginPrepareWrite(ctx, true);
1755 : 5 : logicalrep_write_message(ctx->out,
1756 : : xid,
1757 : : message_lsn,
1758 : : transactional,
1759 : : prefix,
1760 : : sz,
1761 : : message);
1762 : 5 : OutputPluginWrite(ctx, true);
1763 : : }
1764 : :
1765 : : /*
1766 : : * Return true if the data is associated with an origin and the user has
1767 : : * requested the changes that don't have an origin, false otherwise.
1768 : : */
1769 : : static bool
1770 : 533650 : pgoutput_origin_filter(LogicalDecodingContext *ctx,
1771 : : ReplOriginId origin_id)
1772 : : {
1773 : 533650 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1774 : :
1775 [ + + + + ]: 533650 : if (data->publish_no_origin && origin_id != InvalidReplOriginId)
1776 : 215 : return true;
1777 : :
1778 : 533435 : return false;
1779 : : }
1780 : :
1781 : : /*
1782 : : * Shutdown the output plugin.
1783 : : *
1784 : : * Note, we don't need to clean the data->context, data->cachectx, and
1785 : : * data->pubctx as they are child contexts of the ctx->context so they
1786 : : * will be cleaned up by logical decoding machinery.
1787 : : */
1788 : : static void
1789 : 572 : pgoutput_shutdown(LogicalDecodingContext *ctx)
1790 : : {
1791 : 572 : pgoutput_memory_context_reset(NULL);
1792 : 572 : }
1793 : :
1794 : : /*
1795 : : * Load publications from the list of publication names.
1796 : : *
1797 : : * Here, we skip the publications that don't exist yet. This will allow us
1798 : : * to silently continue the replication in the absence of a missing publication.
1799 : : * This is required because we allow the users to create publications after they
1800 : : * have specified the required publications at the time of replication start.
1801 : : */
1802 : : static List *
1803 : 266 : LoadPublications(List *pubnames)
1804 : : {
1805 : 266 : List *result = NIL;
1806 : : ListCell *lc;
1807 : :
1808 [ + - + + : 587 : foreach(lc, pubnames)
+ + ]
1809 : : {
1810 : 321 : char *pubname = (char *) lfirst(lc);
1811 : 321 : Publication *pub = GetPublicationByName(pubname, true);
1812 : :
1813 [ + + ]: 321 : if (pub)
1814 : 318 : result = lappend(result, pub);
1815 : : else
1816 [ + - ]: 3 : ereport(WARNING,
1817 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1818 : : errmsg("skipped loading publication \"%s\"", pubname),
1819 : : errdetail("The publication does not exist at this point in the WAL."),
1820 : : errhint("Create the publication if it does not exist."));
1821 : : }
1822 : :
1823 : 266 : return result;
1824 : : }
1825 : :
1826 : : /*
1827 : : * Publication syscache invalidation callback.
1828 : : *
1829 : : * Called for invalidations on pg_publication.
1830 : : */
1831 : : static void
1832 : 393 : publication_invalidation_cb(Datum arg, SysCacheIdentifier cacheid,
1833 : : uint32 hashvalue)
1834 : : {
1835 : 393 : publications_valid = false;
1836 : 393 : }
1837 : :
1838 : : /*
1839 : : * START STREAM callback
1840 : : */
1841 : : static void
1842 : 620 : pgoutput_stream_start(struct LogicalDecodingContext *ctx,
1843 : : ReorderBufferTXN *txn)
1844 : : {
1845 : 620 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1846 : 620 : bool send_replication_origin = txn->origin_id != InvalidReplOriginId;
1847 : :
1848 : : /* we can't nest streaming of transactions */
1849 : : Assert(!data->in_streaming);
1850 : :
1851 : : /*
1852 : : * If we already sent the first stream for this transaction then don't
1853 : : * send the origin id in the subsequent streams.
1854 : : */
1855 [ + + ]: 620 : if (rbtxn_is_streamed(txn))
1856 : 559 : send_replication_origin = false;
1857 : :
1858 : 620 : OutputPluginPrepareWrite(ctx, !send_replication_origin);
1859 : 620 : logicalrep_write_stream_start(ctx->out, txn->xid, !rbtxn_is_streamed(txn));
1860 : :
1861 : 620 : send_repl_origin(ctx, txn->origin_id, InvalidXLogRecPtr,
1862 : : send_replication_origin);
1863 : :
1864 : 620 : OutputPluginWrite(ctx, true);
1865 : :
1866 : : /* we're streaming a chunk of transaction now */
1867 : 620 : data->in_streaming = true;
1868 : 620 : }
1869 : :
1870 : : /*
1871 : : * STOP STREAM callback
1872 : : */
1873 : : static void
1874 : 620 : pgoutput_stream_stop(struct LogicalDecodingContext *ctx,
1875 : : ReorderBufferTXN *txn)
1876 : : {
1877 : 620 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1878 : :
1879 : : /* we should be streaming a transaction */
1880 : : Assert(data->in_streaming);
1881 : :
1882 : 620 : OutputPluginPrepareWrite(ctx, true);
1883 : 620 : logicalrep_write_stream_stop(ctx->out);
1884 : 620 : OutputPluginWrite(ctx, true);
1885 : :
1886 : : /* we've stopped streaming a transaction */
1887 : 620 : data->in_streaming = false;
1888 : 620 : }
1889 : :
1890 : : /*
1891 : : * Notify downstream to discard the streamed transaction (along with all
1892 : : * its subtransactions, if it's a toplevel transaction).
1893 : : */
1894 : : static void
1895 : 26 : pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
1896 : : ReorderBufferTXN *txn,
1897 : : XLogRecPtr abort_lsn)
1898 : : {
1899 : : ReorderBufferTXN *toptxn;
1900 : 26 : PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
1901 : 26 : bool write_abort_info = (data->streaming == LOGICALREP_STREAM_PARALLEL);
1902 : :
1903 : : /*
1904 : : * The abort should happen outside streaming block, even for streamed
1905 : : * transactions. The transaction has to be marked as streamed, though.
1906 : : */
1907 : : Assert(!data->in_streaming);
1908 : :
1909 : : /* determine the toplevel transaction */
1910 [ + + ]: 26 : toptxn = rbtxn_get_toptxn(txn);
1911 : :
1912 : : Assert(rbtxn_is_streamed(toptxn));
1913 : :
1914 : 26 : OutputPluginPrepareWrite(ctx, true);
1915 : 26 : logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn->xid, abort_lsn,
1916 : : txn->abort_time, write_abort_info);
1917 : :
1918 : 26 : OutputPluginWrite(ctx, true);
1919 : :
1920 : 26 : cleanup_rel_sync_cache(toptxn->xid, false);
1921 : 26 : }
1922 : :
1923 : : /*
1924 : : * Notify downstream to apply the streamed transaction (along with all
1925 : : * its subtransactions).
1926 : : */
1927 : : static void
1928 : 45 : pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
1929 : : ReorderBufferTXN *txn,
1930 : : XLogRecPtr commit_lsn)
1931 : : {
1932 : 45 : PGOutputData *data PG_USED_FOR_ASSERTS_ONLY = (PGOutputData *) ctx->output_plugin_private;
1933 : :
1934 : : /*
1935 : : * The commit should happen outside streaming block, even for streamed
1936 : : * transactions. The transaction has to be marked as streamed, though.
1937 : : */
1938 : : Assert(!data->in_streaming);
1939 : : Assert(rbtxn_is_streamed(txn));
1940 : :
1941 : 45 : OutputPluginUpdateProgress(ctx, false);
1942 : :
1943 : 45 : OutputPluginPrepareWrite(ctx, true);
1944 : 45 : logicalrep_write_stream_commit(ctx->out, txn, commit_lsn);
1945 : 45 : OutputPluginWrite(ctx, true);
1946 : :
1947 : 45 : cleanup_rel_sync_cache(txn->xid, true);
1948 : 45 : }
1949 : :
1950 : : /*
1951 : : * PREPARE callback (for streaming two-phase commit).
1952 : : *
1953 : : * Notify the downstream to prepare the transaction.
1954 : : */
1955 : : static void
1956 : 13 : pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
1957 : : ReorderBufferTXN *txn,
1958 : : XLogRecPtr prepare_lsn)
1959 : : {
1960 : : Assert(rbtxn_is_streamed(txn));
1961 : :
1962 : 13 : OutputPluginUpdateProgress(ctx, false);
1963 : 13 : OutputPluginPrepareWrite(ctx, true);
1964 : 13 : logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
1965 : 13 : OutputPluginWrite(ctx, true);
1966 : 13 : }
1967 : :
1968 : : /*
1969 : : * Initialize the relation schema sync cache for a decoding session.
1970 : : *
1971 : : * The hash table is destroyed at the end of a decoding session. While
1972 : : * relcache invalidations still exist and will still be invoked, they
1973 : : * will just see the null hash table global and take no action.
1974 : : */
1975 : : static void
1976 : 466 : init_rel_sync_cache(MemoryContext cachectx)
1977 : : {
1978 : : HASHCTL ctl;
1979 : : static bool relation_callbacks_registered = false;
1980 : :
1981 : : /* Nothing to do if hash table already exists */
1982 [ - + ]: 466 : if (RelationSyncCache != NULL)
1983 : 2 : return;
1984 : :
1985 : : /* Make a new hash table for the cache */
1986 : 466 : ctl.keysize = sizeof(Oid);
1987 : 466 : ctl.entrysize = sizeof(RelationSyncEntry);
1988 : 466 : ctl.hcxt = cachectx;
1989 : :
1990 : 466 : RelationSyncCache = hash_create("logical replication output relation cache",
1991 : : 128, &ctl,
1992 : : HASH_ELEM | HASH_CONTEXT | HASH_BLOBS);
1993 : :
1994 : : Assert(RelationSyncCache != NULL);
1995 : :
1996 : : /* No more to do if we already registered callbacks */
1997 [ + + ]: 466 : if (relation_callbacks_registered)
1998 : 2 : return;
1999 : :
2000 : : /* We must update the cache entry for a relation after a relcache flush */
2001 : 464 : CacheRegisterRelcacheCallback(rel_sync_cache_relation_cb, (Datum) 0);
2002 : :
2003 : : /*
2004 : : * Flush all cache entries after a pg_namespace change, in case it was a
2005 : : * schema rename affecting a relation being replicated.
2006 : : *
2007 : : * XXX: It is not a good idea to invalidate all the relation entries in
2008 : : * RelationSyncCache on schema rename. We can optimize it to invalidate
2009 : : * only the required relations by either having a specific invalidation
2010 : : * message containing impacted relations or by having schema information
2011 : : * in each RelationSyncCache entry and using hashvalue of pg_namespace.oid
2012 : : * passed to the callback.
2013 : : */
2014 : 464 : CacheRegisterSyscacheCallback(NAMESPACEOID,
2015 : : rel_sync_cache_publication_cb,
2016 : : (Datum) 0);
2017 : :
2018 : 464 : relation_callbacks_registered = true;
2019 : : }
2020 : :
2021 : : /*
2022 : : * We expect relatively small number of streamed transactions.
2023 : : */
2024 : : static bool
2025 : 175923 : get_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
2026 : : {
2027 : 175923 : return list_member_xid(entry->streamed_txns, xid);
2028 : : }
2029 : :
2030 : : /*
2031 : : * Add the xid in the rel sync entry for which we have already sent the schema
2032 : : * of the relation.
2033 : : */
2034 : : static void
2035 : 70 : set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
2036 : : {
2037 : : MemoryContext oldctx;
2038 : :
2039 : 70 : oldctx = MemoryContextSwitchTo(CacheMemoryContext);
2040 : :
2041 : 70 : entry->streamed_txns = lappend_xid(entry->streamed_txns, xid);
2042 : :
2043 : 70 : MemoryContextSwitchTo(oldctx);
2044 : 70 : }
2045 : :
2046 : : /*
2047 : : * Find or create entry in the relation schema cache.
2048 : : *
2049 : : * This looks up publications that the given relation is directly or
2050 : : * indirectly part of (the latter if it's really the relation's ancestor that
2051 : : * is part of a publication) and fills up the found entry with the information
2052 : : * about which operations to publish and whether to use an ancestor's schema
2053 : : * when publishing.
2054 : : */
2055 : : static RelationSyncEntry *
2056 : 204574 : get_rel_sync_entry(PGOutputData *data, Relation relation)
2057 : : {
2058 : : RelationSyncEntry *entry;
2059 : : bool found;
2060 : : MemoryContext oldctx;
2061 : 204574 : Oid relid = RelationGetRelid(relation);
2062 : :
2063 : : Assert(RelationSyncCache != NULL);
2064 : :
2065 : : /* Find cached relation info, creating if not found */
2066 : 204574 : entry = (RelationSyncEntry *) hash_search(RelationSyncCache,
2067 : : &relid,
2068 : : HASH_ENTER, &found);
2069 : : Assert(entry != NULL);
2070 : :
2071 : : /* initialize entry, if it's new */
2072 [ + + ]: 204574 : if (!found)
2073 : : {
2074 : 377 : entry->replicate_valid = false;
2075 : 377 : entry->schema_sent = false;
2076 : 377 : entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
2077 : 377 : entry->streamed_txns = NIL;
2078 : 377 : entry->pubactions.pubinsert = entry->pubactions.pubupdate =
2079 : 377 : entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
2080 : 377 : entry->new_slot = NULL;
2081 : 377 : entry->old_slot = NULL;
2082 : 377 : memset(entry->exprstate, 0, sizeof(entry->exprstate));
2083 : 377 : entry->entry_cxt = NULL;
2084 : 377 : entry->publish_as_relid = InvalidOid;
2085 : 377 : entry->columns = NULL;
2086 : 377 : entry->attrmap = NULL;
2087 : : }
2088 : :
2089 : : /* Validate the entry */
2090 [ + + ]: 204574 : if (!entry->replicate_valid)
2091 : : {
2092 : 495 : Oid schemaId = get_rel_namespace(relid);
2093 : 495 : List *pubids = GetRelationIncludedPublications(relid);
2094 : :
2095 : : /*
2096 : : * We don't acquire a lock on the namespace system table as we build
2097 : : * the cache entry using a historic snapshot and all the later changes
2098 : : * are absorbed while decoding WAL.
2099 : : */
2100 : 495 : List *schemaPubids = GetSchemaPublications(schemaId);
2101 : : ListCell *lc;
2102 : 495 : Oid publish_as_relid = relid;
2103 : 495 : int publish_ancestor_level = 0;
2104 : 495 : bool am_partition = get_rel_relispartition(relid);
2105 : 495 : char relkind = get_rel_relkind(relid);
2106 : 495 : List *rel_publications = NIL;
2107 : :
2108 : : /* Reload publications if needed before use. */
2109 [ + + ]: 495 : if (!publications_valid)
2110 : : {
2111 : 266 : MemoryContextReset(data->pubctx);
2112 : :
2113 : 266 : oldctx = MemoryContextSwitchTo(data->pubctx);
2114 : 266 : data->publications = LoadPublications(data->publication_names);
2115 : 266 : MemoryContextSwitchTo(oldctx);
2116 : 266 : publications_valid = true;
2117 : : }
2118 : :
2119 : : /*
2120 : : * Reset schema_sent status as the relation definition may have
2121 : : * changed. Also reset pubactions to empty in case rel was dropped
2122 : : * from a publication. Also free any objects that depended on the
2123 : : * earlier definition.
2124 : : */
2125 : 495 : entry->schema_sent = false;
2126 : 495 : entry->include_gencols_type = PUBLISH_GENCOLS_NONE;
2127 : 495 : list_free(entry->streamed_txns);
2128 : 495 : entry->streamed_txns = NIL;
2129 : 495 : bms_free(entry->columns);
2130 : 495 : entry->columns = NULL;
2131 : 495 : entry->pubactions.pubinsert = false;
2132 : 495 : entry->pubactions.pubupdate = false;
2133 : 495 : entry->pubactions.pubdelete = false;
2134 : 495 : entry->pubactions.pubtruncate = false;
2135 : :
2136 : : /*
2137 : : * Tuple slots cleanups. (Will be rebuilt later if needed).
2138 : : */
2139 [ + + ]: 495 : if (entry->old_slot)
2140 : : {
2141 : 63 : TupleDesc desc = entry->old_slot->tts_tupleDescriptor;
2142 : :
2143 : : Assert(desc->tdrefcount == -1);
2144 : :
2145 : 63 : ExecDropSingleTupleTableSlot(entry->old_slot);
2146 : :
2147 : : /*
2148 : : * ExecDropSingleTupleTableSlot() would not free the TupleDesc, so
2149 : : * do it now to avoid any leaks.
2150 : : */
2151 : 63 : FreeTupleDesc(desc);
2152 : : }
2153 [ + + ]: 495 : if (entry->new_slot)
2154 : : {
2155 : 63 : TupleDesc desc = entry->new_slot->tts_tupleDescriptor;
2156 : :
2157 : : Assert(desc->tdrefcount == -1);
2158 : :
2159 : 63 : ExecDropSingleTupleTableSlot(entry->new_slot);
2160 : :
2161 : : /*
2162 : : * ExecDropSingleTupleTableSlot() would not free the TupleDesc, so
2163 : : * do it now to avoid any leaks.
2164 : : */
2165 : 63 : FreeTupleDesc(desc);
2166 : : }
2167 : :
2168 : 495 : entry->old_slot = NULL;
2169 : 495 : entry->new_slot = NULL;
2170 : :
2171 [ + + ]: 495 : if (entry->attrmap)
2172 : 3 : free_attrmap(entry->attrmap);
2173 : 495 : entry->attrmap = NULL;
2174 : :
2175 : : /*
2176 : : * Row filter cache cleanups.
2177 : : */
2178 [ + + ]: 495 : if (entry->entry_cxt)
2179 : 63 : MemoryContextDelete(entry->entry_cxt);
2180 : :
2181 : 495 : entry->entry_cxt = NULL;
2182 : 495 : entry->estate = NULL;
2183 : 495 : memset(entry->exprstate, 0, sizeof(entry->exprstate));
2184 : :
2185 : : /*
2186 : : * Build publication cache. We can't use one provided by relcache as
2187 : : * relcache considers all publications that the given relation is in,
2188 : : * but here we only need to consider ones that the subscriber
2189 : : * requested.
2190 : : */
2191 [ + + + + : 1165 : foreach(lc, data->publications)
+ + ]
2192 : : {
2193 : 670 : Publication *pub = lfirst(lc);
2194 : 670 : bool publish = false;
2195 : :
2196 : : /*
2197 : : * Under what relid should we publish changes in this publication?
2198 : : * We'll use the top-most relid across all publications. Also
2199 : : * track the ancestor level for this publication.
2200 : : */
2201 : 670 : Oid pub_relid = relid;
2202 : 670 : int ancestor_level = 0;
2203 : :
2204 : : /*
2205 : : * If this is a FOR ALL TABLES publication, pick the partition
2206 : : * root and set the ancestor level accordingly.
2207 : : */
2208 [ + + ]: 670 : if (pub->alltables)
2209 : : {
2210 : 113 : List *exceptpubids = NIL;
2211 : :
2212 [ + + ]: 113 : if (am_partition)
2213 : : {
2214 : 37 : List *ancestors = get_partition_ancestors(relid);
2215 : 37 : Oid last_ancestor_relid = llast_oid(ancestors);
2216 : :
2217 : : /*
2218 : : * For a partition, changes are published via top-most
2219 : : * ancestor when pubviaroot is true, so populate pub_relid
2220 : : * accordingly.
2221 : : */
2222 [ + + ]: 37 : if (pub->pubviaroot)
2223 : : {
2224 : 23 : pub_relid = last_ancestor_relid;
2225 : 23 : ancestor_level = list_length(ancestors);
2226 : : }
2227 : :
2228 : : /*
2229 : : * Only the top-most ancestor can appear in the EXCEPT
2230 : : * clause. Therefore, for a partition, exclusion must be
2231 : : * evaluated at the top-most ancestor.
2232 : : */
2233 : 37 : exceptpubids = GetRelationExcludedPublications(last_ancestor_relid);
2234 : : }
2235 : : else
2236 : : {
2237 : : /*
2238 : : * For a regular table or a root partitioned table, check
2239 : : * exclusion on table itself.
2240 : : */
2241 : 76 : exceptpubids = GetRelationExcludedPublications(pub_relid);
2242 : : }
2243 : :
2244 [ + + ]: 113 : if (!list_member_oid(exceptpubids, pub->oid))
2245 : 98 : publish = true;
2246 : :
2247 : 113 : list_free(exceptpubids);
2248 : :
2249 [ + + ]: 113 : if (!publish)
2250 : 15 : continue;
2251 : : }
2252 : :
2253 [ + + ]: 655 : if (!publish)
2254 : : {
2255 : 557 : bool ancestor_published = false;
2256 : :
2257 : : /*
2258 : : * For a partition, check if any of the ancestors are
2259 : : * published. If so, note down the topmost ancestor that is
2260 : : * published via this publication, which will be used as the
2261 : : * relation via which to publish the partition's changes.
2262 : : */
2263 [ + + ]: 557 : if (am_partition)
2264 : : {
2265 : : Oid ancestor;
2266 : : int level;
2267 : 121 : List *ancestors = get_partition_ancestors(relid);
2268 : :
2269 : 121 : ancestor = GetTopMostAncestorInPublication(pub->oid,
2270 : : ancestors,
2271 : : &level);
2272 : :
2273 [ + + ]: 121 : if (ancestor != InvalidOid)
2274 : : {
2275 : 48 : ancestor_published = true;
2276 [ + + ]: 48 : if (pub->pubviaroot)
2277 : : {
2278 : 25 : pub_relid = ancestor;
2279 : 25 : ancestor_level = level;
2280 : : }
2281 : : }
2282 : : }
2283 : :
2284 [ + + + + ]: 873 : if (list_member_oid(pubids, pub->oid) ||
2285 [ + + ]: 625 : list_member_oid(schemaPubids, pub->oid) ||
2286 : : ancestor_published)
2287 : 276 : publish = true;
2288 : : }
2289 : :
2290 : : /*
2291 : : * If the relation is to be published, determine actions to
2292 : : * publish, and list of columns, if appropriate.
2293 : : *
2294 : : * Don't publish changes for partitioned tables, because
2295 : : * publishing those of its partitions suffices, unless partition
2296 : : * changes won't be published due to pubviaroot being set.
2297 : : */
2298 [ + + + + ]: 655 : if (publish &&
2299 [ + + ]: 4 : (relkind != RELKIND_PARTITIONED_TABLE || pub->pubviaroot))
2300 : : {
2301 : 371 : entry->pubactions.pubinsert |= pub->pubactions.pubinsert;
2302 : 371 : entry->pubactions.pubupdate |= pub->pubactions.pubupdate;
2303 : 371 : entry->pubactions.pubdelete |= pub->pubactions.pubdelete;
2304 : 371 : entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
2305 : :
2306 : : /*
2307 : : * We want to publish the changes as the top-most ancestor
2308 : : * across all publications. So we need to check if the already
2309 : : * calculated level is higher than the new one. If yes, we can
2310 : : * ignore the new value (as it's a child). Otherwise the new
2311 : : * value is an ancestor, so we keep it.
2312 : : */
2313 [ + + ]: 371 : if (publish_ancestor_level > ancestor_level)
2314 : 1 : continue;
2315 : :
2316 : : /*
2317 : : * If we found an ancestor higher up in the tree, discard the
2318 : : * list of publications through which we replicate it, and use
2319 : : * the new ancestor.
2320 : : */
2321 [ + + ]: 370 : if (publish_ancestor_level < ancestor_level)
2322 : : {
2323 : 44 : publish_as_relid = pub_relid;
2324 : 44 : publish_ancestor_level = ancestor_level;
2325 : :
2326 : : /* reset the publication list for this relation */
2327 : 44 : rel_publications = NIL;
2328 : : }
2329 : : else
2330 : : {
2331 : : /* Same ancestor level, has to be the same OID. */
2332 : : Assert(publish_as_relid == pub_relid);
2333 : : }
2334 : :
2335 : : /* Track publications for this ancestor. */
2336 : 370 : rel_publications = lappend(rel_publications, pub);
2337 : : }
2338 : : }
2339 : :
2340 : 495 : entry->publish_as_relid = publish_as_relid;
2341 : :
2342 : : /*
2343 : : * Initialize the tuple slot, map, and row filter. These are only used
2344 : : * when publishing inserts, updates, or deletes.
2345 : : */
2346 [ + + + - ]: 495 : if (entry->pubactions.pubinsert || entry->pubactions.pubupdate ||
2347 [ - + ]: 138 : entry->pubactions.pubdelete)
2348 : : {
2349 : : /* Initialize the tuple slot and map */
2350 : 357 : init_tuple_slot(data, relation, entry);
2351 : :
2352 : : /* Initialize the row filter */
2353 : 357 : pgoutput_row_filter_init(data, rel_publications, entry);
2354 : :
2355 : : /* Check whether to publish generated columns. */
2356 : 357 : check_and_init_gencol(data, rel_publications, entry);
2357 : :
2358 : : /* Initialize the column list */
2359 : 357 : pgoutput_column_list_init(data, rel_publications, entry);
2360 : : }
2361 : :
2362 : 494 : list_free(pubids);
2363 : 494 : list_free(schemaPubids);
2364 : 494 : list_free(rel_publications);
2365 : :
2366 : 494 : entry->replicate_valid = true;
2367 : : }
2368 : :
2369 : 204573 : return entry;
2370 : : }
2371 : :
2372 : : /*
2373 : : * Cleanup list of streamed transactions and update the schema_sent flag.
2374 : : *
2375 : : * When a streamed transaction commits or aborts, we need to remove the
2376 : : * toplevel XID from the schema cache. If the transaction aborted, the
2377 : : * subscriber will simply throw away the schema records we streamed, so
2378 : : * we don't need to do anything else.
2379 : : *
2380 : : * If the transaction is committed, the subscriber will update the relation
2381 : : * cache - so tweak the schema_sent flag accordingly.
2382 : : */
2383 : : static void
2384 : 71 : cleanup_rel_sync_cache(TransactionId xid, bool is_commit)
2385 : : {
2386 : : HASH_SEQ_STATUS hash_seq;
2387 : : RelationSyncEntry *entry;
2388 : :
2389 : : Assert(RelationSyncCache != NULL);
2390 : :
2391 : 71 : hash_seq_init(&hash_seq, RelationSyncCache);
2392 [ + + ]: 145 : while ((entry = hash_seq_search(&hash_seq)) != NULL)
2393 : : {
2394 : : /*
2395 : : * We can set the schema_sent flag for an entry that has committed xid
2396 : : * in the list as that ensures that the subscriber would have the
2397 : : * corresponding schema and we don't need to send it unless there is
2398 : : * any invalidation for that relation.
2399 : : */
2400 [ + + + - : 167 : foreach_xid(streamed_txn, entry->streamed_txns)
+ + ]
2401 : : {
2402 [ + + ]: 72 : if (xid == streamed_txn)
2403 : : {
2404 [ + + ]: 53 : if (is_commit)
2405 : 42 : entry->schema_sent = true;
2406 : :
2407 : 53 : entry->streamed_txns =
2408 : 53 : foreach_delete_current(entry->streamed_txns, streamed_txn);
2409 : 53 : break;
2410 : : }
2411 : : }
2412 : : }
2413 : 71 : }
2414 : :
2415 : : /*
2416 : : * Relcache invalidation callback
2417 : : */
2418 : : static void
2419 : 4699 : rel_sync_cache_relation_cb(Datum arg, Oid relid)
2420 : : {
2421 : : RelationSyncEntry *entry;
2422 : :
2423 : : /*
2424 : : * We can get here if the plugin was used in SQL interface as the
2425 : : * RelationSyncCache is destroyed when the decoding finishes, but there is
2426 : : * no way to unregister the relcache invalidation callback.
2427 : : */
2428 [ + + ]: 4699 : if (RelationSyncCache == NULL)
2429 : 32 : return;
2430 : :
2431 : : /*
2432 : : * Nobody keeps pointers to entries in this hash table around outside
2433 : : * logical decoding callback calls - but invalidation events can come in
2434 : : * *during* a callback if we do any syscache access in the callback.
2435 : : * Because of that we must mark the cache entry as invalid but not damage
2436 : : * any of its substructure here. The next get_rel_sync_entry() call will
2437 : : * rebuild it all.
2438 : : */
2439 [ + + ]: 4667 : if (OidIsValid(relid))
2440 : : {
2441 : : /*
2442 : : * Getting invalidations for relations that aren't in the table is
2443 : : * entirely normal. So we don't care if it's found or not.
2444 : : */
2445 : 4589 : entry = (RelationSyncEntry *) hash_search(RelationSyncCache, &relid,
2446 : : HASH_FIND, NULL);
2447 [ + + ]: 4589 : if (entry != NULL)
2448 : 803 : entry->replicate_valid = false;
2449 : : }
2450 : : else
2451 : : {
2452 : : /* Whole cache must be flushed. */
2453 : : HASH_SEQ_STATUS status;
2454 : :
2455 : 78 : hash_seq_init(&status, RelationSyncCache);
2456 [ + + ]: 150 : while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL)
2457 : : {
2458 : 72 : entry->replicate_valid = false;
2459 : : }
2460 : : }
2461 : : }
2462 : :
2463 : : /*
2464 : : * Publication relation/schema map syscache invalidation callback
2465 : : *
2466 : : * Called for invalidations on pg_namespace.
2467 : : */
2468 : : static void
2469 : 46 : rel_sync_cache_publication_cb(Datum arg, SysCacheIdentifier cacheid,
2470 : : uint32 hashvalue)
2471 : : {
2472 : : HASH_SEQ_STATUS status;
2473 : : RelationSyncEntry *entry;
2474 : :
2475 : : /*
2476 : : * We can get here if the plugin was used in SQL interface as the
2477 : : * RelationSyncCache is destroyed when the decoding finishes, but there is
2478 : : * no way to unregister the invalidation callbacks.
2479 : : */
2480 [ + + ]: 46 : if (RelationSyncCache == NULL)
2481 : 13 : return;
2482 : :
2483 : : /*
2484 : : * We have no easy way to identify which cache entries this invalidation
2485 : : * event might have affected, so just mark them all invalid.
2486 : : */
2487 : 33 : hash_seq_init(&status, RelationSyncCache);
2488 [ + + ]: 58 : while ((entry = (RelationSyncEntry *) hash_seq_search(&status)) != NULL)
2489 : : {
2490 : 25 : entry->replicate_valid = false;
2491 : : }
2492 : : }
2493 : :
2494 : : /* Send Replication origin */
2495 : : static void
2496 : 1155 : send_repl_origin(LogicalDecodingContext *ctx, ReplOriginId origin_id,
2497 : : XLogRecPtr origin_lsn, bool send_origin)
2498 : : {
2499 [ + + ]: 1155 : if (send_origin)
2500 : : {
2501 : : char *origin;
2502 : :
2503 : : /*----------
2504 : : * XXX: which behaviour do we want here?
2505 : : *
2506 : : * Alternatives:
2507 : : * - don't send origin message if origin name not found
2508 : : * (that's what we do now)
2509 : : * - throw error - that will break replication, not good
2510 : : * - send some special "unknown" origin
2511 : : *----------
2512 : : */
2513 [ + - ]: 8 : if (replorigin_by_oid(origin_id, true, &origin))
2514 : : {
2515 : : /* Message boundary */
2516 : 8 : OutputPluginWrite(ctx, false);
2517 : 8 : OutputPluginPrepareWrite(ctx, true);
2518 : :
2519 : 8 : logicalrep_write_origin(ctx->out, origin, origin_lsn);
2520 : : }
2521 : : }
2522 : 1155 : }
|
