contrib/test_decoding/test_decoding.c

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