#include "postgres.h"
#include "catalog/pg_type.h"
#include "replication/logical.h"
#include "replication/origin.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"

Include dependency graph for test_decoding.c:

Data Structures
struct	TestDecodingData

struct	TestDecodingTxnData

Functions
	PG_MODULE_MAGIC_EXT (.name="test_decoding",.version=PG_VERSION)

static void	pg_decode_startup (LogicalDecodingContext ctx, OutputPluginOptions opt, bool is_init)

static void	pg_decode_shutdown (LogicalDecodingContext *ctx)

static void	pg_decode_begin_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn)

static void	pg_output_begin (LogicalDecodingContext ctx, TestDecodingData data, ReorderBufferTXN *txn, bool last_write)

static void	pg_decode_commit_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

static void	pg_decode_change (LogicalDecodingContext ctx, ReorderBufferTXN txn, Relation relation, ReorderBufferChange *change)

static void	pg_decode_truncate (LogicalDecodingContext ctx, ReorderBufferTXN txn, int nrelations, Relation relations[], ReorderBufferChange *change)

static bool	pg_decode_filter (LogicalDecodingContext *ctx, RepOriginId origin_id)

static void	pg_decode_message (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr lsn, bool transactional, const char prefix, Size sz, const char message)

static bool	pg_decode_filter_prepare (LogicalDecodingContext ctx, TransactionId xid, const char gid)

static void	pg_decode_begin_prepare_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn)

static void	pg_decode_prepare_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr prepare_lsn)

static void	pg_decode_commit_prepared_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

static void	pg_decode_rollback_prepared_txn (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)

static void	pg_decode_stream_start (LogicalDecodingContext ctx, ReorderBufferTXN txn)

static void	pg_output_stream_start (LogicalDecodingContext ctx, TestDecodingData data, ReorderBufferTXN *txn, bool last_write)

static void	pg_decode_stream_stop (LogicalDecodingContext ctx, ReorderBufferTXN txn)

static void	pg_decode_stream_abort (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr abort_lsn)

static void	pg_decode_stream_prepare (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr prepare_lsn)

static void	pg_decode_stream_commit (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr commit_lsn)

static void	pg_decode_stream_change (LogicalDecodingContext ctx, ReorderBufferTXN txn, Relation relation, ReorderBufferChange *change)

static void	pg_decode_stream_message (LogicalDecodingContext ctx, ReorderBufferTXN txn, XLogRecPtr lsn, bool transactional, const char prefix, Size sz, const char message)

static void	pg_decode_stream_truncate (LogicalDecodingContext ctx, ReorderBufferTXN txn, int nrelations, Relation relations[], ReorderBufferChange *change)

void	_PG_init (void)

void	_PG_output_plugin_init (OutputPluginCallbacks *cb)

static void	print_literal (StringInfo s, Oid typid, char *outputstr)

static void	tuple_to_stringinfo (StringInfo s, TupleDesc tupdesc, HeapTuple tuple, bool skip_nulls)

Function Documentation

◆ _PG_init()

void _PG_init ( void )

Definition at line 124 of file test_decoding.c.

{
    /* other plugins can perform things here */
}

◆ _PG_output_plugin_init()

void _PG_output_plugin_init ( OutputPluginCallbacks * cb )

Definition at line 131 of file test_decoding.c.

{
    cb->startup_cb = pg_decode_startup;
    cb->begin_cb = pg_decode_begin_txn;
    cb->change_cb = pg_decode_change;
    cb->truncate_cb = pg_decode_truncate;
    cb->commit_cb = pg_decode_commit_txn;
    cb->filter_by_origin_cb = pg_decode_filter;
    cb->shutdown_cb = pg_decode_shutdown;
    cb->message_cb = pg_decode_message;
    cb->filter_prepare_cb = pg_decode_filter_prepare;
    cb->begin_prepare_cb = pg_decode_begin_prepare_txn;
    cb->prepare_cb = pg_decode_prepare_txn;
    cb->commit_prepared_cb = pg_decode_commit_prepared_txn;
    cb->rollback_prepared_cb = pg_decode_rollback_prepared_txn;
    cb->stream_start_cb = pg_decode_stream_start;
    cb->stream_stop_cb = pg_decode_stream_stop;
    cb->stream_abort_cb = pg_decode_stream_abort;
    cb->stream_prepare_cb = pg_decode_stream_prepare;
    cb->stream_commit_cb = pg_decode_stream_commit;
    cb->stream_change_cb = pg_decode_stream_change;
    cb->stream_message_cb = pg_decode_stream_message;
    cb->stream_truncate_cb = pg_decode_stream_truncate;
}

