Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * logical.c
3 : : * PostgreSQL logical decoding coordination
4 : : *
5 : : * Copyright (c) 2012-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/logical.c
9 : : *
10 : : * NOTES
11 : : * This file coordinates interaction between the various modules that
12 : : * together provide logical decoding, primarily by providing so
13 : : * called LogicalDecodingContexts. The goal is to encapsulate most of the
14 : : * internal complexity for consumers of logical decoding, so they can
15 : : * create and consume a changestream with a low amount of code. Builtin
16 : : * consumers are the walsender and SQL SRF interface, but it's possible to
17 : : * add further ones without changing core code, e.g. to consume changes in
18 : : * a bgworker.
19 : : *
20 : : * The idea is that a consumer provides three callbacks, one to read WAL,
21 : : * one to prepare a data write, and a final one for actually writing since
22 : : * their implementation depends on the type of consumer. Check
23 : : * logicalfuncs.c for an example implementation of a fairly simple consumer
24 : : * and an implementation of a WAL reading callback that's suitable for
25 : : * simple consumers.
26 : : *-------------------------------------------------------------------------
27 : : */
28 : :
29 : : #include "postgres.h"
30 : :
31 : : #include "access/xact.h"
32 : : #include "access/xlog_internal.h"
33 : : #include "access/xlogutils.h"
34 : : #include "commands/repack.h"
35 : : #include "fmgr.h"
36 : : #include "miscadmin.h"
37 : : #include "pgstat.h"
38 : : #include "replication/decode.h"
39 : : #include "replication/logical.h"
40 : : #include "replication/reorderbuffer.h"
41 : : #include "replication/slotsync.h"
42 : : #include "replication/snapbuild.h"
43 : : #include "storage/proc.h"
44 : : #include "storage/procarray.h"
45 : : #include "utils/builtins.h"
46 : : #include "utils/guc.h"
47 : : #include "utils/guc_hooks.h"
48 : : #include "utils/injection_point.h"
49 : : #include "utils/inval.h"
50 : : #include "utils/memutils.h"
51 : : #include "utils/varlena.h"
52 : :
53 : : /* data for errcontext callback */
54 : : typedef struct LogicalErrorCallbackState
55 : : {
56 : : LogicalDecodingContext *ctx;
57 : : const char *callback_name;
58 : : XLogRecPtr report_location;
59 : : } LogicalErrorCallbackState;
60 : :
61 : : /* GUC variables */
62 : : char *output_plugin_libraries_string;
63 : :
64 : : /* wrappers around output plugin callbacks */
65 : : static void output_plugin_error_callback(void *arg);
66 : : static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
67 : : bool is_init);
68 : : static void shutdown_cb_wrapper(LogicalDecodingContext *ctx);
69 : : static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn);
70 : : static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
71 : : XLogRecPtr commit_lsn);
72 : : static void begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn);
73 : : static void prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
74 : : XLogRecPtr prepare_lsn);
75 : : static void commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
76 : : XLogRecPtr commit_lsn);
77 : : static void rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
78 : : XLogRecPtr prepare_end_lsn, TimestampTz prepare_time);
79 : : static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
80 : : Relation relation, ReorderBufferChange *change);
81 : : static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
82 : : int nrelations, Relation relations[], ReorderBufferChange *change);
83 : : static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
84 : : XLogRecPtr message_lsn, bool transactional,
85 : : const char *prefix, Size message_size, const char *message);
86 : :
87 : : /* streaming callbacks */
88 : : static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
89 : : XLogRecPtr first_lsn);
90 : : static void stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
91 : : XLogRecPtr last_lsn);
92 : : static void stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
93 : : XLogRecPtr abort_lsn);
94 : : static void stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
95 : : XLogRecPtr prepare_lsn);
96 : : static void stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
97 : : XLogRecPtr commit_lsn);
98 : : static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
99 : : Relation relation, ReorderBufferChange *change);
100 : : static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
101 : : XLogRecPtr message_lsn, bool transactional,
102 : : const char *prefix, Size message_size, const char *message);
103 : : static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
104 : : int nrelations, Relation relations[], ReorderBufferChange *change);
105 : :
106 : : /* callback to update txn's progress */
107 : : static void update_progress_txn_cb_wrapper(ReorderBuffer *cache,
108 : : ReorderBufferTXN *txn,
109 : : XLogRecPtr lsn);
110 : :
111 : : static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin);
112 : :
113 : : /* check_hook: validate new output_plugin_libraries value */
114 : : bool
115 : 1392 : check_output_plugin_libraries(char **newval, void **extra, GucSource source)
116 : : {
117 : : char *copy;
118 : : List *components;
119 : 1392 : bool ok = true;
120 : :
121 : 1392 : copy = guc_strdup(LOG, *newval);
122 [ - + ]: 1392 : if (!copy)
123 : 0 : return false;
124 : :
125 : : /*
126 : : * XXX SplitGUCList won't respect guc_malloc requirements, but this is
127 : : * consistent with other check_hook implementations...
128 : : */
129 [ + + ]: 1392 : if (!SplitGUCList(copy, ',', &components))
130 : : {
131 : 4 : GUC_check_errdetail("List syntax is invalid.");
132 : 4 : ok = false;
133 : : }
134 : :
135 : : /*
136 : : * Like with related GUC_LIST_QUOTE variables, we check only syntax here
137 : : * and not the existence of the plugins themselves.
138 : : */
139 : :
140 : 1392 : list_free(components);
141 : 1392 : guc_free(copy);
142 : :
143 : 1392 : return ok;
144 : : }
145 : :
146 : : /*
147 : : * Make sure the current settings & environment are capable of doing logical
148 : : * decoding.
149 : : */
150 : : void
151 : 1869 : CheckLogicalDecodingRequirements(bool repack)
152 : : {
153 : 1869 : CheckSlotRequirements(repack);
154 : :
155 : : /*
156 : : * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
157 : : * needs the same check.
158 : : */
159 : :
160 [ + + ]: 1869 : if (MyDatabaseId == InvalidOid)
161 [ + - ]: 1 : ereport(ERROR,
162 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
163 : : errmsg("logical decoding requires a database connection")));
164 : :
165 : : /* CheckSlotRequirements() has already checked if wal_level >= 'replica' */
166 : : Assert(wal_level >= WAL_LEVEL_REPLICA);
167 : :
168 : : /* Check if logical decoding is available on standby */
169 [ + + + + ]: 1868 : if (RecoveryInProgress() && !IsLogicalDecodingEnabled())
170 [ + - ]: 2 : ereport(ERROR,
171 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
172 : : errmsg("logical decoding on standby requires \"effective_wal_level\" >= \"logical\" on the primary"),
173 : : errhint("Set \"wal_level\" >= \"logical\" or create at least one logical slot when \"wal_level\" = \"replica\".")));
174 : 1866 : }
175 : :
176 : : /*
177 : : * Helper function for CreateInitDecodingContext() and
178 : : * CreateDecodingContext() performing common tasks.
179 : : */
180 : : static LogicalDecodingContext *
181 : 1315 : StartupDecodingContext(List *output_plugin_options,
182 : : XLogRecPtr start_lsn,
183 : : TransactionId xmin_horizon,
184 : : bool need_full_snapshot,
185 : : bool fast_forward,
186 : : bool in_create,
187 : : bool for_repack,
188 : : XLogReaderRoutine *xl_routine,
189 : : LogicalOutputPluginWriterPrepareWrite prepare_write,
190 : : LogicalOutputPluginWriterWrite do_write,
191 : : LogicalOutputPluginWriterUpdateProgress update_progress)
192 : : {
193 : : ReplicationSlot *slot;
194 : : MemoryContext context,
195 : : old_context;
196 : : LogicalDecodingContext *ctx;
197 : :
198 : : /* shorter lines... */
199 : 1315 : slot = MyReplicationSlot;
200 : :
201 : 1315 : context = AllocSetContextCreate(CurrentMemoryContext,
202 : : "Logical decoding context",
203 : : ALLOCSET_DEFAULT_SIZES);
204 : 1315 : old_context = MemoryContextSwitchTo(context);
205 : 1315 : ctx = palloc0_object(LogicalDecodingContext);
206 : :
207 : 1315 : ctx->context = context;
208 : :
209 : : /*
210 : : * (re-)load output plugins, so we detect a bad (removed) output plugin
211 : : * now.
212 : : */
213 [ + + ]: 1315 : if (!fast_forward)
214 : : {
215 : : /*
216 : : * Before loading this library, make sure it's been blessed for
217 : : * logical decoding.
218 : : */
219 : 1289 : const char *plugin = NameStr(slot->data.plugin);
220 : 1289 : bool plugin_allowed = false;
221 : :
222 [ + + ]: 1289 : if (strcmp(plugin, "pgrepack") == 0)
223 : : {
224 : : /*
225 : : * REPACK is a special case -- print a more helpful error message
226 : : * here if we're not being called in the correct context.
227 : : */
228 [ + + - + ]: 14 : if (!for_repack || !AmRepackWorker())
229 [ + - ]: 1 : ereport(ERROR,
230 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
231 : : errmsg("unsupported use of logical decoding plugin \"%s\"",
232 : : "pgrepack"),
233 : : errdetail("This plugin can only be used by %s.",
234 : : "REPACK (CONCURRENTLY)"));
235 : :
236 : 13 : plugin_allowed = true;
237 : : }
238 [ + - ]: 1275 : else if (output_plugin_libraries_string &&
239 [ + - ]: 1275 : output_plugin_libraries_string[0])
240 : : {
241 : : /* Check this plugin against output_plugin_libraries. */
242 : : char *rawstring;
243 : 1275 : List *elemlist = NIL;
244 : :
245 : : /* Need a modifiable copy */
246 : 1275 : rawstring = pstrdup(output_plugin_libraries_string);
247 : :
248 : : /* The check_hook should have already handled syntax errors. */
249 [ - + ]: 1275 : if (!SplitGUCList(rawstring, ',', &elemlist))
250 [ # # ]: 0 : elog(ERROR, "internal error: invalid output_plugin_libraries syntax after check_hook");
251 : :
252 [ + - + + : 2947 : foreach_ptr(char, allowed, elemlist)
+ + ]
253 : : {
254 [ + + ]: 1669 : if (strcmp(allowed, plugin) == 0)
255 : : {
256 : 1272 : plugin_allowed = true;
257 : 1272 : break;
258 : : }
259 : : }
260 : :
261 : 1275 : list_free(elemlist);
262 : 1275 : pfree(rawstring);
263 : : }
264 : :
265 [ + + ]: 1288 : if (!plugin_allowed)
266 : : {
267 : : /*
268 : : * Use the same error message as check_restricted_library_name(),
269 : : * but provide additional context for the DBA in the logs. (The
270 : : * HINT will be sent to the client, but that's not a secret.)
271 : : */
272 [ + - ]: 3 : ereport(ERROR,
273 : : errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
274 : : errmsg("library \"%s\" may not be used as an output plugin",
275 : : plugin),
276 : : /*- translator: The second %s is the value of the output_plugin_libraries GUC, which may contain whitespace, commas, and double-quotes (") */
277 : : errdetail_log("The configuration parameter \"%s\" (currently '%s') does not name this library as a trusted output plugin.",
278 : : "output_plugin_libraries",
279 : : output_plugin_libraries_string),
280 : : errhint("If it is safe for all REPLICATION users to use this library as an output plugin, add it to \"%s\" and reload the server configuration.",
281 : : "output_plugin_libraries"));
282 : : }
283 : :
284 : 1285 : LoadOutputPlugin(&ctx->callbacks, plugin);
285 : : }
286 : :
287 : : /*
288 : : * Now that the slot's xmin has been set, we can announce ourselves as a
289 : : * logical decoding backend which doesn't need to be checked individually
290 : : * when computing the xmin horizon because the xmin is enforced via
291 : : * replication slots.
292 : : *
293 : : * We can only do so if we're outside of a transaction (i.e. the case when
294 : : * streaming changes via walsender), otherwise an already setup
295 : : * snapshot/xid would end up being ignored. That's not a particularly
296 : : * bothersome restriction since the SQL interface can't be used for
297 : : * streaming anyway.
298 : : */
299 [ + + ]: 1310 : if (!IsTransactionOrTransactionBlock())
300 : : {
301 : 667 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
302 : 667 : MyProc->statusFlags |= PROC_IN_LOGICAL_DECODING;
303 : 667 : ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
304 : 667 : LWLockRelease(ProcArrayLock);
305 : : }
306 : :
307 : 1310 : ctx->slot = slot;
308 : :
309 : 1310 : ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, xl_routine, ctx);
310 [ - + ]: 1310 : if (!ctx->reader)
311 [ # # ]: 0 : ereport(ERROR,
312 : : (errcode(ERRCODE_OUT_OF_MEMORY),
313 : : errmsg("out of memory"),
314 : : errdetail("Failed while allocating a WAL reading processor.")));
315 : :
316 : 1310 : ctx->reorder = ReorderBufferAllocate();
317 : 1310 : ctx->snapshot_builder =
318 : 1310 : AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
319 : : need_full_snapshot, in_create, slot->data.two_phase_at);
320 : :
321 : 1310 : ctx->reorder->private_data = ctx;
322 : :
323 : : /* wrap output plugin callbacks, so we can add error context information */
324 : 1310 : ctx->reorder->begin = begin_cb_wrapper;
325 : 1310 : ctx->reorder->apply_change = change_cb_wrapper;
326 : 1310 : ctx->reorder->apply_truncate = truncate_cb_wrapper;
327 : 1310 : ctx->reorder->commit = commit_cb_wrapper;
328 : 1310 : ctx->reorder->message = message_cb_wrapper;
329 : :
330 : : /*
331 : : * To support streaming, we require start/stop/abort/commit/change
332 : : * callbacks. The message and truncate callbacks are optional, similar to
333 : : * regular output plugins. We however enable streaming when at least one
334 : : * of the methods is enabled so that we can easily identify missing
335 : : * methods.
336 : : *
337 : : * We decide it here, but only check it later in the wrappers.
338 : : */
339 : 2659 : ctx->streaming = (ctx->callbacks.stream_start_cb != NULL) ||
340 [ + - ]: 39 : (ctx->callbacks.stream_stop_cb != NULL) ||
341 [ + - ]: 39 : (ctx->callbacks.stream_abort_cb != NULL) ||
342 [ + - ]: 39 : (ctx->callbacks.stream_commit_cb != NULL) ||
343 [ + - ]: 39 : (ctx->callbacks.stream_change_cb != NULL) ||
344 [ + + + - ]: 1388 : (ctx->callbacks.stream_message_cb != NULL) ||
345 [ - + ]: 39 : (ctx->callbacks.stream_truncate_cb != NULL);
346 : :
347 : : /*
348 : : * streaming callbacks
349 : : *
350 : : * stream_message and stream_truncate callbacks are optional, so we do not
351 : : * fail with ERROR when missing, but the wrappers simply do nothing. We
352 : : * must set the ReorderBuffer callbacks to something, otherwise the calls
353 : : * from there will crash (we don't want to move the checks there).
354 : : */
355 : 1310 : ctx->reorder->stream_start = stream_start_cb_wrapper;
356 : 1310 : ctx->reorder->stream_stop = stream_stop_cb_wrapper;
357 : 1310 : ctx->reorder->stream_abort = stream_abort_cb_wrapper;
358 : 1310 : ctx->reorder->stream_prepare = stream_prepare_cb_wrapper;
359 : 1310 : ctx->reorder->stream_commit = stream_commit_cb_wrapper;
360 : 1310 : ctx->reorder->stream_change = stream_change_cb_wrapper;
361 : 1310 : ctx->reorder->stream_message = stream_message_cb_wrapper;
362 : 1310 : ctx->reorder->stream_truncate = stream_truncate_cb_wrapper;
363 : :
364 : :
365 : : /*
366 : : * To support two-phase logical decoding, we require
367 : : * begin_prepare/prepare/commit-prepare/abort-prepare callbacks. The
368 : : * filter_prepare callback is optional. We however enable two-phase
369 : : * logical decoding when at least one of the methods is enabled so that we
370 : : * can easily identify missing methods.
371 : : *
372 : : * We decide it here, but only check it later in the wrappers.
373 : : */
374 : 2659 : ctx->twophase = (ctx->callbacks.begin_prepare_cb != NULL) ||
375 [ + - ]: 39 : (ctx->callbacks.prepare_cb != NULL) ||
376 [ + - ]: 39 : (ctx->callbacks.commit_prepared_cb != NULL) ||
377 [ + - ]: 39 : (ctx->callbacks.rollback_prepared_cb != NULL) ||
378 [ + + + - ]: 1388 : (ctx->callbacks.stream_prepare_cb != NULL) ||
379 [ - + ]: 39 : (ctx->callbacks.filter_prepare_cb != NULL);
380 : :
381 : : /*
382 : : * Callback to support decoding at prepare time.
383 : : */
384 : 1310 : ctx->reorder->begin_prepare = begin_prepare_cb_wrapper;
385 : 1310 : ctx->reorder->prepare = prepare_cb_wrapper;
386 : 1310 : ctx->reorder->commit_prepared = commit_prepared_cb_wrapper;
387 : 1310 : ctx->reorder->rollback_prepared = rollback_prepared_cb_wrapper;
388 : :
389 : : /*
390 : : * Callback to support updating progress during sending data of a
391 : : * transaction (and its subtransactions) to the output plugin.
392 : : */
393 : 1310 : ctx->reorder->update_progress_txn = update_progress_txn_cb_wrapper;
394 : :
395 : 1310 : ctx->out = makeStringInfo();
396 : 1310 : ctx->prepare_write = prepare_write;
397 : 1310 : ctx->write = do_write;
398 : 1310 : ctx->update_progress = update_progress;
399 : :
400 : 1310 : ctx->output_plugin_options = output_plugin_options;
401 : :
402 : 1310 : ctx->fast_forward = fast_forward;
403 : :
404 : 1310 : MemoryContextSwitchTo(old_context);
405 : :
406 : 1310 : return ctx;
407 : : }
408 : :
409 : : /*
410 : : * Create a new decoding context, for a new logical slot.
411 : : *
412 : : * plugin -- contains the name of the output plugin
413 : : * output_plugin_options -- contains options passed to the output plugin
414 : : * need_full_snapshot -- if true, must obtain a snapshot able to read all
415 : : * tables; if false, one that can read only catalogs is acceptable.
416 : : * for_repack -- if true, we're going to be decoding for REPACK.
417 : : * restart_lsn -- if given as invalid, it's this routine's responsibility to
418 : : * mark WAL as reserved by setting a convenient restart_lsn for the slot.
419 : : * Otherwise, we set for decoding to start from the given LSN without
420 : : * marking WAL reserved beforehand. In that scenario, it's up to the
421 : : * caller to guarantee that WAL remains available.
422 : : * xl_routine -- XLogReaderRoutine for underlying XLogReader
423 : : * prepare_write, do_write, update_progress --
424 : : * callbacks that perform the use-case dependent, actual, work.
425 : : *
426 : : * Needs to be called while in a memory context that's at least as long lived
427 : : * as the decoding context because further memory contexts will be created
428 : : * inside it.
429 : : *
430 : : * Returns an initialized decoding context after calling the output plugin's
431 : : * startup function.
432 : : */
433 : : LogicalDecodingContext *
434 : 553 : CreateInitDecodingContext(const char *plugin,
435 : : List *output_plugin_options,
436 : : bool need_full_snapshot,
437 : : bool for_repack,
438 : : XLogRecPtr restart_lsn,
439 : : XLogReaderRoutine *xl_routine,
440 : : LogicalOutputPluginWriterPrepareWrite prepare_write,
441 : : LogicalOutputPluginWriterWrite do_write,
442 : : LogicalOutputPluginWriterUpdateProgress update_progress)
443 : : {
444 : 553 : TransactionId xmin_horizon = InvalidTransactionId;
445 : : ReplicationSlot *slot;
446 : : NameData plugin_name;
447 : : LogicalDecodingContext *ctx;
448 : : MemoryContext old_context;
449 : :
450 : : /*
451 : : * On a standby, this check is also required while creating the slot.
452 : : * Check the comments in the function.
453 : : */
454 : 553 : CheckLogicalDecodingRequirements(for_repack);
455 : :
456 : : /* shorter lines... */
457 : 553 : slot = MyReplicationSlot;
458 : :
459 : : /* first some sanity checks that are unlikely to be violated */
460 [ - + ]: 553 : if (slot == NULL)
461 [ # # ]: 0 : elog(ERROR, "cannot perform logical decoding without an acquired slot");
462 : :
463 [ - + ]: 553 : if (plugin == NULL)
464 [ # # ]: 0 : elog(ERROR, "cannot initialize logical decoding without a specified plugin");
465 : :
466 : : /* Make sure the passed slot is suitable. These are user facing errors. */
467 [ - + ]: 553 : if (SlotIsPhysical(slot))
468 [ # # ]: 0 : ereport(ERROR,
469 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
470 : : errmsg("cannot use physical replication slot for logical decoding")));
471 : :
472 [ - + ]: 553 : if (slot->data.database != MyDatabaseId)
473 [ # # ]: 0 : ereport(ERROR,
474 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
475 : : errmsg("replication slot \"%s\" was not created in this database",
476 : : NameStr(slot->data.name))));
477 : :
478 [ + + + + ]: 946 : if (IsTransactionState() &&
479 : 393 : GetTopTransactionIdIfAny() != InvalidTransactionId)
480 [ + - ]: 2 : ereport(ERROR,
481 : : (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
482 : : errmsg("cannot create logical replication slot in transaction that has performed writes")));
483 : :
484 : : /*
485 : : * Register output plugin name with slot. We need the mutex to avoid
486 : : * concurrent reading of a partially copied string. But we don't want any
487 : : * complicated code while holding a spinlock, so do namestrcpy() outside.
488 : : */
489 : 551 : namestrcpy(&plugin_name, plugin);
490 : 551 : SpinLockAcquire(&slot->mutex);
491 : 551 : slot->data.plugin = plugin_name;
492 : 551 : SpinLockRelease(&slot->mutex);
493 : :
494 [ + + ]: 551 : if (!XLogRecPtrIsValid(restart_lsn))
495 : 544 : ReplicationSlotReserveWal();
496 : : else
497 : : {
498 : 7 : SpinLockAcquire(&slot->mutex);
499 : 7 : slot->data.restart_lsn = restart_lsn;
500 : 7 : SpinLockRelease(&slot->mutex);
501 : : }
502 : :
503 : : /* ----
504 : : * This is a bit tricky: We need to determine a safe xmin horizon to start
505 : : * decoding from, to avoid starting from a running xacts record referring
506 : : * to xids whose rows have been vacuumed or pruned
507 : : * already. GetOldestSafeDecodingTransactionId() returns such a value, but
508 : : * without further interlock its return value might immediately be out of
509 : : * date.
510 : : *
511 : : * So we have to acquire both the ReplicationSlotControlLock and the
512 : : * ProcArrayLock to prevent concurrent computation and update of new xmin
513 : : * horizons by other backends, get the safe decoding xid, and inform the
514 : : * slot machinery about the new limit. Once that's done both locks can be
515 : : * released as the slot machinery now is protecting against vacuum.
516 : : *
517 : : * Note that, temporarily, the data, not just the catalog, xmin has to be
518 : : * reserved if a data snapshot is to be exported. Otherwise the initial
519 : : * data snapshot created here is not guaranteed to be valid. After that
520 : : * the data xmin doesn't need to be managed anymore and the global xmin
521 : : * should be recomputed. As we are fine with losing the pegged data xmin
522 : : * after crash - no chance a snapshot would get exported anymore - we can
523 : : * get away with just setting the slot's
524 : : * effective_xmin. ReplicationSlotRelease will reset it again.
525 : : *
526 : : * ----
527 : : */
528 : 551 : LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
529 : 551 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
530 : :
531 : 551 : xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
532 : :
533 : 551 : SpinLockAcquire(&slot->mutex);
534 : 551 : slot->effective_catalog_xmin = xmin_horizon;
535 : 551 : slot->data.catalog_xmin = xmin_horizon;
536 [ + + ]: 551 : if (need_full_snapshot)
537 : 236 : slot->effective_xmin = xmin_horizon;
538 : 551 : SpinLockRelease(&slot->mutex);
539 : :
540 : 551 : ReplicationSlotsComputeRequiredXmin(true);
541 : :
542 : 551 : LWLockRelease(ProcArrayLock);
543 : 551 : LWLockRelease(ReplicationSlotControlLock);
544 : :
545 : 551 : ReplicationSlotMarkDirty();
546 : 551 : ReplicationSlotSave();
547 : :
548 : 551 : ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
549 : : need_full_snapshot, false, true, for_repack,
550 : : xl_routine, prepare_write, do_write,
551 : : update_progress);
552 : :
553 : : /* call output plugin initialization callback */
554 : 546 : old_context = MemoryContextSwitchTo(ctx->context);
555 [ + - ]: 546 : if (ctx->callbacks.startup_cb != NULL)
556 : 546 : startup_cb_wrapper(ctx, &ctx->options, true);
557 : 546 : MemoryContextSwitchTo(old_context);
558 : :
559 : : /*
560 : : * We allow decoding of prepared transactions when the two_phase is
561 : : * enabled at the time of slot creation, or when the two_phase option is
562 : : * given at the streaming start, provided the plugin supports all the
563 : : * callbacks for two-phase.
564 : : */
565 : 546 : ctx->twophase &= slot->data.two_phase;
566 : :
567 : 546 : ctx->reorder->output_rewrites = ctx->options.receive_rewrites;
568 : :
569 : 546 : return ctx;
570 : : }
571 : :
572 : : /*
573 : : * Create a new decoding context, for a logical slot that has previously been
574 : : * used already.
575 : : *
576 : : * start_lsn
577 : : * The LSN at which to start decoding. If InvalidXLogRecPtr, restart
578 : : * from the slot's confirmed_flush; otherwise, start from the specified
579 : : * location (but move it forwards to confirmed_flush if it's older than
580 : : * that, see below).
581 : : *
582 : : * output_plugin_options
583 : : * options passed to the output plugin.
584 : : *
585 : : * fast_forward
586 : : * bypass the generation of logical changes.
587 : : *
588 : : * xl_routine
589 : : * XLogReaderRoutine used by underlying xlogreader
590 : : *
591 : : * prepare_write, do_write, update_progress
592 : : * callbacks that have to be filled to perform the use-case dependent,
593 : : * actual work.
594 : : *
595 : : * Needs to be called while in a memory context that's at least as long lived
596 : : * as the decoding context because further memory contexts will be created
597 : : * inside it.
598 : : *
599 : : * Returns an initialized decoding context after calling the output plugin's
600 : : * startup function.
601 : : */
602 : : LogicalDecodingContext *
603 : 769 : CreateDecodingContext(XLogRecPtr start_lsn,
604 : : List *output_plugin_options,
605 : : bool fast_forward,
606 : : XLogReaderRoutine *xl_routine,
607 : : LogicalOutputPluginWriterPrepareWrite prepare_write,
608 : : LogicalOutputPluginWriterWrite do_write,
609 : : LogicalOutputPluginWriterUpdateProgress update_progress)
610 : : {
611 : : LogicalDecodingContext *ctx;
612 : : ReplicationSlot *slot;
613 : : MemoryContext old_context;
614 : :
615 : : /* shorter lines... */
616 : 769 : slot = MyReplicationSlot;
617 : :
618 : : /* first some sanity checks that are unlikely to be violated */
619 [ - + ]: 769 : if (slot == NULL)
620 [ # # ]: 0 : elog(ERROR, "cannot perform logical decoding without an acquired slot");
621 : :
622 : : /* make sure the passed slot is suitable, these are user facing errors */
623 [ + + ]: 769 : if (SlotIsPhysical(slot))
624 [ + - ]: 1 : ereport(ERROR,
625 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
626 : : errmsg("cannot use physical replication slot for logical decoding")));
627 : :
628 : : /*
629 : : * We need to access the system tables during decoding to build the
630 : : * logical changes unless we are in fast_forward mode where no changes are
631 : : * generated.
632 : : */
633 [ + + + + ]: 768 : if (slot->data.database != MyDatabaseId && !fast_forward)
634 [ + - ]: 3 : ereport(ERROR,
635 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
636 : : errmsg("replication slot \"%s\" was not created in this database",
637 : : NameStr(slot->data.name))));
638 : :
639 : : /*
640 : : * The slots being synced from the primary can't be used for decoding as
641 : : * they are used after failover. However, we do allow advancing the LSNs
642 : : * during the synchronization of slots. See update_local_synced_slot.
643 : : */
644 [ + + + + : 765 : if (RecoveryInProgress() && slot->data.synced && !IsSyncingReplicationSlots())
+ + ]
645 [ + - ]: 1 : ereport(ERROR,
646 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
647 : : errmsg("cannot use replication slot \"%s\" for logical decoding",
648 : : NameStr(slot->data.name)),
649 : : errdetail("This replication slot is being synchronized from the primary server."),
650 : : errhint("Specify another replication slot."));
651 : :
652 : : /* slot must be valid to allow decoding */
653 : : Assert(slot->data.invalidated == RS_INVAL_NONE);
654 : : Assert(XLogRecPtrIsValid(slot->data.restart_lsn));
655 : :
656 [ + + ]: 764 : if (!XLogRecPtrIsValid(start_lsn))
657 : : {
658 : : /* continue from last position */
659 : 421 : start_lsn = slot->data.confirmed_flush;
660 : : }
661 [ + + ]: 343 : else if (start_lsn < slot->data.confirmed_flush)
662 : : {
663 : : /*
664 : : * It might seem like we should error out in this case, but it's
665 : : * pretty common for a client to acknowledge a LSN it doesn't have to
666 : : * do anything for, and thus didn't store persistently, because the
667 : : * xlog records didn't result in anything relevant for logical
668 : : * decoding. Clients have to be able to do that to support synchronous
669 : : * replication.
670 : : *
671 : : * Starting at a different LSN than requested might not catch certain
672 : : * kinds of client errors; so the client may wish to check that
673 : : * confirmed_flush_lsn matches its expectations.
674 : : */
675 [ + - ]: 52 : elog(LOG, "%X/%08X has been already streamed, forwarding to %X/%08X",
676 : : LSN_FORMAT_ARGS(start_lsn),
677 : : LSN_FORMAT_ARGS(slot->data.confirmed_flush));
678 : :
679 : 52 : start_lsn = slot->data.confirmed_flush;
680 : : }
681 : :
682 : 764 : ctx = StartupDecodingContext(output_plugin_options,
683 : : start_lsn, InvalidTransactionId, false,
684 : : fast_forward, false, false,
685 : : xl_routine, prepare_write,
686 : : do_write, update_progress);
687 : :
688 : : /* call output plugin initialization callback */
689 : 764 : old_context = MemoryContextSwitchTo(ctx->context);
690 [ + + ]: 764 : if (ctx->callbacks.startup_cb != NULL)
691 : 738 : startup_cb_wrapper(ctx, &ctx->options, false);
692 : 758 : MemoryContextSwitchTo(old_context);
693 : :
694 : : /*
695 : : * We allow decoding of prepared transactions when the two_phase is
696 : : * enabled at the time of slot creation, or when the two_phase option is
697 : : * given at the streaming start, provided the plugin supports all the
698 : : * callbacks for two-phase.
699 : : */
700 [ + + + + ]: 758 : ctx->twophase &= (slot->data.two_phase || ctx->twophase_opt_given);
701 : :
702 : : /* Mark slot to allow two_phase decoding if not already marked */
703 [ + + + + ]: 758 : if (ctx->twophase && !slot->data.two_phase)
704 : : {
705 : 8 : SpinLockAcquire(&slot->mutex);
706 : 8 : slot->data.two_phase = true;
707 : 8 : slot->data.two_phase_at = start_lsn;
708 : 8 : SpinLockRelease(&slot->mutex);
709 : 8 : ReplicationSlotMarkDirty();
710 : 8 : ReplicationSlotSave();
711 : 8 : SnapBuildSetTwoPhaseAt(ctx->snapshot_builder, start_lsn);
712 : : }
713 : :
714 : 758 : ctx->reorder->output_rewrites = ctx->options.receive_rewrites;
715 : :
716 [ + + + + ]: 758 : ereport(LogicalDecodingLogLevel(),
717 : : (errmsg("starting logical decoding for slot \"%s\"",
718 : : NameStr(slot->data.name)),
719 : : errdetail("Streaming transactions committing after %X/%08X, reading WAL from %X/%08X.",
720 : : LSN_FORMAT_ARGS(slot->data.confirmed_flush),
721 : : LSN_FORMAT_ARGS(slot->data.restart_lsn))));
722 : :
723 : 758 : return ctx;
724 : : }
725 : :
726 : : /*
727 : : * Returns true if a consistent initial decoding snapshot has been built.
728 : : */
729 : : bool
730 : 3521 : DecodingContextReady(LogicalDecodingContext *ctx)
731 : : {
732 : 3521 : return SnapBuildCurrentState(ctx->snapshot_builder) == SNAPBUILD_CONSISTENT;
733 : : }
734 : :
735 : : /*
736 : : * Read from the decoding slot, until it is ready to start extracting changes.
737 : : */
738 : : void
739 : 539 : DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
740 : : {
741 : 539 : ReplicationSlot *slot = ctx->slot;
742 : :
743 : : /* Initialize from where to start reading WAL. */
744 : 539 : XLogBeginRead(ctx->reader, slot->data.restart_lsn);
745 : :
746 [ + + ]: 539 : elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%08X",
747 : : LSN_FORMAT_ARGS(slot->data.restart_lsn));
748 : :
749 : : /* Wait for a consistent starting point */
750 : : for (;;)
751 : 2973 : {
752 : : XLogRecord *record;
753 : 3512 : char *err = NULL;
754 : :
755 : : /* the read_page callback waits for new WAL */
756 : 3512 : record = XLogReadRecord(ctx->reader, &err);
757 [ - + ]: 3512 : if (err)
758 [ # # ]: 0 : elog(ERROR, "could not find logical decoding starting point: %s", err);
759 [ - + ]: 3512 : if (!record)
760 [ # # ]: 0 : elog(ERROR, "could not find logical decoding starting point");
761 : :
762 : 3512 : LogicalDecodingProcessRecord(ctx, ctx->reader);
763 : :
764 : : /* only continue till we found a consistent spot */
765 [ + + ]: 3510 : if (DecodingContextReady(ctx))
766 : 537 : break;
767 : :
768 [ - + ]: 2973 : CHECK_FOR_INTERRUPTS();
769 : : }
770 : :
771 : 537 : SpinLockAcquire(&slot->mutex);
772 : 537 : slot->data.confirmed_flush = ctx->reader->EndRecPtr;
773 [ + + ]: 537 : if (slot->data.two_phase)
774 : 9 : slot->data.two_phase_at = ctx->reader->EndRecPtr;
775 : 537 : SpinLockRelease(&slot->mutex);
776 : 537 : }
777 : :
778 : : /*
779 : : * Free a previously allocated decoding context, invoking the shutdown
780 : : * callback if necessary.
781 : : */
782 : : void
783 : 1012 : FreeDecodingContext(LogicalDecodingContext *ctx)
784 : : {
785 [ + + ]: 1012 : if (ctx->callbacks.shutdown_cb != NULL)
786 : 986 : shutdown_cb_wrapper(ctx);
787 : :
788 : 1012 : ReorderBufferFree(ctx->reorder);
789 : 1012 : FreeSnapshotBuilder(ctx->snapshot_builder);
790 : 1012 : XLogReaderFree(ctx->reader);
791 : 1012 : MemoryContextDelete(ctx->context);
792 : 1012 : }
793 : :
794 : : /*
795 : : * Prepare a write using the context's output routine.
796 : : */
797 : : void
798 : 357205 : OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
799 : : {
800 [ - + ]: 357205 : if (!ctx->accept_writes)
801 [ # # ]: 0 : elog(ERROR, "writes are only accepted in commit, begin and change callbacks");
802 : :
803 : 357205 : ctx->prepare_write(ctx, ctx->write_location, ctx->write_xid, last_write);
804 : 357205 : ctx->prepared_write = true;
805 : 357205 : }
806 : :
807 : : /*
808 : : * Perform a write using the context's output routine.
809 : : */
810 : : void
811 : 357205 : OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
812 : : {
813 [ - + ]: 357205 : if (!ctx->prepared_write)
814 [ # # ]: 0 : elog(ERROR, "OutputPluginPrepareWrite needs to be called before OutputPluginWrite");
815 : :
816 : 357205 : ctx->write(ctx, ctx->write_location, ctx->write_xid, last_write);
817 : 357190 : ctx->prepared_write = false;
818 : 357190 : }
819 : :
820 : : /*
821 : : * Update progress tracking (if supported).
822 : : */
823 : : void
824 : 5172 : OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx,
825 : : bool skipped_xact)
826 : : {
827 [ + + ]: 5172 : if (!ctx->update_progress)
828 : 1596 : return;
829 : :
830 : 3576 : ctx->update_progress(ctx, ctx->write_location, ctx->write_xid,
831 : : skipped_xact);
832 : : }
833 : :
834 : : /*
835 : : * Load the output plugin, lookup its output plugin init function, and check
836 : : * that it provides the required callbacks. The caller must have checked that
837 : : * the current user has the necessary privileges to load the given plugin;
838 : : * standard LOAD restrictions are not applied here.
839 : : */
840 : : static void
841 : 1285 : LoadOutputPlugin(OutputPluginCallbacks *callbacks, const char *plugin)
842 : : {
843 : : LogicalOutputPluginInit plugin_init;
844 : :
845 : 1284 : plugin_init = (LogicalOutputPluginInit)
846 : 1285 : load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
847 : :
848 [ - + ]: 1284 : if (plugin_init == NULL)
849 [ # # ]: 0 : elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
850 : :
851 : : /* ask the output plugin to fill the callback struct */
852 : 1284 : plugin_init(callbacks);
853 : :
854 [ - + ]: 1284 : if (callbacks->begin_cb == NULL)
855 [ # # ]: 0 : elog(ERROR, "output plugins have to register a begin callback");
856 [ - + ]: 1284 : if (callbacks->change_cb == NULL)
857 [ # # ]: 0 : elog(ERROR, "output plugins have to register a change callback");
858 [ - + ]: 1284 : if (callbacks->commit_cb == NULL)
859 [ # # ]: 0 : elog(ERROR, "output plugins have to register a commit callback");
860 : 1284 : }
861 : :
862 : : static void
863 : 730 : output_plugin_error_callback(void *arg)
864 : : {
865 : 730 : LogicalErrorCallbackState *state = (LogicalErrorCallbackState *) arg;
866 : :
867 : : /* not all callbacks have an associated LSN */
868 [ + + ]: 730 : if (XLogRecPtrIsValid(state->report_location))
869 : 724 : errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%08X",
870 : 724 : NameStr(state->ctx->slot->data.name),
871 : 724 : NameStr(state->ctx->slot->data.plugin),
872 : : state->callback_name,
873 : 724 : LSN_FORMAT_ARGS(state->report_location));
874 : : else
875 : 6 : errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
876 : 6 : NameStr(state->ctx->slot->data.name),
877 : 6 : NameStr(state->ctx->slot->data.plugin),
878 : : state->callback_name);
879 : 730 : }
880 : :
881 : : static void
882 : 1284 : startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
883 : : {
884 : : LogicalErrorCallbackState state;
885 : : ErrorContextCallback errcallback;
886 : :
887 : : Assert(!ctx->fast_forward);
888 : :
889 : : /* Push callback + info on the error context stack */
890 : 1284 : state.ctx = ctx;
891 : 1284 : state.callback_name = "startup";
892 : 1284 : state.report_location = InvalidXLogRecPtr;
893 : 1284 : errcallback.callback = output_plugin_error_callback;
894 : 1284 : errcallback.arg = &state;
895 : 1284 : errcallback.previous = error_context_stack;
896 : 1284 : error_context_stack = &errcallback;
897 : :
898 : : /* set output state */
899 : 1284 : ctx->accept_writes = false;
900 : 1284 : ctx->end_xact = false;
901 : :
902 : : /* do the actual work: call callback */
903 : 1284 : ctx->callbacks.startup_cb(ctx, opt, is_init);
904 : :
905 : : /* Pop the error context stack */
906 : 1278 : error_context_stack = errcallback.previous;
907 : 1278 : }
908 : :
909 : : static void
910 : 986 : shutdown_cb_wrapper(LogicalDecodingContext *ctx)
911 : : {
912 : : LogicalErrorCallbackState state;
913 : : ErrorContextCallback errcallback;
914 : :
915 : : Assert(!ctx->fast_forward);
916 : :
917 : : /* Push callback + info on the error context stack */
918 : 986 : state.ctx = ctx;
919 : 986 : state.callback_name = "shutdown";
920 : 986 : state.report_location = InvalidXLogRecPtr;
921 : 986 : errcallback.callback = output_plugin_error_callback;
922 : 986 : errcallback.arg = &state;
923 : 986 : errcallback.previous = error_context_stack;
924 : 986 : error_context_stack = &errcallback;
925 : :
926 : : /* set output state */
927 : 986 : ctx->accept_writes = false;
928 : 986 : ctx->end_xact = false;
929 : :
930 : : /* do the actual work: call callback */
931 : 986 : ctx->callbacks.shutdown_cb(ctx);
932 : :
933 : : /* Pop the error context stack */
934 : 986 : error_context_stack = errcallback.previous;
935 : 986 : }
936 : :
937 : :
938 : : /*
939 : : * Callbacks for ReorderBuffer which add in some more information and then call
940 : : * output_plugin.h plugins.
941 : : */
942 : : static void
943 : 2181 : begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
944 : : {
945 : 2181 : LogicalDecodingContext *ctx = cache->private_data;
946 : : LogicalErrorCallbackState state;
947 : : ErrorContextCallback errcallback;
948 : :
949 : : Assert(!ctx->fast_forward);
950 : :
951 : : /* Push callback + info on the error context stack */
952 : 2181 : state.ctx = ctx;
953 : 2181 : state.callback_name = "begin";
954 : 2181 : state.report_location = txn->first_lsn;
955 : 2181 : errcallback.callback = output_plugin_error_callback;
956 : 2181 : errcallback.arg = &state;
957 : 2181 : errcallback.previous = error_context_stack;
958 : 2181 : error_context_stack = &errcallback;
959 : :
960 : : /* set output state */
961 : 2181 : ctx->accept_writes = true;
962 : 2181 : ctx->write_xid = txn->xid;
963 : 2181 : ctx->write_location = txn->first_lsn;
964 : 2181 : ctx->end_xact = false;
965 : :
966 : : /* do the actual work: call callback */
967 : 2181 : ctx->callbacks.begin_cb(ctx, txn);
968 : :
969 : : /* Pop the error context stack */
970 : 2181 : error_context_stack = errcallback.previous;
971 : 2181 : }
972 : :
973 : : static void
974 : 2175 : commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
975 : : XLogRecPtr commit_lsn)
976 : : {
977 : 2175 : LogicalDecodingContext *ctx = cache->private_data;
978 : : LogicalErrorCallbackState state;
979 : : ErrorContextCallback errcallback;
980 : :
981 : : Assert(!ctx->fast_forward);
982 : :
983 : : /* Push callback + info on the error context stack */
984 : 2175 : state.ctx = ctx;
985 : 2175 : state.callback_name = "commit";
986 : 2175 : state.report_location = txn->final_lsn; /* beginning of commit record */
987 : 2175 : errcallback.callback = output_plugin_error_callback;
988 : 2175 : errcallback.arg = &state;
989 : 2175 : errcallback.previous = error_context_stack;
990 : 2175 : error_context_stack = &errcallback;
991 : :
992 : : /* set output state */
993 : 2175 : ctx->accept_writes = true;
994 : 2175 : ctx->write_xid = txn->xid;
995 : 2175 : ctx->write_location = txn->end_lsn; /* points to the end of the record */
996 : 2175 : ctx->end_xact = true;
997 : :
998 : : /* do the actual work: call callback */
999 : 2175 : ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
1000 : :
1001 : : /* Pop the error context stack */
1002 : 2164 : error_context_stack = errcallback.previous;
1003 : 2164 : }
1004 : :
1005 : : /*
1006 : : * The functionality of begin_prepare is quite similar to begin with the
1007 : : * exception that this will have gid (global transaction id) information which
1008 : : * can be used by plugin. Now, we thought about extending the existing begin
1009 : : * but that would break the replication protocol and additionally this looks
1010 : : * cleaner.
1011 : : */
1012 : : static void
1013 : 30 : begin_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
1014 : : {
1015 : 30 : LogicalDecodingContext *ctx = cache->private_data;
1016 : : LogicalErrorCallbackState state;
1017 : : ErrorContextCallback errcallback;
1018 : :
1019 : : Assert(!ctx->fast_forward);
1020 : :
1021 : : /* We're only supposed to call this when two-phase commits are supported */
1022 : : Assert(ctx->twophase);
1023 : :
1024 : : /* Push callback + info on the error context stack */
1025 : 30 : state.ctx = ctx;
1026 : 30 : state.callback_name = "begin_prepare";
1027 : 30 : state.report_location = txn->first_lsn;
1028 : 30 : errcallback.callback = output_plugin_error_callback;
1029 : 30 : errcallback.arg = &state;
1030 : 30 : errcallback.previous = error_context_stack;
1031 : 30 : error_context_stack = &errcallback;
1032 : :
1033 : : /* set output state */
1034 : 30 : ctx->accept_writes = true;
1035 : 30 : ctx->write_xid = txn->xid;
1036 : 30 : ctx->write_location = txn->first_lsn;
1037 : 30 : ctx->end_xact = false;
1038 : :
1039 : : /*
1040 : : * If the plugin supports two-phase commits then begin prepare callback is
1041 : : * mandatory
1042 : : */
1043 [ - + ]: 30 : if (ctx->callbacks.begin_prepare_cb == NULL)
1044 [ # # ]: 0 : ereport(ERROR,
1045 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1046 : : errmsg("logical replication at prepare time requires a %s callback",
1047 : : "begin_prepare_cb")));
1048 : :
1049 : : /* do the actual work: call callback */
1050 : 30 : ctx->callbacks.begin_prepare_cb(ctx, txn);
1051 : :
1052 : : /* Pop the error context stack */
1053 : 30 : error_context_stack = errcallback.previous;
1054 : 30 : }
1055 : :
1056 : : static void
1057 : 29 : prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1058 : : XLogRecPtr prepare_lsn)
1059 : : {
1060 : 29 : LogicalDecodingContext *ctx = cache->private_data;
1061 : : LogicalErrorCallbackState state;
1062 : : ErrorContextCallback errcallback;
1063 : :
1064 : : Assert(!ctx->fast_forward);
1065 : :
1066 : : /* We're only supposed to call this when two-phase commits are supported */
1067 : : Assert(ctx->twophase);
1068 : :
1069 : : /* Push callback + info on the error context stack */
1070 : 29 : state.ctx = ctx;
1071 : 29 : state.callback_name = "prepare";
1072 : 29 : state.report_location = txn->final_lsn; /* beginning of prepare record */
1073 : 29 : errcallback.callback = output_plugin_error_callback;
1074 : 29 : errcallback.arg = &state;
1075 : 29 : errcallback.previous = error_context_stack;
1076 : 29 : error_context_stack = &errcallback;
1077 : :
1078 : : /* set output state */
1079 : 29 : ctx->accept_writes = true;
1080 : 29 : ctx->write_xid = txn->xid;
1081 : 29 : ctx->write_location = txn->end_lsn; /* points to the end of the record */
1082 : 29 : ctx->end_xact = true;
1083 : :
1084 : : /*
1085 : : * If the plugin supports two-phase commits then prepare callback is
1086 : : * mandatory
1087 : : */
1088 [ - + ]: 29 : if (ctx->callbacks.prepare_cb == NULL)
1089 [ # # ]: 0 : ereport(ERROR,
1090 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1091 : : errmsg("logical replication at prepare time requires a %s callback",
1092 : : "prepare_cb")));
1093 : :
1094 : : /* do the actual work: call callback */
1095 : 29 : ctx->callbacks.prepare_cb(ctx, txn, prepare_lsn);
1096 : :
1097 : : /* Pop the error context stack */
1098 : 29 : error_context_stack = errcallback.previous;
1099 : 29 : }
1100 : :
1101 : : static void
1102 : 34 : commit_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1103 : : XLogRecPtr commit_lsn)
1104 : : {
1105 : 34 : LogicalDecodingContext *ctx = cache->private_data;
1106 : : LogicalErrorCallbackState state;
1107 : : ErrorContextCallback errcallback;
1108 : :
1109 : : Assert(!ctx->fast_forward);
1110 : :
1111 : : /* We're only supposed to call this when two-phase commits are supported */
1112 : : Assert(ctx->twophase);
1113 : :
1114 : : /* Push callback + info on the error context stack */
1115 : 34 : state.ctx = ctx;
1116 : 34 : state.callback_name = "commit_prepared";
1117 : 34 : state.report_location = txn->final_lsn; /* beginning of commit record */
1118 : 34 : errcallback.callback = output_plugin_error_callback;
1119 : 34 : errcallback.arg = &state;
1120 : 34 : errcallback.previous = error_context_stack;
1121 : 34 : error_context_stack = &errcallback;
1122 : :
1123 : : /* set output state */
1124 : 34 : ctx->accept_writes = true;
1125 : 34 : ctx->write_xid = txn->xid;
1126 : 34 : ctx->write_location = txn->end_lsn; /* points to the end of the record */
1127 : 34 : ctx->end_xact = true;
1128 : :
1129 : : /*
1130 : : * If the plugin support two-phase commits then commit prepared callback
1131 : : * is mandatory
1132 : : */
1133 [ - + ]: 34 : if (ctx->callbacks.commit_prepared_cb == NULL)
1134 [ # # ]: 0 : ereport(ERROR,
1135 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1136 : : errmsg("logical replication at prepare time requires a %s callback",
1137 : : "commit_prepared_cb")));
1138 : :
1139 : : /* do the actual work: call callback */
1140 : 34 : ctx->callbacks.commit_prepared_cb(ctx, txn, commit_lsn);
1141 : :
1142 : : /* Pop the error context stack */
1143 : 34 : error_context_stack = errcallback.previous;
1144 : 34 : }
1145 : :
1146 : : static void
1147 : 11 : rollback_prepared_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1148 : : XLogRecPtr prepare_end_lsn,
1149 : : TimestampTz prepare_time)
1150 : : {
1151 : 11 : LogicalDecodingContext *ctx = cache->private_data;
1152 : : LogicalErrorCallbackState state;
1153 : : ErrorContextCallback errcallback;
1154 : :
1155 : : Assert(!ctx->fast_forward);
1156 : :
1157 : : /* We're only supposed to call this when two-phase commits are supported */
1158 : : Assert(ctx->twophase);
1159 : :
1160 : : /* Push callback + info on the error context stack */
1161 : 11 : state.ctx = ctx;
1162 : 11 : state.callback_name = "rollback_prepared";
1163 : 11 : state.report_location = txn->final_lsn; /* beginning of commit record */
1164 : 11 : errcallback.callback = output_plugin_error_callback;
1165 : 11 : errcallback.arg = &state;
1166 : 11 : errcallback.previous = error_context_stack;
1167 : 11 : error_context_stack = &errcallback;
1168 : :
1169 : : /* set output state */
1170 : 11 : ctx->accept_writes = true;
1171 : 11 : ctx->write_xid = txn->xid;
1172 : 11 : ctx->write_location = txn->end_lsn; /* points to the end of the record */
1173 : 11 : ctx->end_xact = true;
1174 : :
1175 : : /*
1176 : : * If the plugin support two-phase commits then rollback prepared callback
1177 : : * is mandatory
1178 : : */
1179 [ - + ]: 11 : if (ctx->callbacks.rollback_prepared_cb == NULL)
1180 [ # # ]: 0 : ereport(ERROR,
1181 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1182 : : errmsg("logical replication at prepare time requires a %s callback",
1183 : : "rollback_prepared_cb")));
1184 : :
1185 : : /* do the actual work: call callback */
1186 : 11 : ctx->callbacks.rollback_prepared_cb(ctx, txn, prepare_end_lsn,
1187 : : prepare_time);
1188 : :
1189 : : /* Pop the error context stack */
1190 : 11 : error_context_stack = errcallback.previous;
1191 : 11 : }
1192 : :
1193 : : static void
1194 : 182752 : change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1195 : : Relation relation, ReorderBufferChange *change)
1196 : : {
1197 : 182752 : LogicalDecodingContext *ctx = cache->private_data;
1198 : : LogicalErrorCallbackState state;
1199 : : ErrorContextCallback errcallback;
1200 : :
1201 : : Assert(!ctx->fast_forward);
1202 : :
1203 : : /* Push callback + info on the error context stack */
1204 : 182752 : state.ctx = ctx;
1205 : 182752 : state.callback_name = "change";
1206 : 182752 : state.report_location = change->lsn;
1207 : 182752 : errcallback.callback = output_plugin_error_callback;
1208 : 182752 : errcallback.arg = &state;
1209 : 182752 : errcallback.previous = error_context_stack;
1210 : 182752 : error_context_stack = &errcallback;
1211 : :
1212 : : /* set output state */
1213 : 182752 : ctx->accept_writes = true;
1214 : 182752 : ctx->write_xid = txn->xid;
1215 : :
1216 : : /*
1217 : : * Report this change's lsn so replies from clients can give an up-to-date
1218 : : * answer. This won't ever be enough (and shouldn't be!) to confirm
1219 : : * receipt of this transaction, but it might allow another transaction's
1220 : : * commit to be confirmed with one message.
1221 : : */
1222 : 182752 : ctx->write_location = change->lsn;
1223 : :
1224 : 182752 : ctx->end_xact = false;
1225 : :
1226 : 182752 : ctx->callbacks.change_cb(ctx, txn, relation, change);
1227 : :
1228 : : /* Pop the error context stack */
1229 : 182745 : error_context_stack = errcallback.previous;
1230 : 182745 : }
1231 : :
1232 : : static void
1233 : 26 : truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1234 : : int nrelations, Relation relations[], ReorderBufferChange *change)
1235 : : {
1236 : 26 : LogicalDecodingContext *ctx = cache->private_data;
1237 : : LogicalErrorCallbackState state;
1238 : : ErrorContextCallback errcallback;
1239 : :
1240 : : Assert(!ctx->fast_forward);
1241 : :
1242 [ - + ]: 26 : if (!ctx->callbacks.truncate_cb)
1243 : 0 : return;
1244 : :
1245 : : /* Push callback + info on the error context stack */
1246 : 26 : state.ctx = ctx;
1247 : 26 : state.callback_name = "truncate";
1248 : 26 : state.report_location = change->lsn;
1249 : 26 : errcallback.callback = output_plugin_error_callback;
1250 : 26 : errcallback.arg = &state;
1251 : 26 : errcallback.previous = error_context_stack;
1252 : 26 : error_context_stack = &errcallback;
1253 : :
1254 : : /* set output state */
1255 : 26 : ctx->accept_writes = true;
1256 : 26 : ctx->write_xid = txn->xid;
1257 : :
1258 : : /*
1259 : : * Report this change's lsn so replies from clients can give an up-to-date
1260 : : * answer. This won't ever be enough (and shouldn't be!) to confirm
1261 : : * receipt of this transaction, but it might allow another transaction's
1262 : : * commit to be confirmed with one message.
1263 : : */
1264 : 26 : ctx->write_location = change->lsn;
1265 : :
1266 : 26 : ctx->end_xact = false;
1267 : :
1268 : 26 : ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
1269 : :
1270 : : /* Pop the error context stack */
1271 : 26 : error_context_stack = errcallback.previous;
1272 : : }
1273 : :
1274 : : bool
1275 : 170 : filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid,
1276 : : const char *gid)
1277 : : {
1278 : : LogicalErrorCallbackState state;
1279 : : ErrorContextCallback errcallback;
1280 : : bool ret;
1281 : :
1282 : : Assert(!ctx->fast_forward);
1283 : :
1284 : : /* Push callback + info on the error context stack */
1285 : 170 : state.ctx = ctx;
1286 : 170 : state.callback_name = "filter_prepare";
1287 : 170 : state.report_location = InvalidXLogRecPtr;
1288 : 170 : errcallback.callback = output_plugin_error_callback;
1289 : 170 : errcallback.arg = &state;
1290 : 170 : errcallback.previous = error_context_stack;
1291 : 170 : error_context_stack = &errcallback;
1292 : :
1293 : : /* set output state */
1294 : 170 : ctx->accept_writes = false;
1295 : 170 : ctx->end_xact = false;
1296 : :
1297 : : /* do the actual work: call callback */
1298 : 170 : ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid);
1299 : :
1300 : : /* Pop the error context stack */
1301 : 170 : error_context_stack = errcallback.previous;
1302 : :
1303 : 170 : return ret;
1304 : : }
1305 : :
1306 : : bool
1307 : 1633601 : filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, ReplOriginId origin_id)
1308 : : {
1309 : : LogicalErrorCallbackState state;
1310 : : ErrorContextCallback errcallback;
1311 : : bool ret;
1312 : :
1313 : : Assert(!ctx->fast_forward);
1314 : :
1315 : : /* Push callback + info on the error context stack */
1316 : 1633601 : state.ctx = ctx;
1317 : 1633601 : state.callback_name = "filter_by_origin";
1318 : 1633601 : state.report_location = InvalidXLogRecPtr;
1319 : 1633601 : errcallback.callback = output_plugin_error_callback;
1320 : 1633601 : errcallback.arg = &state;
1321 : 1633601 : errcallback.previous = error_context_stack;
1322 : 1633601 : error_context_stack = &errcallback;
1323 : :
1324 : : /* set output state */
1325 : 1633601 : ctx->accept_writes = false;
1326 : 1633601 : ctx->end_xact = false;
1327 : :
1328 : : /* do the actual work: call callback */
1329 : 1633601 : ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
1330 : :
1331 : : /* Pop the error context stack */
1332 : 1633601 : error_context_stack = errcallback.previous;
1333 : :
1334 : 1633601 : return ret;
1335 : : }
1336 : :
1337 : : static void
1338 : 16 : message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1339 : : XLogRecPtr message_lsn, bool transactional,
1340 : : const char *prefix, Size message_size, const char *message)
1341 : : {
1342 : 16 : LogicalDecodingContext *ctx = cache->private_data;
1343 : : LogicalErrorCallbackState state;
1344 : : ErrorContextCallback errcallback;
1345 : :
1346 : : Assert(!ctx->fast_forward);
1347 : :
1348 [ - + ]: 16 : if (ctx->callbacks.message_cb == NULL)
1349 : 0 : return;
1350 : :
1351 : : /* Push callback + info on the error context stack */
1352 : 16 : state.ctx = ctx;
1353 : 16 : state.callback_name = "message";
1354 : 16 : state.report_location = message_lsn;
1355 : 16 : errcallback.callback = output_plugin_error_callback;
1356 : 16 : errcallback.arg = &state;
1357 : 16 : errcallback.previous = error_context_stack;
1358 : 16 : error_context_stack = &errcallback;
1359 : :
1360 : : /* set output state */
1361 : 16 : ctx->accept_writes = true;
1362 [ + + ]: 16 : ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1363 : 16 : ctx->write_location = message_lsn;
1364 : 16 : ctx->end_xact = false;
1365 : :
1366 : : /* do the actual work: call callback */
1367 : 16 : ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
1368 : : message_size, message);
1369 : :
1370 : : /* Pop the error context stack */
1371 : 16 : error_context_stack = errcallback.previous;
1372 : : }
1373 : :
1374 : : static void
1375 : 685 : stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1376 : : XLogRecPtr first_lsn)
1377 : : {
1378 : 685 : LogicalDecodingContext *ctx = cache->private_data;
1379 : : LogicalErrorCallbackState state;
1380 : : ErrorContextCallback errcallback;
1381 : :
1382 : : Assert(!ctx->fast_forward);
1383 : :
1384 : : /* We're only supposed to call this when streaming is supported. */
1385 : : Assert(ctx->streaming);
1386 : :
1387 : : /* Push callback + info on the error context stack */
1388 : 685 : state.ctx = ctx;
1389 : 685 : state.callback_name = "stream_start";
1390 : 685 : state.report_location = first_lsn;
1391 : 685 : errcallback.callback = output_plugin_error_callback;
1392 : 685 : errcallback.arg = &state;
1393 : 685 : errcallback.previous = error_context_stack;
1394 : 685 : error_context_stack = &errcallback;
1395 : :
1396 : : /* set output state */
1397 : 685 : ctx->accept_writes = true;
1398 : 685 : ctx->write_xid = txn->xid;
1399 : :
1400 : : /*
1401 : : * Report this message's lsn so replies from clients can give an
1402 : : * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1403 : : * confirm receipt of this transaction, but it might allow another
1404 : : * transaction's commit to be confirmed with one message.
1405 : : */
1406 : 685 : ctx->write_location = first_lsn;
1407 : :
1408 : 685 : ctx->end_xact = false;
1409 : :
1410 : : /* in streaming mode, stream_start_cb is required */
1411 [ - + ]: 685 : if (ctx->callbacks.stream_start_cb == NULL)
1412 [ # # ]: 0 : ereport(ERROR,
1413 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1414 : : errmsg("logical streaming requires a %s callback",
1415 : : "stream_start_cb")));
1416 : :
1417 : 685 : ctx->callbacks.stream_start_cb(ctx, txn);
1418 : :
1419 : : /* Pop the error context stack */
1420 : 685 : error_context_stack = errcallback.previous;
1421 : 685 : }
1422 : :
1423 : : static void
1424 : 685 : stream_stop_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1425 : : XLogRecPtr last_lsn)
1426 : : {
1427 : 685 : LogicalDecodingContext *ctx = cache->private_data;
1428 : : LogicalErrorCallbackState state;
1429 : : ErrorContextCallback errcallback;
1430 : :
1431 : : Assert(!ctx->fast_forward);
1432 : :
1433 : : /* We're only supposed to call this when streaming is supported. */
1434 : : Assert(ctx->streaming);
1435 : :
1436 : : /* Push callback + info on the error context stack */
1437 : 685 : state.ctx = ctx;
1438 : 685 : state.callback_name = "stream_stop";
1439 : 685 : state.report_location = last_lsn;
1440 : 685 : errcallback.callback = output_plugin_error_callback;
1441 : 685 : errcallback.arg = &state;
1442 : 685 : errcallback.previous = error_context_stack;
1443 : 685 : error_context_stack = &errcallback;
1444 : :
1445 : : /* set output state */
1446 : 685 : ctx->accept_writes = true;
1447 : 685 : ctx->write_xid = txn->xid;
1448 : :
1449 : : /*
1450 : : * Report this message's lsn so replies from clients can give an
1451 : : * up-to-date answer. This won't ever be enough (and shouldn't be!) to
1452 : : * confirm receipt of this transaction, but it might allow another
1453 : : * transaction's commit to be confirmed with one message.
1454 : : */
1455 : 685 : ctx->write_location = last_lsn;
1456 : :
1457 : 685 : ctx->end_xact = false;
1458 : :
1459 : : /* in streaming mode, stream_stop_cb is required */
1460 [ - + ]: 685 : if (ctx->callbacks.stream_stop_cb == NULL)
1461 [ # # ]: 0 : ereport(ERROR,
1462 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1463 : : errmsg("logical streaming requires a %s callback",
1464 : : "stream_stop_cb")));
1465 : :
1466 : 685 : ctx->callbacks.stream_stop_cb(ctx, txn);
1467 : :
1468 : : /* Pop the error context stack */
1469 : 685 : error_context_stack = errcallback.previous;
1470 : 685 : }
1471 : :
1472 : : static void
1473 : 30 : stream_abort_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1474 : : XLogRecPtr abort_lsn)
1475 : : {
1476 : 30 : LogicalDecodingContext *ctx = cache->private_data;
1477 : : LogicalErrorCallbackState state;
1478 : : ErrorContextCallback errcallback;
1479 : :
1480 : : Assert(!ctx->fast_forward);
1481 : :
1482 : : /* We're only supposed to call this when streaming is supported. */
1483 : : Assert(ctx->streaming);
1484 : :
1485 : : /* Push callback + info on the error context stack */
1486 : 30 : state.ctx = ctx;
1487 : 30 : state.callback_name = "stream_abort";
1488 : 30 : state.report_location = abort_lsn;
1489 : 30 : errcallback.callback = output_plugin_error_callback;
1490 : 30 : errcallback.arg = &state;
1491 : 30 : errcallback.previous = error_context_stack;
1492 : 30 : error_context_stack = &errcallback;
1493 : :
1494 : : /* set output state */
1495 : 30 : ctx->accept_writes = true;
1496 : 30 : ctx->write_xid = txn->xid;
1497 : 30 : ctx->write_location = abort_lsn;
1498 : 30 : ctx->end_xact = true;
1499 : :
1500 : : /* in streaming mode, stream_abort_cb is required */
1501 [ - + ]: 30 : if (ctx->callbacks.stream_abort_cb == NULL)
1502 [ # # ]: 0 : ereport(ERROR,
1503 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1504 : : errmsg("logical streaming requires a %s callback",
1505 : : "stream_abort_cb")));
1506 : :
1507 : 30 : ctx->callbacks.stream_abort_cb(ctx, txn, abort_lsn);
1508 : :
1509 : : /* Pop the error context stack */
1510 : 30 : error_context_stack = errcallback.previous;
1511 : 30 : }
1512 : :
1513 : : static void
1514 : 17 : stream_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1515 : : XLogRecPtr prepare_lsn)
1516 : : {
1517 : 17 : LogicalDecodingContext *ctx = cache->private_data;
1518 : : LogicalErrorCallbackState state;
1519 : : ErrorContextCallback errcallback;
1520 : :
1521 : : Assert(!ctx->fast_forward);
1522 : :
1523 : : /*
1524 : : * We're only supposed to call this when streaming and two-phase commits
1525 : : * are supported.
1526 : : */
1527 : : Assert(ctx->streaming);
1528 : : Assert(ctx->twophase);
1529 : :
1530 : : /* Push callback + info on the error context stack */
1531 : 17 : state.ctx = ctx;
1532 : 17 : state.callback_name = "stream_prepare";
1533 : 17 : state.report_location = txn->final_lsn;
1534 : 17 : errcallback.callback = output_plugin_error_callback;
1535 : 17 : errcallback.arg = &state;
1536 : 17 : errcallback.previous = error_context_stack;
1537 : 17 : error_context_stack = &errcallback;
1538 : :
1539 : : /* set output state */
1540 : 17 : ctx->accept_writes = true;
1541 : 17 : ctx->write_xid = txn->xid;
1542 : 17 : ctx->write_location = txn->end_lsn;
1543 : 17 : ctx->end_xact = true;
1544 : :
1545 : : /* in streaming mode with two-phase commits, stream_prepare_cb is required */
1546 [ - + ]: 17 : if (ctx->callbacks.stream_prepare_cb == NULL)
1547 [ # # ]: 0 : ereport(ERROR,
1548 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1549 : : errmsg("logical streaming at prepare time requires a %s callback",
1550 : : "stream_prepare_cb")));
1551 : :
1552 : 17 : ctx->callbacks.stream_prepare_cb(ctx, txn, prepare_lsn);
1553 : :
1554 : : /* Pop the error context stack */
1555 : 17 : error_context_stack = errcallback.previous;
1556 : 17 : }
1557 : :
1558 : : static void
1559 : 50 : stream_commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1560 : : XLogRecPtr commit_lsn)
1561 : : {
1562 : 50 : LogicalDecodingContext *ctx = cache->private_data;
1563 : : LogicalErrorCallbackState state;
1564 : : ErrorContextCallback errcallback;
1565 : :
1566 : : Assert(!ctx->fast_forward);
1567 : :
1568 : : /* We're only supposed to call this when streaming is supported. */
1569 : : Assert(ctx->streaming);
1570 : :
1571 : : /* Push callback + info on the error context stack */
1572 : 50 : state.ctx = ctx;
1573 : 50 : state.callback_name = "stream_commit";
1574 : 50 : state.report_location = txn->final_lsn;
1575 : 50 : errcallback.callback = output_plugin_error_callback;
1576 : 50 : errcallback.arg = &state;
1577 : 50 : errcallback.previous = error_context_stack;
1578 : 50 : error_context_stack = &errcallback;
1579 : :
1580 : : /* set output state */
1581 : 50 : ctx->accept_writes = true;
1582 : 50 : ctx->write_xid = txn->xid;
1583 : 50 : ctx->write_location = txn->end_lsn;
1584 : 50 : ctx->end_xact = true;
1585 : :
1586 : : /* in streaming mode, stream_commit_cb is required */
1587 [ - + ]: 50 : if (ctx->callbacks.stream_commit_cb == NULL)
1588 [ # # ]: 0 : ereport(ERROR,
1589 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1590 : : errmsg("logical streaming requires a %s callback",
1591 : : "stream_commit_cb")));
1592 : :
1593 : 50 : ctx->callbacks.stream_commit_cb(ctx, txn, commit_lsn);
1594 : :
1595 : : /* Pop the error context stack */
1596 : 50 : error_context_stack = errcallback.previous;
1597 : 50 : }
1598 : :
1599 : : static void
1600 : 175636 : stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1601 : : Relation relation, ReorderBufferChange *change)
1602 : : {
1603 : 175636 : LogicalDecodingContext *ctx = cache->private_data;
1604 : : LogicalErrorCallbackState state;
1605 : : ErrorContextCallback errcallback;
1606 : :
1607 : : Assert(!ctx->fast_forward);
1608 : :
1609 : : /* We're only supposed to call this when streaming is supported. */
1610 : : Assert(ctx->streaming);
1611 : :
1612 : : /* Push callback + info on the error context stack */
1613 : 175636 : state.ctx = ctx;
1614 : 175636 : state.callback_name = "stream_change";
1615 : 175636 : state.report_location = change->lsn;
1616 : 175636 : errcallback.callback = output_plugin_error_callback;
1617 : 175636 : errcallback.arg = &state;
1618 : 175636 : errcallback.previous = error_context_stack;
1619 : 175636 : error_context_stack = &errcallback;
1620 : :
1621 : : /* set output state */
1622 : 175636 : ctx->accept_writes = true;
1623 : 175636 : ctx->write_xid = txn->xid;
1624 : :
1625 : : /*
1626 : : * Report this change's lsn so replies from clients can give an up-to-date
1627 : : * answer. This won't ever be enough (and shouldn't be!) to confirm
1628 : : * receipt of this transaction, but it might allow another transaction's
1629 : : * commit to be confirmed with one message.
1630 : : */
1631 : 175636 : ctx->write_location = change->lsn;
1632 : :
1633 : 175636 : ctx->end_xact = false;
1634 : :
1635 : : /* in streaming mode, stream_change_cb is required */
1636 [ - + ]: 175636 : if (ctx->callbacks.stream_change_cb == NULL)
1637 [ # # ]: 0 : ereport(ERROR,
1638 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1639 : : errmsg("logical streaming requires a %s callback",
1640 : : "stream_change_cb")));
1641 : :
1642 : 175636 : ctx->callbacks.stream_change_cb(ctx, txn, relation, change);
1643 : :
1644 : : /* Pop the error context stack */
1645 : 175636 : error_context_stack = errcallback.previous;
1646 : 175636 : }
1647 : :
1648 : : static void
1649 : 3 : stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1650 : : XLogRecPtr message_lsn, bool transactional,
1651 : : const char *prefix, Size message_size, const char *message)
1652 : : {
1653 : 3 : LogicalDecodingContext *ctx = cache->private_data;
1654 : : LogicalErrorCallbackState state;
1655 : : ErrorContextCallback errcallback;
1656 : :
1657 : : Assert(!ctx->fast_forward);
1658 : :
1659 : : /* We're only supposed to call this when streaming is supported. */
1660 : : Assert(ctx->streaming);
1661 : :
1662 : : /* this callback is optional */
1663 [ - + ]: 3 : if (ctx->callbacks.stream_message_cb == NULL)
1664 : 0 : return;
1665 : :
1666 : : /* Push callback + info on the error context stack */
1667 : 3 : state.ctx = ctx;
1668 : 3 : state.callback_name = "stream_message";
1669 : 3 : state.report_location = message_lsn;
1670 : 3 : errcallback.callback = output_plugin_error_callback;
1671 : 3 : errcallback.arg = &state;
1672 : 3 : errcallback.previous = error_context_stack;
1673 : 3 : error_context_stack = &errcallback;
1674 : :
1675 : : /* set output state */
1676 : 3 : ctx->accept_writes = true;
1677 [ + - ]: 3 : ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
1678 : 3 : ctx->write_location = message_lsn;
1679 : 3 : ctx->end_xact = false;
1680 : :
1681 : : /* do the actual work: call callback */
1682 : 3 : ctx->callbacks.stream_message_cb(ctx, txn, message_lsn, transactional, prefix,
1683 : : message_size, message);
1684 : :
1685 : : /* Pop the error context stack */
1686 : 3 : error_context_stack = errcallback.previous;
1687 : : }
1688 : :
1689 : : static void
1690 : 0 : stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1691 : : int nrelations, Relation relations[],
1692 : : ReorderBufferChange *change)
1693 : : {
1694 : 0 : LogicalDecodingContext *ctx = cache->private_data;
1695 : : LogicalErrorCallbackState state;
1696 : : ErrorContextCallback errcallback;
1697 : :
1698 : : Assert(!ctx->fast_forward);
1699 : :
1700 : : /* We're only supposed to call this when streaming is supported. */
1701 : : Assert(ctx->streaming);
1702 : :
1703 : : /* this callback is optional */
1704 [ # # ]: 0 : if (!ctx->callbacks.stream_truncate_cb)
1705 : 0 : return;
1706 : :
1707 : : /* Push callback + info on the error context stack */
1708 : 0 : state.ctx = ctx;
1709 : 0 : state.callback_name = "stream_truncate";
1710 : 0 : state.report_location = change->lsn;
1711 : 0 : errcallback.callback = output_plugin_error_callback;
1712 : 0 : errcallback.arg = &state;
1713 : 0 : errcallback.previous = error_context_stack;
1714 : 0 : error_context_stack = &errcallback;
1715 : :
1716 : : /* set output state */
1717 : 0 : ctx->accept_writes = true;
1718 : 0 : ctx->write_xid = txn->xid;
1719 : :
1720 : : /*
1721 : : * Report this change's lsn so replies from clients can give an up-to-date
1722 : : * answer. This won't ever be enough (and shouldn't be!) to confirm
1723 : : * receipt of this transaction, but it might allow another transaction's
1724 : : * commit to be confirmed with one message.
1725 : : */
1726 : 0 : ctx->write_location = change->lsn;
1727 : :
1728 : 0 : ctx->end_xact = false;
1729 : :
1730 : 0 : ctx->callbacks.stream_truncate_cb(ctx, txn, nrelations, relations, change);
1731 : :
1732 : : /* Pop the error context stack */
1733 : 0 : error_context_stack = errcallback.previous;
1734 : : }
1735 : :
1736 : : static void
1737 : 3369 : update_progress_txn_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
1738 : : XLogRecPtr lsn)
1739 : : {
1740 : 3369 : LogicalDecodingContext *ctx = cache->private_data;
1741 : : LogicalErrorCallbackState state;
1742 : : ErrorContextCallback errcallback;
1743 : :
1744 : : Assert(!ctx->fast_forward);
1745 : :
1746 : : /* Push callback + info on the error context stack */
1747 : 3369 : state.ctx = ctx;
1748 : 3369 : state.callback_name = "update_progress_txn";
1749 : 3369 : state.report_location = lsn;
1750 : 3369 : errcallback.callback = output_plugin_error_callback;
1751 : 3369 : errcallback.arg = &state;
1752 : 3369 : errcallback.previous = error_context_stack;
1753 : 3369 : error_context_stack = &errcallback;
1754 : :
1755 : : /* set output state */
1756 : 3369 : ctx->accept_writes = false;
1757 : 3369 : ctx->write_xid = txn->xid;
1758 : :
1759 : : /*
1760 : : * Report this change's lsn so replies from clients can give an up-to-date
1761 : : * answer. This won't ever be enough (and shouldn't be!) to confirm
1762 : : * receipt of this transaction, but it might allow another transaction's
1763 : : * commit to be confirmed with one message.
1764 : : */
1765 : 3369 : ctx->write_location = lsn;
1766 : :
1767 : 3369 : ctx->end_xact = false;
1768 : :
1769 : 3369 : OutputPluginUpdateProgress(ctx, false);
1770 : :
1771 : : /* Pop the error context stack */
1772 : 3369 : error_context_stack = errcallback.previous;
1773 : 3369 : }
1774 : :
1775 : : /*
1776 : : * Set the required catalog xmin horizon for historic snapshots in the current
1777 : : * replication slot.
1778 : : *
1779 : : * Note that in the most cases, we won't be able to immediately use the xmin
1780 : : * to increase the xmin horizon: we need to wait till the client has confirmed
1781 : : * receiving current_lsn with LogicalConfirmReceivedLocation().
1782 : : */
1783 : : void
1784 : 585 : LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
1785 : : {
1786 : 585 : bool updated_xmin = false;
1787 : : ReplicationSlot *slot;
1788 : 585 : bool got_new_xmin = false;
1789 : :
1790 : 585 : slot = MyReplicationSlot;
1791 : :
1792 : : Assert(slot != NULL);
1793 : :
1794 : 585 : SpinLockAcquire(&slot->mutex);
1795 : :
1796 : : /*
1797 : : * don't overwrite if we already have a newer xmin. This can happen if we
1798 : : * restart decoding in a slot.
1799 : : */
1800 [ + + ]: 585 : if (TransactionIdPrecedesOrEquals(xmin, slot->data.catalog_xmin))
1801 : : {
1802 : : }
1803 : :
1804 : : /*
1805 : : * If the client has already confirmed up to this lsn, we directly can
1806 : : * mark this as accepted. This can happen if we restart decoding in a
1807 : : * slot.
1808 : : */
1809 [ + + ]: 219 : else if (current_lsn <= slot->data.confirmed_flush)
1810 : : {
1811 : 71 : slot->candidate_catalog_xmin = xmin;
1812 : 71 : slot->candidate_xmin_lsn = current_lsn;
1813 : :
1814 : : /* our candidate can directly be used */
1815 : 71 : updated_xmin = true;
1816 : : }
1817 : :
1818 : : /*
1819 : : * Only increase if the previous values have been applied, otherwise we
1820 : : * might never end up updating if the receiver acks too slowly.
1821 : : */
1822 [ + + ]: 148 : else if (!XLogRecPtrIsValid(slot->candidate_xmin_lsn))
1823 : : {
1824 : 32 : slot->candidate_catalog_xmin = xmin;
1825 : 32 : slot->candidate_xmin_lsn = current_lsn;
1826 : :
1827 : : /*
1828 : : * Log new xmin at an appropriate log level after releasing the
1829 : : * spinlock.
1830 : : */
1831 : 32 : got_new_xmin = true;
1832 : : }
1833 : 585 : SpinLockRelease(&slot->mutex);
1834 : :
1835 [ + + ]: 585 : if (got_new_xmin)
1836 [ + + ]: 32 : elog(DEBUG1, "got new catalog xmin %u at %X/%08X", xmin,
1837 : : LSN_FORMAT_ARGS(current_lsn));
1838 : :
1839 : : /* candidate already valid with the current flush position, apply */
1840 [ + + ]: 585 : if (updated_xmin)
1841 : 71 : LogicalConfirmReceivedLocation(slot->data.confirmed_flush);
1842 : 585 : }
1843 : :
1844 : : /*
1845 : : * Mark the minimal LSN (restart_lsn) we need to read to replay all
1846 : : * transactions that have not yet committed at current_lsn.
1847 : : *
1848 : : * Just like LogicalIncreaseXminForSlot this only takes effect when the
1849 : : * client has confirmed to have received current_lsn.
1850 : : */
1851 : : void
1852 : 519 : LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
1853 : : {
1854 : 519 : bool updated_lsn = false;
1855 : : ReplicationSlot *slot;
1856 : :
1857 : 519 : slot = MyReplicationSlot;
1858 : :
1859 : : Assert(slot != NULL);
1860 : : Assert(XLogRecPtrIsValid(restart_lsn));
1861 : : Assert(XLogRecPtrIsValid(current_lsn));
1862 : :
1863 : 519 : SpinLockAcquire(&slot->mutex);
1864 : :
1865 : : /* don't overwrite if have a newer restart lsn */
1866 [ + + ]: 519 : if (restart_lsn <= slot->data.restart_lsn)
1867 : : {
1868 : 10 : SpinLockRelease(&slot->mutex);
1869 : : }
1870 : :
1871 : : /*
1872 : : * We might have already flushed far enough to directly accept this lsn,
1873 : : * in this case there is no need to check for existing candidate LSNs
1874 : : */
1875 [ + + ]: 509 : else if (current_lsn <= slot->data.confirmed_flush)
1876 : : {
1877 : 334 : slot->candidate_restart_valid = current_lsn;
1878 : 334 : slot->candidate_restart_lsn = restart_lsn;
1879 : 334 : SpinLockRelease(&slot->mutex);
1880 : :
1881 : : /* our candidate can directly be used */
1882 : 334 : updated_lsn = true;
1883 : : }
1884 : :
1885 : : /*
1886 : : * Only increase if the previous values have been applied, otherwise we
1887 : : * might never end up updating if the receiver acks too slowly. A missed
1888 : : * value here will just cause some extra effort after reconnecting.
1889 : : */
1890 [ + + ]: 175 : else if (!XLogRecPtrIsValid(slot->candidate_restart_valid))
1891 : : {
1892 : 49 : slot->candidate_restart_valid = current_lsn;
1893 : 49 : slot->candidate_restart_lsn = restart_lsn;
1894 : 49 : SpinLockRelease(&slot->mutex);
1895 : :
1896 [ + + ]: 49 : elog(DEBUG1, "got new restart lsn %X/%08X at %X/%08X",
1897 : : LSN_FORMAT_ARGS(restart_lsn),
1898 : : LSN_FORMAT_ARGS(current_lsn));
1899 : : }
1900 : : else
1901 : : {
1902 : : XLogRecPtr candidate_restart_lsn;
1903 : : XLogRecPtr candidate_restart_valid;
1904 : : XLogRecPtr confirmed_flush;
1905 : :
1906 : 126 : candidate_restart_lsn = slot->candidate_restart_lsn;
1907 : 126 : candidate_restart_valid = slot->candidate_restart_valid;
1908 : 126 : confirmed_flush = slot->data.confirmed_flush;
1909 : 126 : SpinLockRelease(&slot->mutex);
1910 : :
1911 [ + + ]: 126 : elog(DEBUG1, "failed to increase restart lsn: proposed %X/%08X, after %X/%08X, current candidate %X/%08X, current after %X/%08X, flushed up to %X/%08X",
1912 : : LSN_FORMAT_ARGS(restart_lsn),
1913 : : LSN_FORMAT_ARGS(current_lsn),
1914 : : LSN_FORMAT_ARGS(candidate_restart_lsn),
1915 : : LSN_FORMAT_ARGS(candidate_restart_valid),
1916 : : LSN_FORMAT_ARGS(confirmed_flush));
1917 : : }
1918 : :
1919 : : /* candidates are already valid with the current flush position, apply */
1920 [ + + ]: 519 : if (updated_lsn)
1921 : 334 : LogicalConfirmReceivedLocation(slot->data.confirmed_flush);
1922 : 519 : }
1923 : :
1924 : : /*
1925 : : * Handle a consumer's confirmation having received all changes up to lsn.
1926 : : */
1927 : : void
1928 : 21154 : LogicalConfirmReceivedLocation(XLogRecPtr lsn)
1929 : : {
1930 : : Assert(XLogRecPtrIsValid(lsn));
1931 : :
1932 : : /* Do an unlocked check for candidate_lsn first. */
1933 [ + + ]: 21154 : if (XLogRecPtrIsValid(MyReplicationSlot->candidate_xmin_lsn) ||
1934 [ + + ]: 20989 : XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_valid))
1935 : 517 : {
1936 : 517 : bool updated_xmin = false;
1937 : 517 : bool updated_restart = false;
1938 : : XLogRecPtr restart_lsn pg_attribute_unused();
1939 : :
1940 : 517 : SpinLockAcquire(&MyReplicationSlot->mutex);
1941 : :
1942 : : /* remember the old restart lsn */
1943 : 517 : restart_lsn = MyReplicationSlot->data.restart_lsn;
1944 : :
1945 : : /*
1946 : : * Prevent moving the confirmed_flush backwards, as this could lead to
1947 : : * data duplication issues caused by replicating already replicated
1948 : : * changes.
1949 : : *
1950 : : * This can happen when a client acknowledges an LSN it doesn't have
1951 : : * to do anything for, and thus didn't store persistently. After a
1952 : : * restart, the client can send the prior LSN that it stored
1953 : : * persistently as an acknowledgement, but we need to ignore such an
1954 : : * LSN. See similar case handling in CreateDecodingContext.
1955 : : */
1956 [ + + ]: 517 : if (lsn > MyReplicationSlot->data.confirmed_flush)
1957 : 42 : MyReplicationSlot->data.confirmed_flush = lsn;
1958 : :
1959 : : /* if we're past the location required for bumping xmin, do so */
1960 [ + + ]: 517 : if (XLogRecPtrIsValid(MyReplicationSlot->candidate_xmin_lsn) &&
1961 [ + + ]: 165 : MyReplicationSlot->candidate_xmin_lsn <= lsn)
1962 : : {
1963 : : /*
1964 : : * We have to write the changed xmin to disk *before* we change
1965 : : * the in-memory value, otherwise after a crash we wouldn't know
1966 : : * that some catalog tuples might have been removed already.
1967 : : *
1968 : : * Ensure that by first writing to ->xmin and only update
1969 : : * ->effective_xmin once the new state is synced to disk. After a
1970 : : * crash ->effective_xmin is set to ->xmin.
1971 : : */
1972 [ + - ]: 92 : if (TransactionIdIsValid(MyReplicationSlot->candidate_catalog_xmin) &&
1973 [ + - ]: 92 : MyReplicationSlot->data.catalog_xmin != MyReplicationSlot->candidate_catalog_xmin)
1974 : : {
1975 : 92 : MyReplicationSlot->data.catalog_xmin = MyReplicationSlot->candidate_catalog_xmin;
1976 : 92 : MyReplicationSlot->candidate_catalog_xmin = InvalidTransactionId;
1977 : 92 : MyReplicationSlot->candidate_xmin_lsn = InvalidXLogRecPtr;
1978 : 92 : updated_xmin = true;
1979 : : }
1980 : : }
1981 : :
1982 [ + + ]: 517 : if (XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_valid) &&
1983 [ + + ]: 446 : MyReplicationSlot->candidate_restart_valid <= lsn)
1984 : : {
1985 : : Assert(XLogRecPtrIsValid(MyReplicationSlot->candidate_restart_lsn));
1986 : :
1987 : 373 : MyReplicationSlot->data.restart_lsn = MyReplicationSlot->candidate_restart_lsn;
1988 : 373 : MyReplicationSlot->candidate_restart_lsn = InvalidXLogRecPtr;
1989 : 373 : MyReplicationSlot->candidate_restart_valid = InvalidXLogRecPtr;
1990 : 373 : updated_restart = true;
1991 : : }
1992 : :
1993 : 517 : SpinLockRelease(&MyReplicationSlot->mutex);
1994 : :
1995 : : /* first write new xmin to disk, so we know what's up after a crash */
1996 [ + + + + ]: 517 : if (updated_xmin || updated_restart)
1997 : : {
1998 : : #ifdef USE_INJECTION_POINTS
1999 : : XLogSegNo seg1,
2000 : : seg2;
2001 : :
2002 : 444 : XLByteToSeg(restart_lsn, seg1, wal_segment_size);
2003 : 444 : XLByteToSeg(MyReplicationSlot->data.restart_lsn, seg2, wal_segment_size);
2004 : :
2005 : : /* trigger injection point, but only if segment changes */
2006 [ + + ]: 444 : if (seg1 != seg2)
2007 : 18 : INJECTION_POINT("logical-replication-slot-advance-segment", NULL);
2008 : : #endif
2009 : :
2010 : 444 : ReplicationSlotMarkDirty();
2011 : 444 : ReplicationSlotSave();
2012 [ + + ]: 444 : elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
2013 : : }
2014 : :
2015 : : /*
2016 : : * Now the new xmin is safely on disk, we can let the global value
2017 : : * advance. We do not take ProcArrayLock or similar since we only
2018 : : * advance xmin here and there's not much harm done by a concurrent
2019 : : * computation missing that.
2020 : : */
2021 [ + + ]: 517 : if (updated_xmin)
2022 : : {
2023 : 92 : SpinLockAcquire(&MyReplicationSlot->mutex);
2024 : 92 : MyReplicationSlot->effective_catalog_xmin = MyReplicationSlot->data.catalog_xmin;
2025 : 92 : SpinLockRelease(&MyReplicationSlot->mutex);
2026 : :
2027 : 92 : ReplicationSlotsComputeRequiredXmin(false);
2028 : : }
2029 : :
2030 : : /*
2031 : : * Now that the new restart_lsn is safely on disk, recompute the
2032 : : * global WAL retention requirement.
2033 : : */
2034 [ + + ]: 517 : if (updated_restart)
2035 : 373 : ReplicationSlotsComputeRequiredLSN();
2036 : : }
2037 : : else
2038 : : {
2039 : 20637 : SpinLockAcquire(&MyReplicationSlot->mutex);
2040 : :
2041 : : /*
2042 : : * Prevent moving the confirmed_flush backwards. See comments above
2043 : : * for the details.
2044 : : */
2045 [ + + ]: 20637 : if (lsn > MyReplicationSlot->data.confirmed_flush)
2046 : 19802 : MyReplicationSlot->data.confirmed_flush = lsn;
2047 : :
2048 : 20637 : SpinLockRelease(&MyReplicationSlot->mutex);
2049 : : }
2050 : 21154 : }
2051 : :
2052 : : /*
2053 : : * Clear logical streaming state during (sub)transaction abort.
2054 : : */
2055 : : void
2056 : 41451 : ResetLogicalStreamingState(void)
2057 : : {
2058 : 41451 : CheckXidAlive = InvalidTransactionId;
2059 : 41451 : bsysscan = false;
2060 : 41451 : }
2061 : :
2062 : : /*
2063 : : * Report stats for a slot.
2064 : : */
2065 : : void
2066 : 7700 : UpdateDecodingStats(LogicalDecodingContext *ctx)
2067 : : {
2068 : 7700 : ReorderBuffer *rb = ctx->reorder;
2069 : : PgStat_StatReplSlotEntry repSlotStat;
2070 : :
2071 : : /* Nothing to do if we don't have any replication stats to be sent. */
2072 [ + + + + : 7700 : if (rb->spillBytes <= 0 && rb->streamBytes <= 0 && rb->totalBytes <= 0 &&
+ + ]
2073 [ + + ]: 536 : rb->memExceededCount <= 0)
2074 : 524 : return;
2075 : :
2076 [ + + ]: 7176 : elog(DEBUG2, "UpdateDecodingStats: updating stats %p %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64 " %" PRId64,
2077 : : rb,
2078 : : rb->spillTxns,
2079 : : rb->spillCount,
2080 : : rb->spillBytes,
2081 : : rb->streamTxns,
2082 : : rb->streamCount,
2083 : : rb->streamBytes,
2084 : : rb->memExceededCount,
2085 : : rb->totalTxns,
2086 : : rb->totalBytes);
2087 : :
2088 : 7176 : repSlotStat.spill_txns = rb->spillTxns;
2089 : 7176 : repSlotStat.spill_count = rb->spillCount;
2090 : 7176 : repSlotStat.spill_bytes = rb->spillBytes;
2091 : 7176 : repSlotStat.stream_txns = rb->streamTxns;
2092 : 7176 : repSlotStat.stream_count = rb->streamCount;
2093 : 7176 : repSlotStat.stream_bytes = rb->streamBytes;
2094 : 7176 : repSlotStat.mem_exceeded_count = rb->memExceededCount;
2095 : 7176 : repSlotStat.total_txns = rb->totalTxns;
2096 : 7176 : repSlotStat.total_bytes = rb->totalBytes;
2097 : :
2098 : 7176 : pgstat_report_replslot(ctx->slot, &repSlotStat);
2099 : :
2100 : 7176 : rb->spillTxns = 0;
2101 : 7176 : rb->spillCount = 0;
2102 : 7176 : rb->spillBytes = 0;
2103 : 7176 : rb->streamTxns = 0;
2104 : 7176 : rb->streamCount = 0;
2105 : 7176 : rb->streamBytes = 0;
2106 : 7176 : rb->memExceededCount = 0;
2107 : 7176 : rb->totalTxns = 0;
2108 : 7176 : rb->totalBytes = 0;
2109 : : }
2110 : :
2111 : : /*
2112 : : * Read up to the end of WAL starting from the decoding slot's restart_lsn
2113 : : * to end_of_wal in order to check if any meaningful/decodable WAL records
2114 : : * are encountered. scan_cutoff_lsn is the LSN, where we can terminate the
2115 : : * WAL scan early if we find a decodable WAL record after this LSN.
2116 : : *
2117 : : * Returns the last LSN decodable WAL record's LSN if found, otherwise
2118 : : * returns InvalidXLogRecPtr.
2119 : : */
2120 : : XLogRecPtr
2121 : 4 : LogicalReplicationSlotCheckPendingWal(XLogRecPtr end_of_wal,
2122 : : XLogRecPtr scan_cutoff_lsn)
2123 : : {
2124 : 4 : XLogRecPtr last_pending_wal = InvalidXLogRecPtr;
2125 : :
2126 : : Assert(MyReplicationSlot);
2127 : : Assert(end_of_wal >= scan_cutoff_lsn);
2128 : :
2129 [ + - ]: 4 : PG_TRY();
2130 : : {
2131 : : LogicalDecodingContext *ctx;
2132 : :
2133 : : /*
2134 : : * Create our decoding context in fast_forward mode, passing start_lsn
2135 : : * as InvalidXLogRecPtr, so that we start processing from the slot's
2136 : : * confirmed_flush.
2137 : : */
2138 : 8 : ctx = CreateDecodingContext(InvalidXLogRecPtr,
2139 : : NIL,
2140 : : true, /* fast_forward */
2141 : 4 : XL_ROUTINE(.page_read = read_local_xlog_page,
2142 : : .segment_open = wal_segment_open,
2143 : : .segment_close = wal_segment_close),
2144 : : NULL, NULL, NULL);
2145 : :
2146 : : /*
2147 : : * Start reading at the slot's restart_lsn, which we know points to a
2148 : : * valid record.
2149 : : */
2150 : 4 : XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
2151 : :
2152 : : /* Invalidate non-timetravel entries */
2153 : 4 : InvalidateSystemCaches();
2154 : :
2155 [ + + ]: 88 : while (ctx->reader->EndRecPtr < end_of_wal)
2156 : : {
2157 : : XLogRecord *record;
2158 : 84 : char *errm = NULL;
2159 : :
2160 : 84 : record = XLogReadRecord(ctx->reader, &errm);
2161 : :
2162 [ - + ]: 84 : if (errm)
2163 [ # # ]: 0 : elog(ERROR, "could not find record for logical decoding: %s", errm);
2164 : :
2165 [ + - ]: 84 : if (record != NULL)
2166 : 84 : LogicalDecodingProcessRecord(ctx, ctx->reader);
2167 : :
2168 [ + + ]: 84 : if (ctx->processing_required)
2169 : : {
2170 : 2 : last_pending_wal = ctx->reader->ReadRecPtr;
2171 : :
2172 : : /*
2173 : : * If we find a decodable WAL after the scan_cutoff_lsn point,
2174 : : * we can terminate the scan early.
2175 : : */
2176 [ - + ]: 2 : if (last_pending_wal >= scan_cutoff_lsn)
2177 : 0 : break;
2178 : :
2179 : : /* Reset the flag and continue checking */
2180 : 2 : ctx->processing_required = false;
2181 : : }
2182 : :
2183 [ - + ]: 84 : CHECK_FOR_INTERRUPTS();
2184 : : }
2185 : :
2186 : : /* Clean up */
2187 : 4 : FreeDecodingContext(ctx);
2188 : 4 : InvalidateSystemCaches();
2189 : : }
2190 : 0 : PG_CATCH();
2191 : : {
2192 : : /* clear all timetravel entries */
2193 : 0 : InvalidateSystemCaches();
2194 : :
2195 : 0 : PG_RE_THROW();
2196 : : }
2197 [ - + ]: 4 : PG_END_TRY();
2198 : :
2199 : 4 : return last_pending_wal;
2200 : : }
2201 : :
2202 : : /*
2203 : : * Helper function for advancing our logical replication slot forward.
2204 : : *
2205 : : * The slot's restart_lsn is used as start point for reading records, while
2206 : : * confirmed_flush is used as base point for the decoding context.
2207 : : *
2208 : : * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
2209 : : * because we need to digest WAL to advance restart_lsn allowing to recycle
2210 : : * WAL and removal of old catalog tuples. As decoding is done in fast_forward
2211 : : * mode, no changes are generated anyway.
2212 : : *
2213 : : * *found_consistent_snapshot will be true if the initial decoding snapshot has
2214 : : * been built; Otherwise, it will be false.
2215 : : */
2216 : : XLogRecPtr
2217 : 22 : LogicalSlotAdvanceAndCheckSnapState(XLogRecPtr moveto,
2218 : : bool *found_consistent_snapshot)
2219 : : {
2220 : : LogicalDecodingContext *ctx;
2221 : : XLogRecPtr retlsn;
2222 : :
2223 : : Assert(XLogRecPtrIsValid(moveto));
2224 : :
2225 [ + + ]: 22 : if (found_consistent_snapshot)
2226 : 11 : *found_consistent_snapshot = false;
2227 : :
2228 [ + - ]: 22 : PG_TRY();
2229 : : {
2230 : : /*
2231 : : * Create our decoding context in fast_forward mode, passing start_lsn
2232 : : * as InvalidXLogRecPtr, so that we start processing from my slot's
2233 : : * confirmed_flush.
2234 : : */
2235 : 44 : ctx = CreateDecodingContext(InvalidXLogRecPtr,
2236 : : NIL,
2237 : : true, /* fast_forward */
2238 : 22 : XL_ROUTINE(.page_read = read_local_xlog_page,
2239 : : .segment_open = wal_segment_open,
2240 : : .segment_close = wal_segment_close),
2241 : : NULL, NULL, NULL);
2242 : :
2243 : : /*
2244 : : * Wait for specified streaming replication standby servers (if any)
2245 : : * to confirm receipt of WAL up to moveto lsn.
2246 : : */
2247 : 22 : WaitForStandbyConfirmation(moveto);
2248 : :
2249 : : /*
2250 : : * Start reading at the slot's restart_lsn, which we know to point to
2251 : : * a valid record.
2252 : : */
2253 : 22 : XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
2254 : :
2255 : : /* invalidate non-timetravel entries */
2256 : 22 : InvalidateSystemCaches();
2257 : :
2258 : : /* Decode records until we reach the requested target */
2259 [ + + ]: 2900 : while (ctx->reader->EndRecPtr < moveto)
2260 : : {
2261 : 2878 : char *errm = NULL;
2262 : : XLogRecord *record;
2263 : :
2264 : : /*
2265 : : * Read records. No changes are generated in fast_forward mode,
2266 : : * but snapbuilder/slot statuses are updated properly.
2267 : : */
2268 : 2878 : record = XLogReadRecord(ctx->reader, &errm);
2269 [ - + ]: 2878 : if (errm)
2270 [ # # ]: 0 : elog(ERROR, "could not find record while advancing replication slot: %s",
2271 : : errm);
2272 : :
2273 : : /*
2274 : : * Process the record. Storage-level changes are ignored in
2275 : : * fast_forward mode, but other modules (such as snapbuilder)
2276 : : * might still have critical updates to do.
2277 : : */
2278 [ + - ]: 2878 : if (record)
2279 : 2878 : LogicalDecodingProcessRecord(ctx, ctx->reader);
2280 : :
2281 [ - + ]: 2878 : CHECK_FOR_INTERRUPTS();
2282 : : }
2283 : :
2284 [ + + + - ]: 22 : if (found_consistent_snapshot && DecodingContextReady(ctx))
2285 : 11 : *found_consistent_snapshot = true;
2286 : :
2287 [ + - ]: 22 : if (XLogRecPtrIsValid(ctx->reader->EndRecPtr))
2288 : : {
2289 : 22 : LogicalConfirmReceivedLocation(moveto);
2290 : :
2291 : : /*
2292 : : * If only the confirmed_flush LSN has changed the slot won't get
2293 : : * marked as dirty by the above. Callers on the walsender
2294 : : * interface are expected to keep track of their own progress and
2295 : : * don't need it written out. But SQL-interface users cannot
2296 : : * specify their own start positions and it's harder for them to
2297 : : * keep track of their progress, so we should make more of an
2298 : : * effort to save it for them.
2299 : : *
2300 : : * Dirty the slot so it is written out at the next checkpoint. The
2301 : : * LSN position advanced to may still be lost on a crash but this
2302 : : * makes the data consistent after a clean shutdown.
2303 : : */
2304 : 22 : ReplicationSlotMarkDirty();
2305 : : }
2306 : :
2307 : 22 : retlsn = MyReplicationSlot->data.confirmed_flush;
2308 : :
2309 : : /* free context, call shutdown callback */
2310 : 22 : FreeDecodingContext(ctx);
2311 : :
2312 : 22 : InvalidateSystemCaches();
2313 : : }
2314 : 0 : PG_CATCH();
2315 : : {
2316 : : /* clear all timetravel entries */
2317 : 0 : InvalidateSystemCaches();
2318 : :
2319 : 0 : PG_RE_THROW();
2320 : : }
2321 [ - + ]: 22 : PG_END_TRY();
2322 : :
2323 : 22 : return retlsn;
2324 : : }
|
