Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * test_decoding.c
4 : : * example logical decoding output plugin
5 : : *
6 : : * Copyright (c) 2012-2026, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * contrib/test_decoding/test_decoding.c
10 : : *
11 : : *-------------------------------------------------------------------------
12 : : */
13 : : #include "postgres.h"
14 : :
15 : : #include "catalog/pg_type.h"
16 : :
17 : : #include "replication/logical.h"
18 : : #include "replication/origin.h"
19 : :
20 : : #include "utils/builtins.h"
21 : : #include "utils/lsyscache.h"
22 : : #include "utils/memutils.h"
23 : : #include "utils/rel.h"
24 : :
519 tgl@sss.pgh.pa.us 25 :CBC 137 : PG_MODULE_MAGIC_EXT(
26 : : .name = "test_decoding",
27 : : .version = PG_VERSION
28 : : );
29 : :
30 : : typedef struct
31 : : {
32 : : MemoryContext context;
33 : : bool include_xids;
34 : : bool include_timestamp;
35 : : bool skip_empty_xacts;
36 : : bool only_local;
37 : : } TestDecodingData;
38 : :
39 : : /*
40 : : * Maintain the per-transaction level variables to track whether the
41 : : * transaction and or streams have written any changes. In streaming mode the
42 : : * transaction can be decoded in streams so along with maintaining whether the
43 : : * transaction has written any changes, we also need to track whether the
44 : : * current stream has written any changes. This is required so that if user
45 : : * has requested to skip the empty transactions we can skip the empty streams
46 : : * even though the transaction has written some changes.
47 : : */
48 : : typedef struct
49 : : {
50 : : bool xact_wrote_changes;
51 : : bool stream_wrote_changes;
52 : : } TestDecodingTxnData;
53 : :
54 : : static void pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
55 : : bool is_init);
56 : : static void pg_decode_shutdown(LogicalDecodingContext *ctx);
57 : : static void pg_decode_begin_txn(LogicalDecodingContext *ctx,
58 : : ReorderBufferTXN *txn);
59 : : static void pg_output_begin(LogicalDecodingContext *ctx,
60 : : TestDecodingData *data,
61 : : ReorderBufferTXN *txn,
62 : : bool last_write);
63 : : static void pg_decode_commit_txn(LogicalDecodingContext *ctx,
64 : : ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
65 : : static void pg_decode_change(LogicalDecodingContext *ctx,
66 : : ReorderBufferTXN *txn, Relation relation,
67 : : ReorderBufferChange *change);
68 : : static void pg_decode_truncate(LogicalDecodingContext *ctx,
69 : : ReorderBufferTXN *txn,
70 : : int nrelations, Relation relations[],
71 : : ReorderBufferChange *change);
72 : : static bool pg_decode_filter(LogicalDecodingContext *ctx,
73 : : ReplOriginId origin_id);
74 : : static void pg_decode_message(LogicalDecodingContext *ctx,
75 : : ReorderBufferTXN *txn, XLogRecPtr lsn,
76 : : bool transactional, const char *prefix,
77 : : Size sz, const char *message);
78 : : static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
79 : : TransactionId xid,
80 : : const char *gid);
81 : : static void pg_decode_begin_prepare_txn(LogicalDecodingContext *ctx,
82 : : ReorderBufferTXN *txn);
83 : : static void pg_decode_prepare_txn(LogicalDecodingContext *ctx,
84 : : ReorderBufferTXN *txn,
85 : : XLogRecPtr prepare_lsn);
86 : : static void pg_decode_commit_prepared_txn(LogicalDecodingContext *ctx,
87 : : ReorderBufferTXN *txn,
88 : : XLogRecPtr commit_lsn);
89 : : static void pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
90 : : ReorderBufferTXN *txn,
91 : : XLogRecPtr prepare_end_lsn,
92 : : TimestampTz prepare_time);
93 : : static void pg_decode_stream_start(LogicalDecodingContext *ctx,
94 : : ReorderBufferTXN *txn);
95 : : static void pg_output_stream_start(LogicalDecodingContext *ctx,
96 : : TestDecodingData *data,
97 : : ReorderBufferTXN *txn,
98 : : bool last_write);
99 : : static void pg_decode_stream_stop(LogicalDecodingContext *ctx,
100 : : ReorderBufferTXN *txn);
101 : : static void pg_decode_stream_abort(LogicalDecodingContext *ctx,
102 : : ReorderBufferTXN *txn,
103 : : XLogRecPtr abort_lsn);
104 : : static void pg_decode_stream_prepare(LogicalDecodingContext *ctx,
105 : : ReorderBufferTXN *txn,
106 : : XLogRecPtr prepare_lsn);
107 : : static void pg_decode_stream_commit(LogicalDecodingContext *ctx,
108 : : ReorderBufferTXN *txn,
109 : : XLogRecPtr commit_lsn);
110 : : static void pg_decode_stream_change(LogicalDecodingContext *ctx,
111 : : ReorderBufferTXN *txn,
112 : : Relation relation,
113 : : ReorderBufferChange *change);
114 : : static void pg_decode_stream_message(LogicalDecodingContext *ctx,
115 : : ReorderBufferTXN *txn, XLogRecPtr lsn,
116 : : bool transactional, const char *prefix,
117 : : Size sz, const char *message);
118 : : static void pg_decode_stream_truncate(LogicalDecodingContext *ctx,
119 : : ReorderBufferTXN *txn,
120 : : int nrelations, Relation relations[],
121 : : ReorderBufferChange *change);
122 : :
123 : : void
4560 rhaas@postgresql.org 124 : 137 : _PG_init(void)
125 : : {
126 : : /* other plugins can perform things here */
127 : 137 : }
128 : :
129 : : /* specify output plugin callbacks */
130 : : void
131 : 371 : _PG_output_plugin_init(OutputPluginCallbacks *cb)
132 : : {
133 : 371 : cb->startup_cb = pg_decode_startup;
134 : 371 : cb->begin_cb = pg_decode_begin_txn;
135 : 371 : cb->change_cb = pg_decode_change;
3064 peter_e@gmx.net 136 : 371 : cb->truncate_cb = pg_decode_truncate;
4560 rhaas@postgresql.org 137 : 371 : cb->commit_cb = pg_decode_commit_txn;
4138 andres@anarazel.de 138 : 371 : cb->filter_by_origin_cb = pg_decode_filter;
4560 rhaas@postgresql.org 139 : 371 : cb->shutdown_cb = pg_decode_shutdown;
3795 simon@2ndQuadrant.co 140 : 371 : cb->message_cb = pg_decode_message;
2066 akapila@postgresql.o 141 : 371 : cb->filter_prepare_cb = pg_decode_filter_prepare;
142 : 371 : cb->begin_prepare_cb = pg_decode_begin_prepare_txn;
143 : 371 : cb->prepare_cb = pg_decode_prepare_txn;
144 : 371 : cb->commit_prepared_cb = pg_decode_commit_prepared_txn;
145 : 371 : cb->rollback_prepared_cb = pg_decode_rollback_prepared_txn;
2221 146 : 371 : cb->stream_start_cb = pg_decode_stream_start;
147 : 371 : cb->stream_stop_cb = pg_decode_stream_stop;
148 : 371 : cb->stream_abort_cb = pg_decode_stream_abort;
2066 149 : 371 : cb->stream_prepare_cb = pg_decode_stream_prepare;
2221 150 : 371 : cb->stream_commit_cb = pg_decode_stream_commit;
151 : 371 : cb->stream_change_cb = pg_decode_stream_change;
152 : 371 : cb->stream_message_cb = pg_decode_stream_message;
153 : 371 : cb->stream_truncate_cb = pg_decode_stream_truncate;
4560 rhaas@postgresql.org 154 : 371 : }
155 : :
156 : :
157 : : /* initialize this plugin */
158 : : static void
159 : 371 : pg_decode_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
160 : : bool is_init)
161 : : {
162 : : ListCell *option;
163 : : TestDecodingData *data;
2210 akapila@postgresql.o 164 : 371 : bool enable_streaming = false;
165 : :
265 michael@paquier.xyz 166 : 371 : data = palloc0_object(TestDecodingData);
4560 rhaas@postgresql.org 167 : 371 : data->context = AllocSetContextCreate(ctx->context,
168 : : "text conversion context",
169 : : ALLOCSET_DEFAULT_SIZES);
170 : 371 : data->include_xids = true;
171 : 371 : data->include_timestamp = false;
4378 andres@anarazel.de 172 : 371 : data->skip_empty_xacts = false;
4138 173 : 371 : data->only_local = false;
174 : :
4560 rhaas@postgresql.org 175 : 371 : ctx->output_plugin_private = data;
176 : :
177 : 371 : opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
3081 peter_e@gmx.net 178 : 371 : opt->receive_rewrites = false;
179 : :
4560 rhaas@postgresql.org 180 [ + + + + : 750 : foreach(option, ctx->output_plugin_options)
+ + ]
181 : : {
182 : 382 : DefElem *elem = lfirst(option);
183 : :
184 [ + - - + ]: 382 : Assert(elem->arg == NULL || IsA(elem->arg, String));
185 : :
186 [ + + ]: 382 : if (strcmp(elem->defname, "include-xids") == 0)
187 : : {
188 : : /* if option does not provide a value, it means its value is true */
189 [ - + ]: 180 : if (elem->arg == NULL)
4560 rhaas@postgresql.org 190 :UBC 0 : data->include_xids = true;
4560 rhaas@postgresql.org 191 [ + + ]:CBC 180 : else if (!parse_bool(strVal(elem->arg), &data->include_xids))
192 [ + - ]: 2 : ereport(ERROR,
193 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
194 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
195 : : strVal(elem->arg), elem->defname)));
196 : : }
197 [ + + ]: 202 : else if (strcmp(elem->defname, "include-timestamp") == 0)
198 : : {
199 [ - + ]: 1 : if (elem->arg == NULL)
4560 rhaas@postgresql.org 200 :UBC 0 : data->include_timestamp = true;
4560 rhaas@postgresql.org 201 [ - + ]:CBC 1 : else if (!parse_bool(strVal(elem->arg), &data->include_timestamp))
4560 rhaas@postgresql.org 202 [ # # ]:UBC 0 : ereport(ERROR,
203 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
204 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
205 : : strVal(elem->arg), elem->defname)));
206 : : }
4560 rhaas@postgresql.org 207 [ + + ]:CBC 201 : else if (strcmp(elem->defname, "force-binary") == 0)
208 : : {
209 : : bool force_binary;
210 : :
211 [ - + ]: 6 : if (elem->arg == NULL)
4560 rhaas@postgresql.org 212 :UBC 0 : continue;
4560 rhaas@postgresql.org 213 [ - + ]:CBC 6 : else if (!parse_bool(strVal(elem->arg), &force_binary))
4560 rhaas@postgresql.org 214 [ # # ]:UBC 0 : ereport(ERROR,
215 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
216 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
217 : : strVal(elem->arg), elem->defname)));
218 : :
4560 rhaas@postgresql.org 219 [ + + ]:CBC 6 : if (force_binary)
220 : 2 : opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
221 : : }
4378 andres@anarazel.de 222 [ + + ]: 195 : else if (strcmp(elem->defname, "skip-empty-xacts") == 0)
223 : : {
224 : :
225 [ - + ]: 178 : if (elem->arg == NULL)
4378 andres@anarazel.de 226 :UBC 0 : data->skip_empty_xacts = true;
4378 andres@anarazel.de 227 [ - + ]:CBC 178 : else if (!parse_bool(strVal(elem->arg), &data->skip_empty_xacts))
4378 andres@anarazel.de 228 [ # # ]:UBC 0 : ereport(ERROR,
229 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
230 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
231 : : strVal(elem->arg), elem->defname)));
232 : : }
4138 andres@anarazel.de 233 [ + + ]:CBC 17 : else if (strcmp(elem->defname, "only-local") == 0)
234 : : {
235 : :
236 [ - + ]: 3 : if (elem->arg == NULL)
4138 andres@anarazel.de 237 :UBC 0 : data->only_local = true;
4138 andres@anarazel.de 238 [ - + ]:CBC 3 : else if (!parse_bool(strVal(elem->arg), &data->only_local))
4138 andres@anarazel.de 239 [ # # ]:UBC 0 : ereport(ERROR,
240 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
241 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
242 : : strVal(elem->arg), elem->defname)));
243 : : }
3081 peter_e@gmx.net 244 [ + + ]:CBC 14 : else if (strcmp(elem->defname, "include-rewrites") == 0)
245 : : {
246 : :
247 [ - + ]: 1 : if (elem->arg == NULL)
3081 peter_e@gmx.net 248 :UBC 0 : continue;
3081 peter_e@gmx.net 249 [ - + ]:CBC 1 : else if (!parse_bool(strVal(elem->arg), &opt->receive_rewrites))
3081 peter_e@gmx.net 250 [ # # ]:UBC 0 : ereport(ERROR,
251 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
252 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
253 : : strVal(elem->arg), elem->defname)));
254 : : }
2210 akapila@postgresql.o 255 [ + + ]:CBC 13 : else if (strcmp(elem->defname, "stream-changes") == 0)
256 : : {
257 [ - + ]: 12 : if (elem->arg == NULL)
2210 akapila@postgresql.o 258 :UBC 0 : continue;
2210 akapila@postgresql.o 259 [ - + ]:CBC 12 : else if (!parse_bool(strVal(elem->arg), &enable_streaming))
2210 akapila@postgresql.o 260 [ # # ]:UBC 0 : ereport(ERROR,
261 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
262 : : errmsg("could not parse value \"%s\" for parameter \"%s\"",
263 : : strVal(elem->arg), elem->defname)));
264 : : }
265 : : else
266 : : {
4560 rhaas@postgresql.org 267 [ + - + - ]:CBC 1 : ereport(ERROR,
268 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
269 : : errmsg("option \"%s\" = \"%s\" is unknown",
270 : : elem->defname,
271 : : elem->arg ? strVal(elem->arg) : "(null)")));
272 : : }
273 : : }
274 : :
2210 akapila@postgresql.o 275 : 368 : ctx->streaming &= enable_streaming;
4560 rhaas@postgresql.org 276 : 368 : }
277 : :
278 : : /* cleanup this plugin's resources */
279 : : static void
280 : 357 : pg_decode_shutdown(LogicalDecodingContext *ctx)
281 : : {
282 : 357 : TestDecodingData *data = ctx->output_plugin_private;
283 : :
284 : : /* cleanup our own resources via memory context reset */
285 : 357 : MemoryContextDelete(data->context);
286 : 357 : }
287 : :
288 : : /* BEGIN callback */
289 : : static void
290 : 443 : pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
291 : : {
292 : 443 : TestDecodingData *data = ctx->output_plugin_private;
293 : : TestDecodingTxnData *txndata =
1196 tgl@sss.pgh.pa.us 294 : 443 : MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
295 : :
2109 akapila@postgresql.o 296 : 443 : txndata->xact_wrote_changes = false;
297 : 443 : txn->output_plugin_private = txndata;
298 : :
299 : : /*
300 : : * If asked to skip empty transactions, we'll emit BEGIN at the point
301 : : * where the first operation is received for this transaction.
302 : : */
4378 andres@anarazel.de 303 [ + + ]: 443 : if (data->skip_empty_xacts)
304 : 399 : return;
305 : :
306 : 44 : pg_output_begin(ctx, data, txn, true);
307 : : }
308 : :
309 : : static void
310 : 276 : pg_output_begin(LogicalDecodingContext *ctx, TestDecodingData *data, ReorderBufferTXN *txn, bool last_write)
311 : : {
312 : 276 : OutputPluginPrepareWrite(ctx, last_write);
4560 rhaas@postgresql.org 313 [ + + ]: 276 : if (data->include_xids)
314 : 40 : appendStringInfo(ctx->out, "BEGIN %u", txn->xid);
315 : : else
316 : 236 : appendStringInfoString(ctx->out, "BEGIN");
4378 andres@anarazel.de 317 : 276 : OutputPluginWrite(ctx, last_write);
4560 rhaas@postgresql.org 318 : 276 : }
319 : :
320 : : /* COMMIT callback */
321 : : static void
322 : 443 : pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
323 : : XLogRecPtr commit_lsn)
324 : : {
325 : 443 : TestDecodingData *data = ctx->output_plugin_private;
2109 akapila@postgresql.o 326 : 443 : TestDecodingTxnData *txndata = txn->output_plugin_private;
327 : 443 : bool xact_wrote_changes = txndata->xact_wrote_changes;
328 : :
329 : 443 : pfree(txndata);
330 : 443 : txn->output_plugin_private = NULL;
331 : :
332 [ + + + + ]: 443 : if (data->skip_empty_xacts && !xact_wrote_changes)
4378 andres@anarazel.de 333 : 175 : return;
334 : :
4560 rhaas@postgresql.org 335 : 268 : OutputPluginPrepareWrite(ctx, true);
336 [ + + ]: 268 : if (data->include_xids)
337 : 39 : appendStringInfo(ctx->out, "COMMIT %u", txn->xid);
338 : : else
339 : 229 : appendStringInfoString(ctx->out, "COMMIT");
340 : :
341 [ + + ]: 268 : if (data->include_timestamp)
342 : 1 : appendStringInfo(ctx->out, " (at %s)",
343 : : timestamptz_to_str(txn->commit_time));
344 : :
345 : 268 : OutputPluginWrite(ctx, true);
346 : : }
347 : :
348 : : /* BEGIN PREPARE callback */
349 : : static void
2066 akapila@postgresql.o 350 : 9 : pg_decode_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
351 : : {
352 : 9 : TestDecodingData *data = ctx->output_plugin_private;
353 : : TestDecodingTxnData *txndata =
1196 tgl@sss.pgh.pa.us 354 : 9 : MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
355 : :
2066 akapila@postgresql.o 356 : 9 : txndata->xact_wrote_changes = false;
357 : 9 : txn->output_plugin_private = txndata;
358 : :
359 : : /*
360 : : * If asked to skip empty transactions, we'll emit BEGIN at the point
361 : : * where the first operation is received for this transaction.
362 : : */
363 [ + + ]: 9 : if (data->skip_empty_xacts)
364 : 8 : return;
365 : :
366 : 1 : pg_output_begin(ctx, data, txn, true);
367 : : }
368 : :
369 : : /* PREPARE callback */
370 : : static void
371 : 9 : pg_decode_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
372 : : XLogRecPtr prepare_lsn)
373 : : {
374 : 9 : TestDecodingData *data = ctx->output_plugin_private;
375 : 9 : TestDecodingTxnData *txndata = txn->output_plugin_private;
376 : :
377 : : /*
378 : : * If asked to skip empty transactions, we'll emit PREPARE at the point
379 : : * where the first operation is received for this transaction.
380 : : */
381 [ + + + + ]: 9 : if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
382 : 1 : return;
383 : :
384 : 8 : OutputPluginPrepareWrite(ctx, true);
385 : :
386 : 8 : appendStringInfo(ctx->out, "PREPARE TRANSACTION %s",
387 : 8 : quote_literal_cstr(txn->gid));
388 : :
389 [ + + ]: 8 : if (data->include_xids)
390 : 1 : appendStringInfo(ctx->out, ", txid %u", txn->xid);
391 : :
392 [ - + ]: 8 : if (data->include_timestamp)
2066 akapila@postgresql.o 393 :UBC 0 : appendStringInfo(ctx->out, " (at %s)",
394 : : timestamptz_to_str(txn->prepare_time));
395 : :
2066 akapila@postgresql.o 396 :CBC 8 : OutputPluginWrite(ctx, true);
397 : : }
398 : :
399 : : /* COMMIT PREPARED callback */
400 : : static void
401 : 8 : pg_decode_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
402 : : XLogRecPtr commit_lsn)
403 : : {
404 : 8 : TestDecodingData *data = ctx->output_plugin_private;
405 : :
406 : 8 : OutputPluginPrepareWrite(ctx, true);
407 : :
408 : 8 : appendStringInfo(ctx->out, "COMMIT PREPARED %s",
409 : 8 : quote_literal_cstr(txn->gid));
410 : :
411 [ + + ]: 8 : if (data->include_xids)
412 : 1 : appendStringInfo(ctx->out, ", txid %u", txn->xid);
413 : :
414 [ - + ]: 8 : if (data->include_timestamp)
2066 akapila@postgresql.o 415 :UBC 0 : appendStringInfo(ctx->out, " (at %s)",
416 : : timestamptz_to_str(txn->commit_time));
417 : :
2066 akapila@postgresql.o 418 :CBC 8 : OutputPluginWrite(ctx, true);
419 : 8 : }
420 : :
421 : : /* ROLLBACK PREPARED callback */
422 : : static void
423 : 2 : pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
424 : : ReorderBufferTXN *txn,
425 : : XLogRecPtr prepare_end_lsn,
426 : : TimestampTz prepare_time)
427 : : {
428 : 2 : TestDecodingData *data = ctx->output_plugin_private;
429 : :
430 : 2 : OutputPluginPrepareWrite(ctx, true);
431 : :
432 : 2 : appendStringInfo(ctx->out, "ROLLBACK PREPARED %s",
433 : 2 : quote_literal_cstr(txn->gid));
434 : :
435 [ - + ]: 2 : if (data->include_xids)
2066 akapila@postgresql.o 436 :UBC 0 : appendStringInfo(ctx->out, ", txid %u", txn->xid);
437 : :
2066 akapila@postgresql.o 438 [ - + ]:CBC 2 : if (data->include_timestamp)
2066 akapila@postgresql.o 439 :UBC 0 : appendStringInfo(ctx->out, " (at %s)",
440 : : timestamptz_to_str(txn->commit_time));
441 : :
2066 akapila@postgresql.o 442 :CBC 2 : OutputPluginWrite(ctx, true);
443 : 2 : }
444 : :
445 : : /*
446 : : * Filter out two-phase transactions.
447 : : *
448 : : * Each plugin can implement its own filtering logic. Here we demonstrate a
449 : : * simple logic by checking the GID. If the GID contains the "_nodecode"
450 : : * substring, then we filter it out.
451 : : */
452 : : static bool
1976 453 : 170 : pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid,
454 : : const char *gid)
455 : : {
2066 456 [ + + ]: 170 : if (strstr(gid, "_nodecode") != NULL)
457 : 14 : return true;
458 : :
459 : 156 : return false;
460 : : }
461 : :
462 : : static bool
4138 andres@anarazel.de 463 : 1243103 : pg_decode_filter(LogicalDecodingContext *ctx,
464 : : ReplOriginId origin_id)
465 : : {
466 : 1243103 : TestDecodingData *data = ctx->output_plugin_private;
467 : :
211 msawada@postgresql.o 468 [ + + + + ]: 1243103 : if (data->only_local && origin_id != InvalidReplOriginId)
4138 andres@anarazel.de 469 : 9 : return true;
470 : 1243094 : return false;
471 : : }
472 : :
473 : : /*
474 : : * Print literal `outputstr' already represented as string of type `typid'
475 : : * into stringbuf `s'.
476 : : *
477 : : * Some builtin types aren't quoted, the rest is quoted. Escaping is done
478 : : * per standard SQL rules.
479 : : */
480 : : static void
4560 rhaas@postgresql.org 481 : 176087 : print_literal(StringInfo s, Oid typid, char *outputstr)
482 : : {
483 : : const char *valptr;
484 : :
485 [ + - - + ]: 176087 : switch (typid)
486 : : {
487 : 60325 : case INT2OID:
488 : : case INT4OID:
489 : : case INT8OID:
490 : : case OIDOID:
491 : : case FLOAT4OID:
492 : : case FLOAT8OID:
493 : : case NUMERICOID:
494 : : /* NB: We don't care about Inf, NaN et al. */
495 : 60325 : appendStringInfoString(s, outputstr);
496 : 60325 : break;
497 : :
4560 rhaas@postgresql.org 498 :UBC 0 : case BITOID:
499 : : case VARBITOID:
500 : 0 : appendStringInfo(s, "B'%s'", outputstr);
501 : 0 : break;
502 : :
503 : 0 : case BOOLOID:
504 [ # # ]: 0 : if (strcmp(outputstr, "t") == 0)
505 : 0 : appendStringInfoString(s, "true");
506 : : else
507 : 0 : appendStringInfoString(s, "false");
508 : 0 : break;
509 : :
4560 rhaas@postgresql.org 510 :CBC 115762 : default:
511 : 115762 : appendStringInfoChar(s, '\'');
512 [ + + ]: 5427563 : for (valptr = outputstr; *valptr; valptr++)
513 : : {
514 : 5311801 : char ch = *valptr;
515 : :
516 [ + + ]: 5311801 : if (SQL_STR_DOUBLE(ch, false))
517 : 64 : appendStringInfoChar(s, ch);
518 : 5311801 : appendStringInfoChar(s, ch);
519 : : }
520 : 115762 : appendStringInfoChar(s, '\'');
521 : 115762 : break;
522 : : }
523 : 176087 : }
524 : :
525 : : /* print the tuple 'tuple' into the StringInfo s */
526 : : static void
527 : 145629 : tuple_to_stringinfo(StringInfo s, TupleDesc tupdesc, HeapTuple tuple, bool skip_nulls)
528 : : {
529 : : int natt;
530 : :
531 : : /* print all columns individually */
532 [ + + ]: 347480 : for (natt = 0; natt < tupdesc->natts; natt++)
533 : : {
534 : : Form_pg_attribute attr; /* the attribute itself */
535 : : Oid typid; /* type of current attribute */
536 : : Oid typoutput; /* output function */
537 : : bool typisvarlena;
538 : : Datum origval; /* possibly toasted Datum */
539 : : bool isnull; /* column is null? */
540 : :
3294 andres@anarazel.de 541 : 201851 : attr = TupleDescAttr(tupdesc, natt);
542 : :
543 : : /*
544 : : * don't print dropped columns, we can't be sure everything is
545 : : * available for them
546 : : */
4560 rhaas@postgresql.org 547 [ + + ]: 201851 : if (attr->attisdropped)
548 : 5135 : continue;
549 : :
550 : : /*
551 : : * Don't print system columns, oid will already have been printed if
552 : : * present.
553 : : */
554 [ - + ]: 201779 : if (attr->attnum < 0)
4560 rhaas@postgresql.org 555 :UBC 0 : continue;
556 : :
557 : : /*
558 : : * Virtual generated columns are always stored as null in the tuple,
559 : : * so don't print them at all. A printed null would not be
560 : : * distinguishable from a column that really contains a null. Stored
561 : : * generated columns are printed as usual since their values are
562 : : * actually on disk.
563 : : */
9 msawada@postgresql.o 564 [ + + ]:GNC 201779 : if (attr->attgenerated == ATTRIBUTE_GENERATED_VIRTUAL)
565 : 1 : continue;
566 : :
4560 rhaas@postgresql.org 567 :CBC 201778 : typid = attr->atttypid;
568 : :
569 : : /* get Datum from tuple */
4079 andres@anarazel.de 570 : 201778 : origval = heap_getattr(tuple, natt + 1, tupdesc, &isnull);
571 : :
4560 rhaas@postgresql.org 572 [ + + + + ]: 201778 : if (isnull && skip_nulls)
573 : 5062 : continue;
574 : :
575 : : /* print attribute name */
576 : 196716 : appendStringInfoChar(s, ' ');
577 : 196716 : appendStringInfoString(s, quote_identifier(NameStr(attr->attname)));
578 : :
579 : : /* print attribute type */
580 : 196716 : appendStringInfoChar(s, '[');
581 : 196716 : appendStringInfoString(s, format_type_be(typid));
582 : 196716 : appendStringInfoChar(s, ']');
583 : :
584 : : /* query output function */
585 : 196716 : getTypeOutputInfo(typid,
586 : : &typoutput, &typisvarlena);
587 : :
588 : : /* print separator */
589 : 196716 : appendStringInfoChar(s, ':');
590 : :
591 : : /* print data */
592 [ + + ]: 196716 : if (isnull)
593 : 20617 : appendStringInfoString(s, "null");
387 peter@eisentraut.org 594 [ + + + + ]: 176099 : else if (typisvarlena && VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(origval)))
4560 rhaas@postgresql.org 595 : 12 : appendStringInfoString(s, "unchanged-toast-datum");
596 [ + + ]: 176087 : else if (!typisvarlena)
597 : 60329 : print_literal(s, typid,
598 : : OidOutputFunctionCall(typoutput, origval));
599 : : else
600 : : {
601 : : Datum val; /* definitely detoasted Datum */
602 : :
603 : 115758 : val = PointerGetDatum(PG_DETOAST_DATUM(origval));
604 : 115758 : print_literal(s, typid, OidOutputFunctionCall(typoutput, val));
605 : : }
606 : : }
607 : 145629 : }
608 : :
609 : : /*
610 : : * callback for individual changed tuples
611 : : */
612 : : static void
613 : 150617 : pg_decode_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
614 : : Relation relation, ReorderBufferChange *change)
615 : : {
616 : : TestDecodingData *data;
617 : : TestDecodingTxnData *txndata;
618 : : Form_pg_class class_form;
619 : : TupleDesc tupdesc;
620 : : MemoryContext old;
621 : :
622 : 150617 : data = ctx->output_plugin_private;
2109 akapila@postgresql.o 623 : 150617 : txndata = txn->output_plugin_private;
624 : :
625 : : /* output BEGIN if we haven't yet */
626 [ + + + + ]: 150617 : if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
627 : : {
4378 andres@anarazel.de 628 : 221 : pg_output_begin(ctx, data, txn, false);
629 : : }
2109 akapila@postgresql.o 630 : 150617 : txndata->xact_wrote_changes = true;
631 : :
4560 rhaas@postgresql.org 632 : 150617 : class_form = RelationGetForm(relation);
633 : 150617 : tupdesc = RelationGetDescr(relation);
634 : :
635 : : /* Avoid leaking memory by using and resetting our own context */
636 : 150617 : old = MemoryContextSwitchTo(data->context);
637 : :
638 : 150617 : OutputPluginPrepareWrite(ctx, true);
639 : :
640 : 150617 : appendStringInfoString(ctx->out, "table ");
641 : 150617 : appendStringInfoString(ctx->out,
2401 alvherre@alvh.no-ip. 642 : 150617 : quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
3081 peter_e@gmx.net 643 [ + + ]: 150617 : class_form->relrewrite ?
644 : 1 : get_rel_name(class_form->relrewrite) :
645 : : NameStr(class_form->relname)));
4126 646 : 150617 : appendStringInfoChar(ctx->out, ':');
647 : :
4560 rhaas@postgresql.org 648 [ + + + - ]: 150617 : switch (change->action)
649 : : {
650 : 133050 : case REORDER_BUFFER_CHANGE_INSERT:
651 : 133050 : appendStringInfoString(ctx->out, " INSERT:");
4556 tgl@sss.pgh.pa.us 652 [ - + ]: 133050 : if (change->data.tp.newtuple == NULL)
4560 rhaas@postgresql.org 653 :UBC 0 : appendStringInfoString(ctx->out, " (no-tuple-data)");
654 : : else
4560 rhaas@postgresql.org 655 :CBC 133050 : tuple_to_stringinfo(ctx->out, tupdesc,
656 : : change->data.tp.newtuple,
657 : : false);
658 : 133050 : break;
659 : 7544 : case REORDER_BUFFER_CHANGE_UPDATE:
660 : 7544 : appendStringInfoString(ctx->out, " UPDATE:");
4556 tgl@sss.pgh.pa.us 661 [ + + ]: 7544 : if (change->data.tp.oldtuple != NULL)
662 : : {
4560 rhaas@postgresql.org 663 : 19 : appendStringInfoString(ctx->out, " old-key:");
664 : 19 : tuple_to_stringinfo(ctx->out, tupdesc,
665 : : change->data.tp.oldtuple,
666 : : true);
667 : 19 : appendStringInfoString(ctx->out, " new-tuple:");
668 : : }
669 : :
4556 tgl@sss.pgh.pa.us 670 [ - + ]: 7544 : if (change->data.tp.newtuple == NULL)
4560 rhaas@postgresql.org 671 :UBC 0 : appendStringInfoString(ctx->out, " (no-tuple-data)");
672 : : else
4560 rhaas@postgresql.org 673 :CBC 7544 : tuple_to_stringinfo(ctx->out, tupdesc,
674 : : change->data.tp.newtuple,
675 : : false);
676 : 7544 : break;
677 : 10023 : case REORDER_BUFFER_CHANGE_DELETE:
678 : 10023 : appendStringInfoString(ctx->out, " DELETE:");
679 : :
680 : : /* if there was no PK, we only know that a delete happened */
4556 tgl@sss.pgh.pa.us 681 [ + + ]: 10023 : if (change->data.tp.oldtuple == NULL)
4560 rhaas@postgresql.org 682 : 5007 : appendStringInfoString(ctx->out, " (no-tuple-data)");
683 : : /* In DELETE, only the replica identity is present; display that */
684 : : else
685 : 5016 : tuple_to_stringinfo(ctx->out, tupdesc,
686 : : change->data.tp.oldtuple,
687 : : true);
688 : 10023 : break;
4556 tgl@sss.pgh.pa.us 689 :UBC 0 : default:
690 : 0 : Assert(false);
691 : : }
692 : :
4560 rhaas@postgresql.org 693 :CBC 150617 : MemoryContextSwitchTo(old);
694 : 150617 : MemoryContextReset(data->context);
695 : :
696 : 150617 : OutputPluginWrite(ctx, true);
697 : 150617 : }
698 : :
699 : : static void
3064 peter_e@gmx.net 700 : 8 : pg_decode_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
701 : : int nrelations, Relation relations[], ReorderBufferChange *change)
702 : : {
703 : : TestDecodingData *data;
704 : : TestDecodingTxnData *txndata;
705 : : MemoryContext old;
706 : : int i;
707 : :
708 : 8 : data = ctx->output_plugin_private;
2109 akapila@postgresql.o 709 : 8 : txndata = txn->output_plugin_private;
710 : :
711 : : /* output BEGIN if we haven't yet */
712 [ + + + - ]: 8 : if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
713 : : {
3064 peter_e@gmx.net 714 : 7 : pg_output_begin(ctx, data, txn, false);
715 : : }
2109 akapila@postgresql.o 716 : 8 : txndata->xact_wrote_changes = true;
717 : :
718 : : /* Avoid leaking memory by using and resetting our own context */
3064 peter_e@gmx.net 719 : 8 : old = MemoryContextSwitchTo(data->context);
720 : :
721 : 8 : OutputPluginPrepareWrite(ctx, true);
722 : :
723 : 8 : appendStringInfoString(ctx->out, "table ");
724 : :
725 [ + + ]: 17 : for (i = 0; i < nrelations; i++)
726 : : {
727 [ + + ]: 9 : if (i > 0)
728 : 1 : appendStringInfoString(ctx->out, ", ");
729 : :
730 : 9 : appendStringInfoString(ctx->out,
731 : 9 : quote_qualified_identifier(get_namespace_name(relations[i]->rd_rel->relnamespace),
732 : 9 : NameStr(relations[i]->rd_rel->relname)));
733 : : }
734 : :
735 : 8 : appendStringInfoString(ctx->out, ": TRUNCATE:");
736 : :
737 [ + + ]: 8 : if (change->data.truncate.restart_seqs
738 [ - + ]: 7 : || change->data.truncate.cascade)
739 : : {
740 [ + - ]: 1 : if (change->data.truncate.restart_seqs)
2611 drowley@postgresql.o 741 : 1 : appendStringInfoString(ctx->out, " restart_seqs");
3064 peter_e@gmx.net 742 [ + - ]: 1 : if (change->data.truncate.cascade)
2611 drowley@postgresql.o 743 : 1 : appendStringInfoString(ctx->out, " cascade");
744 : : }
745 : : else
3064 peter_e@gmx.net 746 : 7 : appendStringInfoString(ctx->out, " (no-flags)");
747 : :
748 : 8 : MemoryContextSwitchTo(old);
749 : 8 : MemoryContextReset(data->context);
750 : :
751 : 8 : OutputPluginWrite(ctx, true);
752 : 8 : }
753 : :
754 : : static void
3795 simon@2ndQuadrant.co 755 : 9 : pg_decode_message(LogicalDecodingContext *ctx,
756 : : ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional,
757 : : const char *prefix, Size sz, const char *message)
758 : : {
1143 akapila@postgresql.o 759 : 9 : TestDecodingData *data = ctx->output_plugin_private;
760 : : TestDecodingTxnData *txndata;
761 : :
762 [ + + ]: 9 : txndata = transactional ? txn->output_plugin_private : NULL;
763 : :
764 : : /* output BEGIN if we haven't yet for transactional messages */
765 [ + + + - : 9 : if (transactional && data->skip_empty_xacts && !txndata->xact_wrote_changes)
+ + ]
766 : 3 : pg_output_begin(ctx, data, txn, false);
767 : :
768 [ + + ]: 9 : if (transactional)
769 : 5 : txndata->xact_wrote_changes = true;
770 : :
3795 simon@2ndQuadrant.co 771 : 9 : OutputPluginPrepareWrite(ctx, true);
772 : 9 : appendStringInfo(ctx->out, "message: transactional: %d prefix: %s, sz: %zu content:",
773 : : transactional, prefix, sz);
774 : 9 : appendBinaryStringInfo(ctx->out, message, sz);
775 : 9 : OutputPluginWrite(ctx, true);
776 : 9 : }
777 : :
778 : : static void
2221 akapila@postgresql.o 779 : 40 : pg_decode_stream_start(LogicalDecodingContext *ctx,
780 : : ReorderBufferTXN *txn)
781 : : {
782 : 40 : TestDecodingData *data = ctx->output_plugin_private;
2109 783 : 40 : TestDecodingTxnData *txndata = txn->output_plugin_private;
784 : :
785 : : /*
786 : : * Allocate the txn plugin data for the first stream in the transaction.
787 : : */
788 [ + + ]: 40 : if (txndata == NULL)
789 : : {
790 : : txndata =
791 : 9 : MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
792 : 9 : txndata->xact_wrote_changes = false;
793 : 9 : txn->output_plugin_private = txndata;
794 : : }
795 : :
796 : 40 : txndata->stream_wrote_changes = false;
2176 797 [ + - ]: 40 : if (data->skip_empty_xacts)
798 : 40 : return;
2176 akapila@postgresql.o 799 :UBC 0 : pg_output_stream_start(ctx, data, txn, true);
800 : : }
801 : :
802 : : static void
2176 akapila@postgresql.o 803 :CBC 11 : pg_output_stream_start(LogicalDecodingContext *ctx, TestDecodingData *data, ReorderBufferTXN *txn, bool last_write)
804 : : {
805 : 11 : OutputPluginPrepareWrite(ctx, last_write);
2221 806 [ - + ]: 11 : if (data->include_xids)
2221 akapila@postgresql.o 807 :UBC 0 : appendStringInfo(ctx->out, "opening a streamed block for transaction TXN %u", txn->xid);
808 : : else
2142 drowley@postgresql.o 809 :CBC 11 : appendStringInfoString(ctx->out, "opening a streamed block for transaction");
2176 akapila@postgresql.o 810 : 11 : OutputPluginWrite(ctx, last_write);
2221 811 : 11 : }
812 : :
813 : : static void
814 : 40 : pg_decode_stream_stop(LogicalDecodingContext *ctx,
815 : : ReorderBufferTXN *txn)
816 : : {
817 : 40 : TestDecodingData *data = ctx->output_plugin_private;
2109 818 : 40 : TestDecodingTxnData *txndata = txn->output_plugin_private;
819 : :
820 [ + - + + ]: 40 : if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
2176 821 : 29 : return;
822 : :
2221 823 : 11 : OutputPluginPrepareWrite(ctx, true);
824 [ - + ]: 11 : if (data->include_xids)
2221 akapila@postgresql.o 825 :UBC 0 : appendStringInfo(ctx->out, "closing a streamed block for transaction TXN %u", txn->xid);
826 : : else
2142 drowley@postgresql.o 827 :CBC 11 : appendStringInfoString(ctx->out, "closing a streamed block for transaction");
2221 akapila@postgresql.o 828 : 11 : OutputPluginWrite(ctx, true);
829 : : }
830 : :
831 : : static void
832 : 4 : pg_decode_stream_abort(LogicalDecodingContext *ctx,
833 : : ReorderBufferTXN *txn,
834 : : XLogRecPtr abort_lsn)
835 : : {
836 : 4 : TestDecodingData *data = ctx->output_plugin_private;
837 : :
838 : : /*
839 : : * stream abort can be sent for an individual subtransaction but we
840 : : * maintain the output_plugin_private only under the toptxn so if this is
841 : : * not the toptxn then fetch the toptxn.
842 : : */
1259 843 [ + - ]: 4 : ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
2109 844 : 4 : TestDecodingTxnData *txndata = toptxn->output_plugin_private;
845 : 4 : bool xact_wrote_changes = txndata->xact_wrote_changes;
846 : :
1259 847 [ - + ]: 4 : if (rbtxn_is_toptxn(txn))
848 : : {
2109 akapila@postgresql.o 849 [ # # ]:UBC 0 : Assert(txn->output_plugin_private != NULL);
850 : 0 : pfree(txndata);
851 : 0 : txn->output_plugin_private = NULL;
852 : : }
853 : :
2109 akapila@postgresql.o 854 [ + - - + ]:CBC 4 : if (data->skip_empty_xacts && !xact_wrote_changes)
2176 akapila@postgresql.o 855 :UBC 0 : return;
856 : :
2221 akapila@postgresql.o 857 :CBC 4 : OutputPluginPrepareWrite(ctx, true);
858 [ - + ]: 4 : if (data->include_xids)
2221 akapila@postgresql.o 859 :UBC 0 : appendStringInfo(ctx->out, "aborting streamed (sub)transaction TXN %u", txn->xid);
860 : : else
2142 drowley@postgresql.o 861 :CBC 4 : appendStringInfoString(ctx->out, "aborting streamed (sub)transaction");
2221 akapila@postgresql.o 862 : 4 : OutputPluginWrite(ctx, true);
863 : : }
864 : :
865 : : static void
2066 866 : 1 : pg_decode_stream_prepare(LogicalDecodingContext *ctx,
867 : : ReorderBufferTXN *txn,
868 : : XLogRecPtr prepare_lsn)
869 : : {
870 : 1 : TestDecodingData *data = ctx->output_plugin_private;
871 : 1 : TestDecodingTxnData *txndata = txn->output_plugin_private;
872 : :
873 [ + - - + ]: 1 : if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
2066 akapila@postgresql.o 874 :UBC 0 : return;
875 : :
2066 akapila@postgresql.o 876 :CBC 1 : OutputPluginPrepareWrite(ctx, true);
877 : :
878 [ - + ]: 1 : if (data->include_xids)
2066 akapila@postgresql.o 879 :UBC 0 : appendStringInfo(ctx->out, "preparing streamed transaction TXN %s, txid %u",
880 : 0 : quote_literal_cstr(txn->gid), txn->xid);
881 : : else
2066 akapila@postgresql.o 882 :CBC 1 : appendStringInfo(ctx->out, "preparing streamed transaction %s",
883 : 1 : quote_literal_cstr(txn->gid));
884 : :
885 [ - + ]: 1 : if (data->include_timestamp)
2066 akapila@postgresql.o 886 :UBC 0 : appendStringInfo(ctx->out, " (at %s)",
887 : : timestamptz_to_str(txn->prepare_time));
888 : :
2066 akapila@postgresql.o 889 :CBC 1 : OutputPluginWrite(ctx, true);
890 : : }
891 : :
892 : : static void
2221 893 : 5 : pg_decode_stream_commit(LogicalDecodingContext *ctx,
894 : : ReorderBufferTXN *txn,
895 : : XLogRecPtr commit_lsn)
896 : : {
897 : 5 : TestDecodingData *data = ctx->output_plugin_private;
2109 898 : 5 : TestDecodingTxnData *txndata = txn->output_plugin_private;
899 : 5 : bool xact_wrote_changes = txndata->xact_wrote_changes;
900 : :
901 : 5 : pfree(txndata);
902 : 5 : txn->output_plugin_private = NULL;
903 : :
904 [ + - - + ]: 5 : if (data->skip_empty_xacts && !xact_wrote_changes)
2176 akapila@postgresql.o 905 :UBC 0 : return;
906 : :
2221 akapila@postgresql.o 907 :CBC 5 : OutputPluginPrepareWrite(ctx, true);
908 : :
909 [ - + ]: 5 : if (data->include_xids)
2221 akapila@postgresql.o 910 :UBC 0 : appendStringInfo(ctx->out, "committing streamed transaction TXN %u", txn->xid);
911 : : else
2142 drowley@postgresql.o 912 :CBC 5 : appendStringInfoString(ctx->out, "committing streamed transaction");
913 : :
2221 akapila@postgresql.o 914 [ - + ]: 5 : if (data->include_timestamp)
2221 akapila@postgresql.o 915 :UBC 0 : appendStringInfo(ctx->out, " (at %s)",
916 : : timestamptz_to_str(txn->commit_time));
917 : :
2221 akapila@postgresql.o 918 :CBC 5 : OutputPluginWrite(ctx, true);
919 : : }
920 : :
921 : : /*
922 : : * In streaming mode, we don't display the changes as the transaction can abort
923 : : * at a later point in time. We don't want users to see the changes until the
924 : : * transaction is committed.
925 : : */
926 : : static void
927 : 65 : pg_decode_stream_change(LogicalDecodingContext *ctx,
928 : : ReorderBufferTXN *txn,
929 : : Relation relation,
930 : : ReorderBufferChange *change)
931 : : {
932 : 65 : TestDecodingData *data = ctx->output_plugin_private;
2109 933 : 65 : TestDecodingTxnData *txndata = txn->output_plugin_private;
934 : :
935 : : /* output stream start if we haven't yet */
936 [ + - + + ]: 65 : if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
937 : : {
2176 938 : 8 : pg_output_stream_start(ctx, data, txn, false);
939 : : }
2109 940 : 65 : txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
941 : :
2221 942 : 65 : OutputPluginPrepareWrite(ctx, true);
943 [ - + ]: 65 : if (data->include_xids)
2221 akapila@postgresql.o 944 :UBC 0 : appendStringInfo(ctx->out, "streaming change for TXN %u", txn->xid);
945 : : else
2142 drowley@postgresql.o 946 :CBC 65 : appendStringInfoString(ctx->out, "streaming change for transaction");
2221 akapila@postgresql.o 947 : 65 : OutputPluginWrite(ctx, true);
948 : 65 : }
949 : :
950 : : /*
951 : : * In streaming mode, we don't display the contents for transactional messages
952 : : * as the transaction can abort at a later point in time. We don't want users to
953 : : * see the message contents until the transaction is committed.
954 : : */
955 : : static void
956 : 3 : pg_decode_stream_message(LogicalDecodingContext *ctx,
957 : : ReorderBufferTXN *txn, XLogRecPtr lsn, bool transactional,
958 : : const char *prefix, Size sz, const char *message)
959 : : {
960 : : /* Output stream start if we haven't yet for transactional messages. */
1032 961 [ + - ]: 3 : if (transactional)
962 : : {
963 : 3 : TestDecodingData *data = ctx->output_plugin_private;
964 : 3 : TestDecodingTxnData *txndata = txn->output_plugin_private;
965 : :
966 [ + - + - ]: 3 : if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
967 : : {
968 : 3 : pg_output_stream_start(ctx, data, txn, false);
969 : : }
970 : 3 : txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
971 : : }
972 : :
2221 973 : 3 : OutputPluginPrepareWrite(ctx, true);
974 : :
975 [ + - ]: 3 : if (transactional)
976 : : {
977 : 3 : appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu",
978 : : transactional, prefix, sz);
979 : : }
980 : : else
981 : : {
2221 akapila@postgresql.o 982 :UBC 0 : appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu content:",
983 : : transactional, prefix, sz);
984 : 0 : appendBinaryStringInfo(ctx->out, message, sz);
985 : : }
986 : :
2221 akapila@postgresql.o 987 :CBC 3 : OutputPluginWrite(ctx, true);
988 : 3 : }
989 : :
990 : : /*
991 : : * In streaming mode, we don't display the detailed information of Truncate.
992 : : * See pg_decode_stream_change.
993 : : */
994 : : static void
2221 akapila@postgresql.o 995 :UBC 0 : pg_decode_stream_truncate(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
996 : : int nrelations, Relation relations[],
997 : : ReorderBufferChange *change)
998 : : {
999 : 0 : TestDecodingData *data = ctx->output_plugin_private;
2109 1000 : 0 : TestDecodingTxnData *txndata = txn->output_plugin_private;
1001 : :
1002 [ # # # # ]: 0 : if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
1003 : : {
2176 1004 : 0 : pg_output_stream_start(ctx, data, txn, false);
1005 : : }
2109 1006 : 0 : txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
1007 : :
2221 1008 : 0 : OutputPluginPrepareWrite(ctx, true);
1009 [ # # ]: 0 : if (data->include_xids)
1010 : 0 : appendStringInfo(ctx->out, "streaming truncate for TXN %u", txn->xid);
1011 : : else
2142 drowley@postgresql.o 1012 : 0 : appendStringInfoString(ctx->out, "streaming truncate for transaction");
2221 akapila@postgresql.o 1013 : 0 : OutputPluginWrite(ctx, true);
1014 : 0 : }
|