◆ pg_decode_begin_prepare_txn()

static void pg_decode_begin_prepare_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn
	)

static

Definition at line 350 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata =
        MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
 
    txndata->xact_wrote_changes = false;
    txn->output_plugin_private = txndata;
 
    /*
     * If asked to skip empty transactions, we'll emit BEGIN at the point
     * where the first operation is received for this transaction.
     */
    if (data->skip_empty_xacts)
        return;
 
    pg_output_begin(ctx, data, txn, true);
}

References LogicalDecodingContext::context, data, MemoryContextAllocZero(), LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, pg_output_begin(), and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_begin_txn()

static void pg_decode_begin_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn
	)

static

Definition at line 290 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata =
        MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
 
    txndata->xact_wrote_changes = false;
    txn->output_plugin_private = txndata;
 
    /*
     * If asked to skip empty transactions, we'll emit BEGIN at the point
     * where the first operation is received for this transaction.
     */
    if (data->skip_empty_xacts)
        return;
 
    pg_output_begin(ctx, data, txn, true);
}

References LogicalDecodingContext::context, data, MemoryContextAllocZero(), LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, pg_output_begin(), and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_change()

static void pg_decode_change	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		Relation	relation,
		ReorderBufferChange *	change
	)

static

Definition at line 603 of file test_decoding.c.

{
    TestDecodingData *data;
    TestDecodingTxnData *txndata;
    Form_pg_class class_form;
    TupleDesc   tupdesc;
    MemoryContext old;
 
    data = ctx->output_plugin_private;
    txndata = txn->output_plugin_private;
 
    /* output BEGIN if we haven't yet */
    if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
    {
        pg_output_begin(ctx, data, txn, false);
    }
    txndata->xact_wrote_changes = true;
 
    class_form = RelationGetForm(relation);
    tupdesc = RelationGetDescr(relation);
 
    /* Avoid leaking memory by using and resetting our own context */
    old = MemoryContextSwitchTo(data->context);
 
    OutputPluginPrepareWrite(ctx, true);
 
    appendStringInfoString(ctx->out, "table ");
    appendStringInfoString(ctx->out,
                           quote_qualified_identifier(get_namespace_name(get_rel_namespace(RelationGetRelid(relation))),
                                                      class_form->relrewrite ?
                                                      get_rel_name(class_form->relrewrite) :
                                                      NameStr(class_form->relname)));
    appendStringInfoChar(ctx->out, ':');
 
    switch (change->action)
    {
        case REORDER_BUFFER_CHANGE_INSERT:
            appendStringInfoString(ctx->out, " INSERT:");
            if (change->data.tp.newtuple == NULL)
                appendStringInfoString(ctx->out, " (no-tuple-data)");
            else
                tuple_to_stringinfo(ctx->out, tupdesc,
                                    change->data.tp.newtuple,
                                    false);
            break;
        case REORDER_BUFFER_CHANGE_UPDATE:
            appendStringInfoString(ctx->out, " UPDATE:");
            if (change->data.tp.oldtuple != NULL)
            {
                appendStringInfoString(ctx->out, " old-key:");
                tuple_to_stringinfo(ctx->out, tupdesc,
                                    change->data.tp.oldtuple,
                                    true);
                appendStringInfoString(ctx->out, " new-tuple:");
            }
 
            if (change->data.tp.newtuple == NULL)
                appendStringInfoString(ctx->out, " (no-tuple-data)");
            else
                tuple_to_stringinfo(ctx->out, tupdesc,
                                    change->data.tp.newtuple,
                                    false);
            break;
        case REORDER_BUFFER_CHANGE_DELETE:
            appendStringInfoString(ctx->out, " DELETE:");
 
            /* if there was no PK, we only know that a delete happened */
            if (change->data.tp.oldtuple == NULL)
                appendStringInfoString(ctx->out, " (no-tuple-data)");
            /* In DELETE, only the replica identity is present; display that */
            else
                tuple_to_stringinfo(ctx->out, tupdesc,
                                    change->data.tp.oldtuple,
                                    true);
            break;
        default:
            Assert(false);
    }
 
    MemoryContextSwitchTo(old);
    MemoryContextReset(data->context);
 
    OutputPluginWrite(ctx, true);
}

References ReorderBufferChange::action, appendStringInfoChar(), appendStringInfoString(), Assert(), ReorderBufferChange::data, data, get_namespace_name(), get_rel_name(), get_rel_namespace(), MemoryContextReset(), MemoryContextSwitchTo(), NameStr, ReorderBufferChange::newtuple, ReorderBufferChange::oldtuple, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_begin(), quote_qualified_identifier(), RelationGetDescr, RelationGetForm, RelationGetRelid, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferChange::tp, tuple_to_stringinfo(), and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_commit_prepared_txn()

static void pg_decode_commit_prepared_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

static

Definition at line 401 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
 
    OutputPluginPrepareWrite(ctx, true);
 
    appendStringInfo(ctx->out, "COMMIT PREPARED %s",
                     quote_literal_cstr(txn->gid));
 
    if (data->include_xids)
        appendStringInfo(ctx->out, ", txid %u", txn->xid);
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.commit_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), ReorderBufferTXN::commit_time, data, ReorderBufferTXN::gid, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), quote_literal_cstr(), timestamptz_to_str(), ReorderBufferTXN::xact_time, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_commit_txn()

static void pg_decode_commit_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

static

Definition at line 322 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
    bool        xact_wrote_changes = txndata->xact_wrote_changes;
 
    pfree(txndata);
    txn->output_plugin_private = NULL;
 
    if (data->skip_empty_xacts && !xact_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
    if (data->include_xids)
        appendStringInfo(ctx->out, "COMMIT %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "COMMIT");
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.commit_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), ReorderBufferTXN::commit_time, data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pfree(), timestamptz_to_str(), ReorderBufferTXN::xact_time, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_filter()

static bool pg_decode_filter	(	LogicalDecodingContext *	ctx,
		RepOriginId	origin_id
	)

static

Definition at line 463 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
 
    if (data->only_local && origin_id != InvalidRepOriginId)
        return true;
    return false;
}

References data, InvalidRepOriginId, and LogicalDecodingContext::output_plugin_private.

Referenced by _PG_output_plugin_init().

◆ pg_decode_filter_prepare()

static bool pg_decode_filter_prepare	(	LogicalDecodingContext *	ctx,
		TransactionId	xid,
		const char *	gid
	)

static

Definition at line 453 of file test_decoding.c.

{
    if (strstr(gid, "_nodecode") != NULL)
        return true;
 
    return false;
}

Referenced by _PG_output_plugin_init().

◆ pg_decode_message()

static void pg_decode_message	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	lsn,
		bool	transactional,
		const char *	prefix,
		Size	sz,
		const char *	message
	)

static

Definition at line 745 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata;
 
    txndata = transactional ? txn->output_plugin_private : NULL;
 
    /* output BEGIN if we haven't yet for transactional messages */
    if (transactional && data->skip_empty_xacts && !txndata->xact_wrote_changes)
        pg_output_begin(ctx, data, txn, false);
 
    if (transactional)
        txndata->xact_wrote_changes = true;
 
    OutputPluginPrepareWrite(ctx, true);
    appendStringInfo(ctx->out, "message: transactional: %d prefix: %s, sz: %zu content:",
                     transactional, prefix, sz);
    appendBinaryStringInfo(ctx->out, message, sz);
    OutputPluginWrite(ctx, true);
}

References appendBinaryStringInfo(), appendStringInfo(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_begin(), and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_prepare_txn()

static void pg_decode_prepare_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_lsn
	)

static

Definition at line 371 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    /*
     * If asked to skip empty transactions, we'll emit PREPARE at the point
     * where the first operation is received for this transaction.
     */
    if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
 
    appendStringInfo(ctx->out, "PREPARE TRANSACTION %s",
                     quote_literal_cstr(txn->gid));
 
    if (data->include_xids)
        appendStringInfo(ctx->out, ", txid %u", txn->xid);
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.prepare_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), data, ReorderBufferTXN::gid, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), ReorderBufferTXN::prepare_time, quote_literal_cstr(), timestamptz_to_str(), ReorderBufferTXN::xact_time, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_rollback_prepared_txn()

static void pg_decode_rollback_prepared_txn	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_end_lsn,
		TimestampTz	prepare_time
	)

static

Definition at line 423 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
 
    OutputPluginPrepareWrite(ctx, true);
 
    appendStringInfo(ctx->out, "ROLLBACK PREPARED %s",
                     quote_literal_cstr(txn->gid));
 
    if (data->include_xids)
        appendStringInfo(ctx->out, ", txid %u", txn->xid);
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.commit_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), ReorderBufferTXN::commit_time, data, ReorderBufferTXN::gid, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), quote_literal_cstr(), timestamptz_to_str(), ReorderBufferTXN::xact_time, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_shutdown()

static void pg_decode_shutdown ( LogicalDecodingContext * ctx )

static

Definition at line 280 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
 
    /* cleanup our own resources via memory context reset */
    MemoryContextDelete(data->context);
}

References data, MemoryContextDelete(), and LogicalDecodingContext::output_plugin_private.

Referenced by _PG_output_plugin_init().

◆ pg_decode_startup()

static void pg_decode_startup	(	LogicalDecodingContext *	ctx,
		OutputPluginOptions *	opt,
		bool	is_init
	)

static

Definition at line 159 of file test_decoding.c.

{
    ListCell   *option;
    TestDecodingData *data;
    bool        enable_streaming = false;
 
    data = palloc0(sizeof(TestDecodingData));
    data->context = AllocSetContextCreate(ctx->context,
                                          "text conversion context",
                                          ALLOCSET_DEFAULT_SIZES);
    data->include_xids = true;
    data->include_timestamp = false;
    data->skip_empty_xacts = false;
    data->only_local = false;
 
    ctx->output_plugin_private = data;
 
    opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
    opt->receive_rewrites = false;
 
    foreach(option, ctx->output_plugin_options)
    {
        DefElem    *elem = lfirst(option);
 
        Assert(elem->arg == NULL || IsA(elem->arg, String));
 
        if (strcmp(elem->defname, "include-xids") == 0)
        {
            /* if option does not provide a value, it means its value is true */
            if (elem->arg == NULL)
                data->include_xids = true;
            else if (!parse_bool(strVal(elem->arg), &data->include_xids))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else if (strcmp(elem->defname, "include-timestamp") == 0)
        {
            if (elem->arg == NULL)
                data->include_timestamp = true;
            else if (!parse_bool(strVal(elem->arg), &data->include_timestamp))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else if (strcmp(elem->defname, "force-binary") == 0)
        {
            bool        force_binary;
 
            if (elem->arg == NULL)
                continue;
            else if (!parse_bool(strVal(elem->arg), &force_binary))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
 
            if (force_binary)
                opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
        }
        else if (strcmp(elem->defname, "skip-empty-xacts") == 0)
        {
 
            if (elem->arg == NULL)
                data->skip_empty_xacts = true;
            else if (!parse_bool(strVal(elem->arg), &data->skip_empty_xacts))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else if (strcmp(elem->defname, "only-local") == 0)
        {
 
            if (elem->arg == NULL)
                data->only_local = true;
            else if (!parse_bool(strVal(elem->arg), &data->only_local))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else if (strcmp(elem->defname, "include-rewrites") == 0)
        {
 
            if (elem->arg == NULL)
                continue;
            else if (!parse_bool(strVal(elem->arg), &opt->receive_rewrites))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else if (strcmp(elem->defname, "stream-changes") == 0)
        {
            if (elem->arg == NULL)
                continue;
            else if (!parse_bool(strVal(elem->arg), &enable_streaming))
                ereport(ERROR,
                        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                         errmsg("could not parse value \"%s\" for parameter \"%s\"",
                                strVal(elem->arg), elem->defname)));
        }
        else
        {
            ereport(ERROR,
                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                     errmsg("option \"%s\" = \"%s\" is unknown",
                            elem->defname,
                            elem->arg ? strVal(elem->arg) : "(null)")));
        }
    }
 
    ctx->streaming &= enable_streaming;
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, DefElem::arg, Assert(), LogicalDecodingContext::context, data, DefElem::defname, ereport, errcode(), errmsg(), ERROR, IsA, lfirst, OUTPUT_PLUGIN_BINARY_OUTPUT, LogicalDecodingContext::output_plugin_options, LogicalDecodingContext::output_plugin_private, OUTPUT_PLUGIN_TEXTUAL_OUTPUT, OutputPluginOptions::output_type, palloc0(), parse_bool(), OutputPluginOptions::receive_rewrites, LogicalDecodingContext::streaming, and strVal.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_abort()

static void pg_decode_stream_abort	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	abort_lsn
	)

static

Definition at line 822 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
 
    /*
     * stream abort can be sent for an individual subtransaction but we
     * maintain the output_plugin_private only under the toptxn so if this is
     * not the toptxn then fetch the toptxn.
     */
    ReorderBufferTXN *toptxn = rbtxn_get_toptxn(txn);
    TestDecodingTxnData *txndata = toptxn->output_plugin_private;
    bool        xact_wrote_changes = txndata->xact_wrote_changes;
 
    if (rbtxn_is_toptxn(txn))
    {
        Assert(txn->output_plugin_private != NULL);
        pfree(txndata);
        txn->output_plugin_private = NULL;
    }
 
    if (data->skip_empty_xacts && !xact_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
    if (data->include_xids)
        appendStringInfo(ctx->out, "aborting streamed (sub)transaction TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "aborting streamed (sub)transaction");
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), Assert(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pfree(), rbtxn_get_toptxn, rbtxn_is_toptxn, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_change()

static void pg_decode_stream_change	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		Relation	relation,
		ReorderBufferChange *	change
	)

static

Definition at line 917 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    /* output stream start if we haven't yet */
    if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
    {
        pg_output_stream_start(ctx, data, txn, false);
    }
    txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
 
    OutputPluginPrepareWrite(ctx, true);
    if (data->include_xids)
        appendStringInfo(ctx->out, "streaming change for TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "streaming change for transaction");
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_stream_start(), TestDecodingTxnData::stream_wrote_changes, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_commit()

static void pg_decode_stream_commit	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

static

Definition at line 883 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
    bool        xact_wrote_changes = txndata->xact_wrote_changes;
 
    pfree(txndata);
    txn->output_plugin_private = NULL;
 
    if (data->skip_empty_xacts && !xact_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
 
    if (data->include_xids)
        appendStringInfo(ctx->out, "committing streamed transaction TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "committing streamed transaction");
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.commit_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), ReorderBufferTXN::commit_time, data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pfree(), timestamptz_to_str(), ReorderBufferTXN::xact_time, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_message()

static void pg_decode_stream_message	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	lsn,
		bool	transactional,
		const char *	prefix,
		Size	sz,
		const char *	message
	)

static

Definition at line 946 of file test_decoding.c.

{
    /* Output stream start if we haven't yet for transactional messages. */
    if (transactional)
    {
        TestDecodingData *data = ctx->output_plugin_private;
        TestDecodingTxnData *txndata = txn->output_plugin_private;
 
        if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
        {
            pg_output_stream_start(ctx, data, txn, false);
        }
        txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
    }
 
    OutputPluginPrepareWrite(ctx, true);
 
    if (transactional)
    {
        appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu",
                         transactional, prefix, sz);
    }
    else
    {
        appendStringInfo(ctx->out, "streaming message: transactional: %d prefix: %s, sz: %zu content:",
                         transactional, prefix, sz);
        appendBinaryStringInfo(ctx->out, message, sz);
    }
 
    OutputPluginWrite(ctx, true);
}

References appendBinaryStringInfo(), appendStringInfo(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_stream_start(), TestDecodingTxnData::stream_wrote_changes, and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_prepare()

static void pg_decode_stream_prepare	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_lsn
	)

static

Definition at line 856 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
 
    if (data->include_xids)
        appendStringInfo(ctx->out, "preparing streamed transaction TXN %s, txid %u",
                         quote_literal_cstr(txn->gid), txn->xid);
    else
        appendStringInfo(ctx->out, "preparing streamed transaction %s",
                         quote_literal_cstr(txn->gid));
 
    if (data->include_timestamp)
        appendStringInfo(ctx->out, " (at %s)",
                         timestamptz_to_str(txn->xact_time.prepare_time));
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), data, ReorderBufferTXN::gid, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), ReorderBufferTXN::prepare_time, quote_literal_cstr(), timestamptz_to_str(), ReorderBufferTXN::xact_time, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_start()

static void pg_decode_stream_start	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn
	)

static

Definition at line 769 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    /*
     * Allocate the txn plugin data for the first stream in the transaction.
     */
    if (txndata == NULL)
    {
        txndata =
            MemoryContextAllocZero(ctx->context, sizeof(TestDecodingTxnData));
        txndata->xact_wrote_changes = false;
        txn->output_plugin_private = txndata;
    }
 
    txndata->stream_wrote_changes = false;
    if (data->skip_empty_xacts)
        return;
    pg_output_stream_start(ctx, data, txn, true);
}

References LogicalDecodingContext::context, data, MemoryContextAllocZero(), LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, pg_output_stream_start(), TestDecodingTxnData::stream_wrote_changes, and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_stop()

static void pg_decode_stream_stop	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn
	)

static

Definition at line 804 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
        return;
 
    OutputPluginPrepareWrite(ctx, true);
    if (data->include_xids)
        appendStringInfo(ctx->out, "closing a streamed block for transaction TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "closing a streamed block for transaction");
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), TestDecodingTxnData::stream_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_stream_truncate()

static void pg_decode_stream_truncate	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		int	nrelations,
		Relation	relations[],
		ReorderBufferChange *	change
	)

static

Definition at line 985 of file test_decoding.c.

{
    TestDecodingData *data = ctx->output_plugin_private;
    TestDecodingTxnData *txndata = txn->output_plugin_private;
 
    if (data->skip_empty_xacts && !txndata->stream_wrote_changes)
    {
        pg_output_stream_start(ctx, data, txn, false);
    }
    txndata->xact_wrote_changes = txndata->stream_wrote_changes = true;
 
    OutputPluginPrepareWrite(ctx, true);
    if (data->include_xids)
        appendStringInfo(ctx->out, "streaming truncate for TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "streaming truncate for transaction");
    OutputPluginWrite(ctx, true);
}

References appendStringInfo(), appendStringInfoString(), data, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_stream_start(), TestDecodingTxnData::stream_wrote_changes, TestDecodingTxnData::xact_wrote_changes, and ReorderBufferTXN::xid.

Referenced by _PG_output_plugin_init().

◆ pg_decode_truncate()

static void pg_decode_truncate	(	LogicalDecodingContext *	ctx,
		ReorderBufferTXN *	txn,
		int	nrelations,
		Relation	relations[],
		ReorderBufferChange *	change
	)

static

Definition at line 690 of file test_decoding.c.

{
    TestDecodingData *data;
    TestDecodingTxnData *txndata;
    MemoryContext old;
    int         i;
 
    data = ctx->output_plugin_private;
    txndata = txn->output_plugin_private;
 
    /* output BEGIN if we haven't yet */
    if (data->skip_empty_xacts && !txndata->xact_wrote_changes)
    {
        pg_output_begin(ctx, data, txn, false);
    }
    txndata->xact_wrote_changes = true;
 
    /* Avoid leaking memory by using and resetting our own context */
    old = MemoryContextSwitchTo(data->context);
 
    OutputPluginPrepareWrite(ctx, true);
 
    appendStringInfoString(ctx->out, "table ");
 
    for (i = 0; i < nrelations; i++)
    {
        if (i > 0)
            appendStringInfoString(ctx->out, ", ");
 
        appendStringInfoString(ctx->out,
                               quote_qualified_identifier(get_namespace_name(relations[i]->rd_rel->relnamespace),
                                                          NameStr(relations[i]->rd_rel->relname)));
    }
 
    appendStringInfoString(ctx->out, ": TRUNCATE:");
 
    if (change->data.truncate.restart_seqs
        || change->data.truncate.cascade)
    {
        if (change->data.truncate.restart_seqs)
            appendStringInfoString(ctx->out, " restart_seqs");
        if (change->data.truncate.cascade)
            appendStringInfoString(ctx->out, " cascade");
    }
    else
        appendStringInfoString(ctx->out, " (no-flags)");
 
    MemoryContextSwitchTo(old);
    MemoryContextReset(data->context);
 
    OutputPluginWrite(ctx, true);
}

References appendStringInfoString(), ReorderBufferChange::cascade, ReorderBufferChange::data, data, get_namespace_name(), i, MemoryContextReset(), MemoryContextSwitchTo(), NameStr, LogicalDecodingContext::out, LogicalDecodingContext::output_plugin_private, ReorderBufferTXN::output_plugin_private, OutputPluginPrepareWrite(), OutputPluginWrite(), pg_output_begin(), quote_qualified_identifier(), ReorderBufferChange::restart_seqs, ReorderBufferChange::truncate, and TestDecodingTxnData::xact_wrote_changes.

Referenced by _PG_output_plugin_init().

◆ PG_MODULE_MAGIC_EXT()

PG_MODULE_MAGIC_EXT	(	.	name = `"test_decoding"`,
		.	version = `PG_VERSION`
	)

◆ pg_output_begin()

static void pg_output_begin	(	LogicalDecodingContext *	ctx,
		TestDecodingData *	data,
		ReorderBufferTXN *	txn,
		bool	last_write
	)

static

Definition at line 310 of file test_decoding.c.

{
    OutputPluginPrepareWrite(ctx, last_write);
    if (data->include_xids)
        appendStringInfo(ctx->out, "BEGIN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "BEGIN");
    OutputPluginWrite(ctx, last_write);
}

References appendStringInfo(), appendStringInfoString(), data, LogicalDecodingContext::out, OutputPluginPrepareWrite(), OutputPluginWrite(), and ReorderBufferTXN::xid.

Referenced by pg_decode_begin_prepare_txn(), pg_decode_begin_txn(), pg_decode_change(), pg_decode_message(), and pg_decode_truncate().

◆ pg_output_stream_start()

static void pg_output_stream_start	(	LogicalDecodingContext *	ctx,
		TestDecodingData *	data,
		ReorderBufferTXN *	txn,
		bool	last_write
	)

static

Definition at line 793 of file test_decoding.c.

{
    OutputPluginPrepareWrite(ctx, last_write);
    if (data->include_xids)
        appendStringInfo(ctx->out, "opening a streamed block for transaction TXN %u", txn->xid);
    else
        appendStringInfoString(ctx->out, "opening a streamed block for transaction");
    OutputPluginWrite(ctx, last_write);
}

References appendStringInfo(), appendStringInfoString(), data, LogicalDecodingContext::out, OutputPluginPrepareWrite(), OutputPluginWrite(), and ReorderBufferTXN::xid.

Referenced by pg_decode_stream_change(), pg_decode_stream_message(), pg_decode_stream_start(), and pg_decode_stream_truncate().

◆ print_literal()

static void print_literal	(	StringInfo	s,
		Oid	typid,
		char *	outputstr
	)

static

Definition at line 481 of file test_decoding.c.

{
    const char *valptr;
 
    switch (typid)
    {
        case INT2OID:
        case INT4OID:
        case INT8OID:
        case OIDOID:
        case FLOAT4OID:
        case FLOAT8OID:
        case NUMERICOID:
            /* NB: We don't care about Inf, NaN et al. */
            appendStringInfoString(s, outputstr);
            break;
 
        case BITOID:
        case VARBITOID:
            appendStringInfo(s, "B'%s'", outputstr);
            break;
 
        case BOOLOID:
            if (strcmp(outputstr, "t") == 0)
                appendStringInfoString(s, "true");
            else
                appendStringInfoString(s, "false");
            break;
 
        default:
            appendStringInfoChar(s, '\'');
            for (valptr = outputstr; *valptr; valptr++)
            {
                char        ch = *valptr;
 
                if (SQL_STR_DOUBLE(ch, false))
                    appendStringInfoChar(s, ch);
                appendStringInfoChar(s, ch);
            }
            appendStringInfoChar(s, '\'');
            break;
    }
}

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), and SQL_STR_DOUBLE.

Referenced by tuple_to_stringinfo().

◆ tuple_to_stringinfo()

static void tuple_to_stringinfo	(	StringInfo	s,
		TupleDesc	tupdesc,
		HeapTuple	tuple,
		bool	skip_nulls
	)

static

Definition at line 527 of file test_decoding.c.

{
    int         natt;
 
    /* print all columns individually */
    for (natt = 0; natt < tupdesc->natts; natt++)
    {
        Form_pg_attribute attr; /* the attribute itself */
        Oid         typid;      /* type of current attribute */
        Oid         typoutput;  /* output function */
        bool        typisvarlena;
        Datum       origval;    /* possibly toasted Datum */
        bool        isnull;     /* column is null? */
 
        attr = TupleDescAttr(tupdesc, natt);
 
        /*
         * don't print dropped columns, we can't be sure everything is
         * available for them
         */
        if (attr->attisdropped)
            continue;
 
        /*
         * Don't print system columns, oid will already have been printed if
         * present.
         */
        if (attr->attnum < 0)
            continue;
 
        typid = attr->atttypid;
 
        /* get Datum from tuple */
        origval = heap_getattr(tuple, natt + 1, tupdesc, &isnull);
 
        if (isnull && skip_nulls)
            continue;
 
        /* print attribute name */
        appendStringInfoChar(s, ' ');
        appendStringInfoString(s, quote_identifier(NameStr(attr->attname)));
 
        /* print attribute type */
        appendStringInfoChar(s, '[');
        appendStringInfoString(s, format_type_be(typid));
        appendStringInfoChar(s, ']');
 
        /* query output function */
        getTypeOutputInfo(typid,
                          &typoutput, &typisvarlena);
 
        /* print separator */
        appendStringInfoChar(s, ':');
 
        /* print data */
        if (isnull)
            appendStringInfoString(s, "null");
        else if (typisvarlena && VARATT_IS_EXTERNAL_ONDISK(origval))
            appendStringInfoString(s, "unchanged-toast-datum");
        else if (!typisvarlena)
            print_literal(s, typid,
                          OidOutputFunctionCall(typoutput, origval));
        else
        {
            Datum       val;    /* definitely detoasted Datum */
 
            val = PointerGetDatum(PG_DETOAST_DATUM(origval));
            print_literal(s, typid, OidOutputFunctionCall(typoutput, val));
        }
    }
}

References appendStringInfoChar(), appendStringInfoString(), format_type_be(), getTypeOutputInfo(), heap_getattr(), NameStr, TupleDescData::natts, OidOutputFunctionCall(), PG_DETOAST_DATUM, PointerGetDatum(), print_literal(), quote_identifier(), TupleDescAttr(), val, and VARATT_IS_EXTERNAL_ONDISK.

Referenced by pg_decode_change().

Data Structures

Functions

Function Documentation

◆ _PG_init()

◆ _PG_output_plugin_init()

◆ pg_decode_begin_prepare_txn()

◆ pg_decode_begin_txn()

◆ pg_decode_change()

◆ pg_decode_commit_prepared_txn()

◆ pg_decode_commit_txn()

◆ pg_decode_filter()

◆ pg_decode_filter_prepare()

◆ pg_decode_message()

◆ pg_decode_prepare_txn()

◆ pg_decode_rollback_prepared_txn()

◆ pg_decode_shutdown()

◆ pg_decode_startup()

◆ pg_decode_stream_abort()

◆ pg_decode_stream_change()

◆ pg_decode_stream_commit()

◆ pg_decode_stream_message()

◆ pg_decode_stream_prepare()

◆ pg_decode_stream_start()

◆ pg_decode_stream_stop()

◆ pg_decode_stream_truncate()

◆ pg_decode_truncate()

◆ PG_MODULE_MAGIC_EXT()

◆ pg_output_begin()

◆ pg_output_stream_start()

◆ print_literal()

◆ tuple_to_stringinfo()