proto.c File Reference

#include "postgres.h"
#include "access/sysattr.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_type.h"
#include "libpq/pqformat.h"
#include "replication/logicalproto.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"

Include dependency graph for proto.c:

Go to the source code of this file.

Macros
#define	LOGICALREP_IS_REPLICA_IDENTITY   1
#define	MESSAGE_TRANSACTIONAL   (1<<0)
#define	TRUNCATE_CASCADE   (1<<0)
#define	TRUNCATE_RESTART_SEQS   (1<<1)

Functions
static void	logicalrep_write_attrs (StringInfo out, Relation rel, Bitmapset *columns, PublishGencolsType include_gencols_type)
static void	logicalrep_write_tuple (StringInfo out, Relation rel, TupleTableSlot *slot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
static void	logicalrep_read_attrs (StringInfo in, LogicalRepRelation *rel)
static void	logicalrep_read_tuple (StringInfo in, LogicalRepTupleData *tuple)
static void	logicalrep_write_namespace (StringInfo out, Oid nspid)
static const char *	logicalrep_read_namespace (StringInfo in)
void	logicalrep_write_begin (StringInfo out, ReorderBufferTXN *txn)
void	logicalrep_read_begin (StringInfo in, LogicalRepBeginData *begin_data)
void	logicalrep_write_commit (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
void	logicalrep_read_commit (StringInfo in, LogicalRepCommitData *commit_data)
void	logicalrep_write_begin_prepare (StringInfo out, ReorderBufferTXN *txn)
void	logicalrep_read_begin_prepare (StringInfo in, LogicalRepPreparedTxnData *begin_data)
static void	logicalrep_write_prepare_common (StringInfo out, LogicalRepMsgType type, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
void	logicalrep_write_prepare (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
static void	logicalrep_read_prepare_common (StringInfo in, char *msgtype, LogicalRepPreparedTxnData *prepare_data)
void	logicalrep_read_prepare (StringInfo in, LogicalRepPreparedTxnData *prepare_data)
void	logicalrep_write_commit_prepared (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
void	logicalrep_read_commit_prepared (StringInfo in, LogicalRepCommitPreparedTxnData *prepare_data)
void	logicalrep_write_rollback_prepared (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_end_lsn, TimestampTz prepare_time)
void	logicalrep_read_rollback_prepared (StringInfo in, LogicalRepRollbackPreparedTxnData *rollback_data)
void	logicalrep_write_stream_prepare (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
void	logicalrep_read_stream_prepare (StringInfo in, LogicalRepPreparedTxnData *prepare_data)
void	logicalrep_write_origin (StringInfo out, const char *origin, XLogRecPtr origin_lsn)
char *	logicalrep_read_origin (StringInfo in, XLogRecPtr *origin_lsn)
void	logicalrep_write_insert (StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *newslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
LogicalRepRelId	logicalrep_read_insert (StringInfo in, LogicalRepTupleData *newtup)
void	logicalrep_write_update (StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, TupleTableSlot *newslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
LogicalRepRelId	logicalrep_read_update (StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
void	logicalrep_write_delete (StringInfo out, TransactionId xid, Relation rel, TupleTableSlot *oldslot, bool binary, Bitmapset *columns, PublishGencolsType include_gencols_type)
LogicalRepRelId	logicalrep_read_delete (StringInfo in, LogicalRepTupleData *oldtup)
void	logicalrep_write_truncate (StringInfo out, TransactionId xid, int nrelids, Oid relids[], bool cascade, bool restart_seqs)
List *	logicalrep_read_truncate (StringInfo in, bool *cascade, bool *restart_seqs)
void	logicalrep_write_message (StringInfo out, TransactionId xid, XLogRecPtr lsn, bool transactional, const char *prefix, Size sz, const char *message)
void	logicalrep_write_rel (StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns, PublishGencolsType include_gencols_type)
LogicalRepRelation *	logicalrep_read_rel (StringInfo in)
void	logicalrep_write_typ (StringInfo out, TransactionId xid, Oid typoid)
void	logicalrep_read_typ (StringInfo in, LogicalRepTyp *ltyp)
void	logicalrep_write_stream_start (StringInfo out, TransactionId xid, bool first_segment)
TransactionId	logicalrep_read_stream_start (StringInfo in, bool *first_segment)
void	logicalrep_write_stream_stop (StringInfo out)
void	logicalrep_write_stream_commit (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn)
TransactionId	logicalrep_read_stream_commit (StringInfo in, LogicalRepCommitData *commit_data)
void	logicalrep_write_stream_abort (StringInfo out, TransactionId xid, TransactionId subxid, XLogRecPtr abort_lsn, TimestampTz abort_time, bool write_abort_info)
void	logicalrep_read_stream_abort (StringInfo in, LogicalRepStreamAbortData *abort_data, bool read_abort_info)
const char *	logicalrep_message_type (LogicalRepMsgType action)
bool	logicalrep_should_publish_column (Form_pg_attribute att, Bitmapset *columns, PublishGencolsType include_gencols_type)

Macro Definition Documentation

LOGICALREP_IS_REPLICA_IDENTITY

#define LOGICALREP_IS_REPLICA_IDENTITY   1

Definition at line 26 of file proto.c.

MESSAGE_TRANSACTIONAL

#define MESSAGE_TRANSACTIONAL   (1<<0)

Definition at line 28 of file proto.c.

TRUNCATE_CASCADE

#define TRUNCATE_CASCADE   (1<<0)

Definition at line 29 of file proto.c.

TRUNCATE_RESTART_SEQS

#define TRUNCATE_RESTART_SEQS   (1<<1)

Definition at line 30 of file proto.c.

Function Documentation

logicalrep_message_type()

const char * logicalrep_message_type

(

LogicalRepMsgType

action

)

Definition at line 1212 of file proto.c.

{
    static char err_unknown[20];
 
    switch (action)
    {
        case LOGICAL_REP_MSG_BEGIN:
            return "BEGIN";
        case LOGICAL_REP_MSG_COMMIT:
            return "COMMIT";
        case LOGICAL_REP_MSG_ORIGIN:
            return "ORIGIN";
        case LOGICAL_REP_MSG_INSERT:
            return "INSERT";
        case LOGICAL_REP_MSG_UPDATE:
            return "UPDATE";
        case LOGICAL_REP_MSG_DELETE:
            return "DELETE";
        case LOGICAL_REP_MSG_TRUNCATE:
            return "TRUNCATE";
        case LOGICAL_REP_MSG_RELATION:
            return "RELATION";
        case LOGICAL_REP_MSG_TYPE:
            return "TYPE";
        case LOGICAL_REP_MSG_MESSAGE:
            return "MESSAGE";
        case LOGICAL_REP_MSG_BEGIN_PREPARE:
            return "BEGIN PREPARE";
        case LOGICAL_REP_MSG_PREPARE:
            return "PREPARE";
        case LOGICAL_REP_MSG_COMMIT_PREPARED:
            return "COMMIT PREPARED";
        case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
            return "ROLLBACK PREPARED";
        case LOGICAL_REP_MSG_STREAM_START:
            return "STREAM START";
        case LOGICAL_REP_MSG_STREAM_STOP:
            return "STREAM STOP";
        case LOGICAL_REP_MSG_STREAM_COMMIT:
            return "STREAM COMMIT";
        case LOGICAL_REP_MSG_STREAM_ABORT:
            return "STREAM ABORT";
        case LOGICAL_REP_MSG_STREAM_PREPARE:
            return "STREAM PREPARE";
    }
 
    /*
     * This message provides context in the error raised when applying a
     * logical message. So we can't throw an error here. Return an unknown
     * indicator value so that the original error is still reported.
     */
    snprintf(err_unknown, sizeof(err_unknown), "??? (%d)", action);
 
    return err_unknown;
}

References generate_unaccent_rules::action, LOGICAL_REP_MSG_BEGIN, LOGICAL_REP_MSG_BEGIN_PREPARE, LOGICAL_REP_MSG_COMMIT, LOGICAL_REP_MSG_COMMIT_PREPARED, LOGICAL_REP_MSG_DELETE, LOGICAL_REP_MSG_INSERT, LOGICAL_REP_MSG_MESSAGE, LOGICAL_REP_MSG_ORIGIN, LOGICAL_REP_MSG_PREPARE, LOGICAL_REP_MSG_RELATION, LOGICAL_REP_MSG_ROLLBACK_PREPARED, LOGICAL_REP_MSG_STREAM_ABORT, LOGICAL_REP_MSG_STREAM_COMMIT, LOGICAL_REP_MSG_STREAM_PREPARE, LOGICAL_REP_MSG_STREAM_START, LOGICAL_REP_MSG_STREAM_STOP, LOGICAL_REP_MSG_TRUNCATE, LOGICAL_REP_MSG_TYPE, LOGICAL_REP_MSG_UPDATE, and snprintf.

Referenced by apply_error_callback().

logicalrep_read_attrs()

static void logicalrep_read_attrs ( StringInfo  in, LogicalRepRelation *  rel  )

static

Definition at line 988 of file proto.c.

{
    int         i;
    int         natts;
    char      **attnames;
    Oid        *atttyps;
    Bitmapset  *attkeys = NULL;
 
    natts = pq_getmsgint(in, 2);
    attnames = palloc(natts * sizeof(char *));
    atttyps = palloc(natts * sizeof(Oid));
 
    /* read the attributes */
    for (i = 0; i < natts; i++)
    {
        uint8       flags;
 
        /* Check for replica identity column */
        flags = pq_getmsgbyte(in);
        if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
            attkeys = bms_add_member(attkeys, i);
 
        /* attribute name */
        attnames[i] = pstrdup(pq_getmsgstring(in));
 
        /* attribute type id */
        atttyps[i] = (Oid) pq_getmsgint(in, 4);
 
        /* we ignore attribute mode for now */
        (void) pq_getmsgint(in, 4);
    }
 
    rel->attnames = attnames;
    rel->atttyps = atttyps;
    rel->attkeys = attkeys;
    rel->natts = natts;
}

References LogicalRepRelation::attkeys, LogicalRepRelation::attnames, LogicalRepRelation::atttyps, bms_add_member(), i, LOGICALREP_IS_REPLICA_IDENTITY, LogicalRepRelation::natts, palloc(), pq_getmsgbyte(), pq_getmsgint(), pq_getmsgstring(), and pstrdup().

Referenced by logicalrep_read_rel().

logicalrep_read_begin()

void logicalrep_read_begin	(	StringInfo	in,
		LogicalRepBeginData *	begin_data
	)

Definition at line 63 of file proto.c.

{
    /* read fields */
    begin_data->final_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(begin_data->final_lsn))
        elog(ERROR, "final_lsn not set in begin message");
    begin_data->committime = pq_getmsgint64(in);
    begin_data->xid = pq_getmsgint(in, 4);
}

References LogicalRepBeginData::committime, elog, ERROR, LogicalRepBeginData::final_lsn, pq_getmsgint(), pq_getmsgint64(), LogicalRepBeginData::xid, and XLogRecPtrIsValid.

Referenced by apply_handle_begin().

logicalrep_read_begin_prepare()

void logicalrep_read_begin_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	begin_data
	)

Definition at line 134 of file proto.c.

{
    /* read fields */
    begin_data->prepare_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(begin_data->prepare_lsn))
        elog(ERROR, "prepare_lsn not set in begin prepare message");
    begin_data->end_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(begin_data->end_lsn))
        elog(ERROR, "end_lsn not set in begin prepare message");
    begin_data->prepare_time = pq_getmsgint64(in);
    begin_data->xid = pq_getmsgint(in, 4);
 
    /* read gid (copy it into a pre-allocated buffer) */
    strlcpy(begin_data->gid, pq_getmsgstring(in), sizeof(begin_data->gid));
}

References elog, LogicalRepPreparedTxnData::end_lsn, ERROR, LogicalRepPreparedTxnData::gid, pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), LogicalRepPreparedTxnData::prepare_lsn, LogicalRepPreparedTxnData::prepare_time, strlcpy(), LogicalRepPreparedTxnData::xid, and XLogRecPtrIsValid.

Referenced by apply_handle_begin_prepare().

logicalrep_read_commit()

void logicalrep_read_commit	(	StringInfo	in,
		LogicalRepCommitData *	commit_data
	)

Definition at line 98 of file proto.c.

{
    /* read flags (unused for now) */
    uint8       flags = pq_getmsgbyte(in);
 
    if (flags != 0)
        elog(ERROR, "unrecognized flags %u in commit message", flags);
 
    /* read fields */
    commit_data->commit_lsn = pq_getmsgint64(in);
    commit_data->end_lsn = pq_getmsgint64(in);
    commit_data->committime = pq_getmsgint64(in);
}

References LogicalRepCommitData::commit_lsn, LogicalRepCommitData::committime, elog, LogicalRepCommitData::end_lsn, ERROR, pq_getmsgbyte(), and pq_getmsgint64().

Referenced by apply_handle_commit().

logicalrep_read_commit_prepared()

void logicalrep_read_commit_prepared	(	StringInfo	in,
		LogicalRepCommitPreparedTxnData *	prepare_data
	)

Definition at line 267 of file proto.c.

{
    /* read flags */
    uint8       flags = pq_getmsgbyte(in);
 
    if (flags != 0)
        elog(ERROR, "unrecognized flags %u in commit prepared message", flags);
 
    /* read fields */
    prepare_data->commit_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(prepare_data->commit_lsn))
        elog(ERROR, "commit_lsn is not set in commit prepared message");
    prepare_data->end_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(prepare_data->end_lsn))
        elog(ERROR, "end_lsn is not set in commit prepared message");
    prepare_data->commit_time = pq_getmsgint64(in);
    prepare_data->xid = pq_getmsgint(in, 4);
 
    /* read gid (copy it into a pre-allocated buffer) */
    strlcpy(prepare_data->gid, pq_getmsgstring(in), sizeof(prepare_data->gid));
}

References LogicalRepCommitPreparedTxnData::commit_lsn, LogicalRepCommitPreparedTxnData::commit_time, elog, LogicalRepCommitPreparedTxnData::end_lsn, ERROR, LogicalRepCommitPreparedTxnData::gid, pq_getmsgbyte(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), strlcpy(), LogicalRepCommitPreparedTxnData::xid, and XLogRecPtrIsValid.

Referenced by apply_handle_commit_prepared().

logicalrep_read_delete()

LogicalRepRelId logicalrep_read_delete	(	StringInfo	in,
		LogicalRepTupleData *	oldtup
	)

Definition at line 561 of file proto.c.

{
    char        action;
    LogicalRepRelId relid;
 
    /* read the relation id */
    relid = pq_getmsgint(in, 4);
 
    /* read and verify action */
    action = pq_getmsgbyte(in);
    if (action != 'K' && action != 'O')
        elog(ERROR, "expected action 'O' or 'K', got %c", action);
 
    logicalrep_read_tuple(in, oldtup);
 
    return relid;
}

References generate_unaccent_rules::action, elog, ERROR, logicalrep_read_tuple(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_delete().

logicalrep_read_insert()

LogicalRepRelId logicalrep_read_insert	(	StringInfo	in,
		LogicalRepTupleData *	newtup
	)

Definition at line 428 of file proto.c.

{
    char        action;
    LogicalRepRelId relid;
 
    /* read the relation id */
    relid = pq_getmsgint(in, 4);
 
    action = pq_getmsgbyte(in);
    if (action != 'N')
        elog(ERROR, "expected new tuple but got %d",
             action);
 
    logicalrep_read_tuple(in, newtup);
 
    return relid;
}

References generate_unaccent_rules::action, elog, ERROR, logicalrep_read_tuple(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_insert().

logicalrep_read_namespace()

static const char * logicalrep_read_namespace ( StringInfo  in )

static

Definition at line 1050 of file proto.c.

{
    const char *nspname = pq_getmsgstring(in);
 
    if (nspname[0] == '\0')
        nspname = "pg_catalog";
 
    return nspname;
}

References pq_getmsgstring().

Referenced by logicalrep_read_rel(), and logicalrep_read_typ().

logicalrep_read_origin()

char * logicalrep_read_origin	(	StringInfo	in,
		XLogRecPtr *	origin_lsn
	)

Definition at line 390 of file proto.c.

{
    /* fixed fields */
    *origin_lsn = pq_getmsgint64(in);
 
    /* return origin */
    return pstrdup(pq_getmsgstring(in));
}

References pq_getmsgint64(), pq_getmsgstring(), and pstrdup().

logicalrep_read_prepare()

void logicalrep_read_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	prepare_data
	)

Definition at line 228 of file proto.c.

{
    logicalrep_read_prepare_common(in, "prepare", prepare_data);
}

References logicalrep_read_prepare_common().

Referenced by apply_handle_prepare().

logicalrep_read_prepare_common()

static void logicalrep_read_prepare_common ( StringInfo  in, char *  msgtype, LogicalRepPreparedTxnData *  prepare_data  )

static

Definition at line 199 of file proto.c.

{
    /* read flags */
    uint8       flags = pq_getmsgbyte(in);
 
    if (flags != 0)
        elog(ERROR, "unrecognized flags %u in %s message", flags, msgtype);
 
    /* read fields */
    prepare_data->prepare_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(prepare_data->prepare_lsn))
        elog(ERROR, "prepare_lsn is not set in %s message", msgtype);
    prepare_data->end_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(prepare_data->end_lsn))
        elog(ERROR, "end_lsn is not set in %s message", msgtype);
    prepare_data->prepare_time = pq_getmsgint64(in);
    prepare_data->xid = pq_getmsgint(in, 4);
    if (prepare_data->xid == InvalidTransactionId)
        elog(ERROR, "invalid two-phase transaction ID in %s message", msgtype);
 
    /* read gid (copy it into a pre-allocated buffer) */
    strlcpy(prepare_data->gid, pq_getmsgstring(in), sizeof(prepare_data->gid));
}

References elog, LogicalRepPreparedTxnData::end_lsn, ERROR, LogicalRepPreparedTxnData::gid, InvalidTransactionId, pq_getmsgbyte(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), LogicalRepPreparedTxnData::prepare_lsn, LogicalRepPreparedTxnData::prepare_time, strlcpy(), LogicalRepPreparedTxnData::xid, and XLogRecPtrIsValid.

Referenced by logicalrep_read_prepare(), and logicalrep_read_stream_prepare().

logicalrep_read_rel()

LogicalRepRelation * logicalrep_read_rel

(

StringInfo

in

)

Definition at line 698 of file proto.c.

{
    LogicalRepRelation *rel = palloc(sizeof(LogicalRepRelation));
 
    rel->remoteid = pq_getmsgint(in, 4);
 
    /* Read relation name from stream */
    rel->nspname = pstrdup(logicalrep_read_namespace(in));
    rel->relname = pstrdup(pq_getmsgstring(in));
 
    /* Read the replica identity. */
    rel->replident = pq_getmsgbyte(in);
 
    /* relkind is not sent */
    rel->relkind = 0;
 
    /* Get attribute description */
    logicalrep_read_attrs(in, rel);
 
    return rel;
}

References logicalrep_read_attrs(), logicalrep_read_namespace(), LogicalRepRelation::nspname, palloc(), pq_getmsgbyte(), pq_getmsgint(), pq_getmsgstring(), pstrdup(), LogicalRepRelation::relkind, LogicalRepRelation::relname, LogicalRepRelation::remoteid, and LogicalRepRelation::replident.

Referenced by apply_handle_relation().

logicalrep_read_rollback_prepared()

void logicalrep_read_rollback_prepared	(	StringInfo	in,
		LogicalRepRollbackPreparedTxnData *	rollback_data
	)

Definition at line 325 of file proto.c.

{
    /* read flags */
    uint8       flags = pq_getmsgbyte(in);
 
    if (flags != 0)
        elog(ERROR, "unrecognized flags %u in rollback prepared message", flags);
 
    /* read fields */
    rollback_data->prepare_end_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(rollback_data->prepare_end_lsn))
        elog(ERROR, "prepare_end_lsn is not set in rollback prepared message");
    rollback_data->rollback_end_lsn = pq_getmsgint64(in);
    if (!XLogRecPtrIsValid(rollback_data->rollback_end_lsn))
        elog(ERROR, "rollback_end_lsn is not set in rollback prepared message");
    rollback_data->prepare_time = pq_getmsgint64(in);
    rollback_data->rollback_time = pq_getmsgint64(in);
    rollback_data->xid = pq_getmsgint(in, 4);
 
    /* read gid (copy it into a pre-allocated buffer) */
    strlcpy(rollback_data->gid, pq_getmsgstring(in), sizeof(rollback_data->gid));
}

References elog, ERROR, LogicalRepRollbackPreparedTxnData::gid, pq_getmsgbyte(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), LogicalRepRollbackPreparedTxnData::prepare_end_lsn, LogicalRepRollbackPreparedTxnData::prepare_time, LogicalRepRollbackPreparedTxnData::rollback_end_lsn, LogicalRepRollbackPreparedTxnData::rollback_time, strlcpy(), LogicalRepRollbackPreparedTxnData::xid, and XLogRecPtrIsValid.

Referenced by apply_handle_rollback_prepared().

logicalrep_read_stream_abort()

void logicalrep_read_stream_abort	(	StringInfo	in,
		LogicalRepStreamAbortData *	abort_data,
		bool	read_abort_info
	)

Definition at line 1187 of file proto.c.

{
    Assert(abort_data);
 
    abort_data->xid = pq_getmsgint(in, 4);
    abort_data->subxid = pq_getmsgint(in, 4);
 
    if (read_abort_info)
    {
        abort_data->abort_lsn = pq_getmsgint64(in);
        abort_data->abort_time = pq_getmsgint64(in);
    }
    else
    {
        abort_data->abort_lsn = InvalidXLogRecPtr;
        abort_data->abort_time = 0;
    }
}

References LogicalRepStreamAbortData::abort_lsn, LogicalRepStreamAbortData::abort_time, Assert(), InvalidXLogRecPtr, pq_getmsgint(), pq_getmsgint64(), LogicalRepStreamAbortData::subxid, and LogicalRepStreamAbortData::xid.

Referenced by apply_handle_stream_abort().

logicalrep_read_stream_commit()

TransactionId logicalrep_read_stream_commit	(	StringInfo	in,
		LogicalRepCommitData *	commit_data
	)

Definition at line 1132 of file proto.c.

{
    TransactionId xid;
    uint8       flags;
 
    xid = pq_getmsgint(in, 4);
 
    /* read flags (unused for now) */
    flags = pq_getmsgbyte(in);
 
    if (flags != 0)
        elog(ERROR, "unrecognized flags %u in commit message", flags);
 
    /* read fields */
    commit_data->commit_lsn = pq_getmsgint64(in);
    commit_data->end_lsn = pq_getmsgint64(in);
    commit_data->committime = pq_getmsgint64(in);
 
    return xid;
}

References LogicalRepCommitData::commit_lsn, LogicalRepCommitData::committime, elog, LogicalRepCommitData::end_lsn, ERROR, pq_getmsgbyte(), pq_getmsgint(), and pq_getmsgint64().

Referenced by apply_handle_stream_commit().

logicalrep_read_stream_prepare()

void logicalrep_read_stream_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	prepare_data
	)

Definition at line 365 of file proto.c.

{
    logicalrep_read_prepare_common(in, "stream prepare", prepare_data);
}

References logicalrep_read_prepare_common().

Referenced by apply_handle_stream_prepare().

logicalrep_read_stream_start()

TransactionId logicalrep_read_stream_start	(	StringInfo	in,
		bool *	first_segment
	)

Definition at line 1082 of file proto.c.

{
    TransactionId xid;
 
    Assert(first_segment);
 
    xid = pq_getmsgint(in, 4);
    *first_segment = (pq_getmsgbyte(in) == 1);
 
    return xid;
}

References Assert(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_stream_start().

logicalrep_read_truncate()

List * logicalrep_read_truncate	(	StringInfo	in,
		bool *	cascade,
		bool *	restart_seqs
	)

Definition at line 615 of file proto.c.

{
    int         i;
    int         nrelids;
    List       *relids = NIL;
    uint8       flags;
 
    nrelids = pq_getmsgint(in, 4);
 
    /* read and decode truncate flags */
    flags = pq_getmsgint(in, 1);
    *cascade = (flags & TRUNCATE_CASCADE) > 0;
    *restart_seqs = (flags & TRUNCATE_RESTART_SEQS) > 0;
 
    for (i = 0; i < nrelids; i++)
        relids = lappend_oid(relids, pq_getmsgint(in, 4));
 
    return relids;
}

References i, lappend_oid(), NIL, pq_getmsgint(), TRUNCATE_CASCADE, and TRUNCATE_RESTART_SEQS.

Referenced by apply_handle_truncate().

logicalrep_read_tuple()

static void logicalrep_read_tuple ( StringInfo  in, LogicalRepTupleData *  tuple  )

static

Definition at line 864 of file proto.c.

{
    int         i;
    int         natts;
 
    /* Get number of attributes */
    natts = pq_getmsgint(in, 2);
 
    /* Allocate space for per-column values; zero out unused StringInfoDatas */
    tuple->colvalues = (StringInfoData *) palloc0(natts * sizeof(StringInfoData));
    tuple->colstatus = (char *) palloc(natts * sizeof(char));
    tuple->ncols = natts;
 
    /* Read the data */
    for (i = 0; i < natts; i++)
    {
        char       *buff;
        char        kind;
        int         len;
        StringInfo  value = &tuple->colvalues[i];
 
        kind = pq_getmsgbyte(in);
        tuple->colstatus[i] = kind;
 
        switch (kind)
        {
            case LOGICALREP_COLUMN_NULL:
                /* nothing more to do */
                break;
            case LOGICALREP_COLUMN_UNCHANGED:
                /* we don't receive the value of an unchanged column */
                break;
            case LOGICALREP_COLUMN_TEXT:
            case LOGICALREP_COLUMN_BINARY:
                len = pq_getmsgint(in, 4);  /* read length */
 
                /* and data */
                buff = palloc(len + 1);
                pq_copymsgbytes(in, buff, len);
 
                /*
                 * NUL termination is required for LOGICALREP_COLUMN_TEXT mode
                 * as input functions require that.  For
                 * LOGICALREP_COLUMN_BINARY it's not technically required, but
                 * it's harmless.
                 */
                buff[len] = '\0';
 
                initStringInfoFromString(value, buff, len);
                break;
            default:
                elog(ERROR, "unrecognized data representation type '%c'", kind);
        }
    }
}

References LogicalRepTupleData::colstatus, LogicalRepTupleData::colvalues, elog, ERROR, i, initStringInfoFromString(), len, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_NULL, LOGICALREP_COLUMN_TEXT, LOGICALREP_COLUMN_UNCHANGED, LogicalRepTupleData::ncols, palloc(), palloc0(), pq_copymsgbytes(), pq_getmsgbyte(), pq_getmsgint(), and value.

Referenced by logicalrep_read_delete(), logicalrep_read_insert(), and logicalrep_read_update().

logicalrep_read_typ()

void logicalrep_read_typ	(	StringInfo	in,
		LogicalRepTyp *	ltyp
	)

Definition at line 757 of file proto.c.

{
    ltyp->remoteid = pq_getmsgint(in, 4);
 
    /* Read type name from stream */
    ltyp->nspname = pstrdup(logicalrep_read_namespace(in));
    ltyp->typname = pstrdup(pq_getmsgstring(in));
}

References logicalrep_read_namespace(), LogicalRepTyp::nspname, pq_getmsgint(), pq_getmsgstring(), pstrdup(), LogicalRepTyp::remoteid, and LogicalRepTyp::typname.

Referenced by apply_handle_type().

logicalrep_read_update()

LogicalRepRelId logicalrep_read_update	(	StringInfo	in,
		bool *	has_oldtuple,
		LogicalRepTupleData *	oldtup,
		LogicalRepTupleData *	newtup
	)

Definition at line 487 of file proto.c.

{
    char        action;
    LogicalRepRelId relid;
 
    /* read the relation id */
    relid = pq_getmsgint(in, 4);
 
    /* read and verify action */
    action = pq_getmsgbyte(in);
    if (action != 'K' && action != 'O' && action != 'N')
        elog(ERROR, "expected action 'N', 'O' or 'K', got %c",
             action);
 
    /* check for old tuple */
    if (action == 'K' || action == 'O')
    {
        logicalrep_read_tuple(in, oldtup);
        *has_oldtuple = true;
 
        action = pq_getmsgbyte(in);
    }
    else
        *has_oldtuple = false;
 
    /* check for new  tuple */
    if (action != 'N')
        elog(ERROR, "expected action 'N', got %c",
             action);
 
    logicalrep_read_tuple(in, newtup);
 
    return relid;
}

References generate_unaccent_rules::action, elog, ERROR, logicalrep_read_tuple(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_update().

logicalrep_should_publish_column()

bool logicalrep_should_publish_column	(	Form_pg_attribute	att,
		Bitmapset *	columns,
		PublishGencolsType	include_gencols_type
	)

Definition at line 1282 of file proto.c.

{
    if (att->attisdropped)
        return false;
 
    /* If a column list is provided, publish only the cols in that list. */
    if (columns)
        return bms_is_member(att->attnum, columns);
 
    /* All non-generated columns are always published. */
    if (!att->attgenerated)
        return true;
 
    /*
     * Stored generated columns are only published when the user sets
     * publish_generated_columns as stored.
     */
    if (att->attgenerated == ATTRIBUTE_GENERATED_STORED)
        return include_gencols_type == PUBLISH_GENCOLS_STORED;
 
    return false;
}

References bms_is_member().

Referenced by logicalrep_write_attrs(), logicalrep_write_tuple(), and send_relation_and_attrs().

logicalrep_write_attrs()

static void logicalrep_write_attrs ( StringInfo  out, Relation  rel, Bitmapset *  columns, PublishGencolsType  include_gencols_type  )

static

Definition at line 924 of file proto.c.

{
    TupleDesc   desc;
    int         i;
    uint16      nliveatts = 0;
    Bitmapset  *idattrs = NULL;
    bool        replidentfull;
 
    desc = RelationGetDescr(rel);
 
    /* send number of live attributes */
    for (i = 0; i < desc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(desc, i);
 
        if (!logicalrep_should_publish_column(att, columns,
                                              include_gencols_type))
            continue;
 
        nliveatts++;
    }
    pq_sendint16(out, nliveatts);
 
    /* fetch bitmap of REPLICATION IDENTITY attributes */
    replidentfull = (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL);
    if (!replidentfull)
        idattrs = RelationGetIdentityKeyBitmap(rel);
 
    /* send the attributes */
    for (i = 0; i < desc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(desc, i);
        uint8       flags = 0;
 
        if (!logicalrep_should_publish_column(att, columns,
                                              include_gencols_type))
            continue;
 
        /* REPLICA IDENTITY FULL means all columns are sent as part of key. */
        if (replidentfull ||
            bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
                          idattrs))
            flags |= LOGICALREP_IS_REPLICA_IDENTITY;
 
        pq_sendbyte(out, flags);
 
        /* attribute name */
        pq_sendstring(out, NameStr(att->attname));
 
        /* attribute type id */
        pq_sendint32(out, (int) att->atttypid);
 
        /* attribute mode */
        pq_sendint32(out, att->atttypmod);
    }
 
    bms_free(idattrs);
}

References bms_free(), bms_is_member(), FirstLowInvalidHeapAttributeNumber, i, LOGICALREP_IS_REPLICA_IDENTITY, logicalrep_should_publish_column(), NameStr, TupleDescData::natts, pq_sendbyte(), pq_sendint16(), pq_sendint32(), pq_sendstring(), RelationData::rd_rel, RelationGetDescr, RelationGetIdentityKeyBitmap(), and TupleDescAttr().

Referenced by logicalrep_write_rel().

logicalrep_write_begin()

void logicalrep_write_begin	(	StringInfo	out,
		ReorderBufferTXN *	txn
	)

Definition at line 49 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_BEGIN);
 
    /* fixed fields */
    pq_sendint64(out, txn->final_lsn);
    pq_sendint64(out, txn->commit_time);
    pq_sendint32(out, txn->xid);
}

References ReorderBufferTXN::commit_time, ReorderBufferTXN::final_lsn, LOGICAL_REP_MSG_BEGIN, pq_sendbyte(), pq_sendint32(), pq_sendint64(), and ReorderBufferTXN::xid.

Referenced by pgoutput_send_begin().

logicalrep_write_begin_prepare()

void logicalrep_write_begin_prepare	(	StringInfo	out,
		ReorderBufferTXN *	txn
	)

Definition at line 116 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_BEGIN_PREPARE);
 
    /* fixed fields */
    pq_sendint64(out, txn->final_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, txn->prepare_time);
    pq_sendint32(out, txn->xid);
 
    /* send gid */
    pq_sendstring(out, txn->gid);
}

References ReorderBufferTXN::end_lsn, ReorderBufferTXN::final_lsn, ReorderBufferTXN::gid, LOGICAL_REP_MSG_BEGIN_PREPARE, pq_sendbyte(), pq_sendint32(), pq_sendint64(), pq_sendstring(), ReorderBufferTXN::prepare_time, and ReorderBufferTXN::xid.

Referenced by pgoutput_begin_prepare_txn().

logicalrep_write_commit()

void logicalrep_write_commit	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

Definition at line 78 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_COMMIT);
 
    /* send the flags field (unused for now) */
    pq_sendbyte(out, flags);
 
    /* send fields */
    pq_sendint64(out, commit_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, txn->commit_time);
}

References ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, LOGICAL_REP_MSG_COMMIT, pq_sendbyte(), and pq_sendint64().

Referenced by pgoutput_commit_txn().

logicalrep_write_commit_prepared()

void logicalrep_write_commit_prepared	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

Definition at line 237 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_COMMIT_PREPARED);
 
    /*
     * This should only ever happen for two-phase commit transactions, in
     * which case we expect to have a valid GID.
     */
    Assert(txn->gid != NULL);
 
    /* send the flags field */
    pq_sendbyte(out, flags);
 
    /* send fields */
    pq_sendint64(out, commit_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, txn->commit_time);
    pq_sendint32(out, txn->xid);
 
    /* send gid */
    pq_sendstring(out, txn->gid);
}

References Assert(), ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, ReorderBufferTXN::gid, LOGICAL_REP_MSG_COMMIT_PREPARED, pq_sendbyte(), pq_sendint32(), pq_sendint64(), pq_sendstring(), and ReorderBufferTXN::xid.

Referenced by pgoutput_commit_prepared_txn().

logicalrep_write_delete()

void logicalrep_write_delete	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		TupleTableSlot *	oldslot,
		bool	binary,
		Bitmapset *	columns,
		PublishGencolsType	include_gencols_type
	)

Definition at line 528 of file proto.c.

{
    Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);
 
    pq_sendbyte(out, LOGICAL_REP_MSG_DELETE);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    /* use Oid as relation identifier */
    pq_sendint32(out, RelationGetRelid(rel));
 
    if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
        pq_sendbyte(out, 'O');  /* old tuple follows */
    else
        pq_sendbyte(out, 'K');  /* old key follows */
 
    logicalrep_write_tuple(out, rel, oldslot, binary, columns,
                           include_gencols_type);
}

References Assert(), LOGICAL_REP_MSG_DELETE, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint32(), RelationData::rd_rel, RelationGetRelid, and TransactionIdIsValid.

Referenced by pgoutput_change().

logicalrep_write_insert()

void logicalrep_write_insert	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		TupleTableSlot *	newslot,
		bool	binary,
		Bitmapset *	columns,
		PublishGencolsType	include_gencols_type
	)

Definition at line 403 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_INSERT);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    /* use Oid as relation identifier */
    pq_sendint32(out, RelationGetRelid(rel));
 
    pq_sendbyte(out, 'N');      /* new tuple follows */
    logicalrep_write_tuple(out, rel, newslot, binary, columns,
                           include_gencols_type);
}

References LOGICAL_REP_MSG_INSERT, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint32(), RelationGetRelid, and TransactionIdIsValid.

Referenced by pgoutput_change().

logicalrep_write_message()

void logicalrep_write_message	(	StringInfo	out,
		TransactionId	xid,
		XLogRecPtr	lsn,
		bool	transactional,
		const char *	prefix,
		Size	sz,
		const char *	message
	)

Definition at line 640 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_MESSAGE);
 
    /* encode and send message flags */
    if (transactional)
        flags |= MESSAGE_TRANSACTIONAL;
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    pq_sendint8(out, flags);
    pq_sendint64(out, lsn);
    pq_sendstring(out, prefix);
    pq_sendint32(out, sz);
    pq_sendbytes(out, message, sz);
}

References LOGICAL_REP_MSG_MESSAGE, MESSAGE_TRANSACTIONAL, pq_sendbyte(), pq_sendbytes(), pq_sendint32(), pq_sendint64(), pq_sendint8(), pq_sendstring(), and TransactionIdIsValid.

Referenced by pgoutput_message().

logicalrep_write_namespace()

static void logicalrep_write_namespace ( StringInfo  out, Oid  nspid  )

static

Definition at line 1030 of file proto.c.

{
    if (nspid == PG_CATALOG_NAMESPACE)
        pq_sendbyte(out, '\0');
    else
    {
        char       *nspname = get_namespace_name(nspid);
 
        if (nspname == NULL)
            elog(ERROR, "cache lookup failed for namespace %u",
                 nspid);
 
        pq_sendstring(out, nspname);
    }
}

References elog, ERROR, get_namespace_name(), nspid, pq_sendbyte(), and pq_sendstring().

Referenced by logicalrep_write_rel(), and logicalrep_write_typ().

logicalrep_write_origin()

void logicalrep_write_origin	(	StringInfo	out,
		const char *	origin,
		XLogRecPtr	origin_lsn
	)

Definition at line 374 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_ORIGIN);
 
    /* fixed fields */
    pq_sendint64(out, origin_lsn);
 
    /* origin string */
    pq_sendstring(out, origin);
}

References LOGICAL_REP_MSG_ORIGIN, pq_sendbyte(), pq_sendint64(), and pq_sendstring().

Referenced by send_repl_origin().

logicalrep_write_prepare()

void logicalrep_write_prepare	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_lsn
	)

Definition at line 187 of file proto.c.

{
    logicalrep_write_prepare_common(out, LOGICAL_REP_MSG_PREPARE,
                                    txn, prepare_lsn);
}

References LOGICAL_REP_MSG_PREPARE, and logicalrep_write_prepare_common().

Referenced by pgoutput_prepare_txn().

logicalrep_write_prepare_common()

static void logicalrep_write_prepare_common ( StringInfo  out, LogicalRepMsgType  type, ReorderBufferTXN *  txn, XLogRecPtr  prepare_lsn  )

static

Definition at line 155 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, type);
 
    /*
     * This should only ever happen for two-phase commit transactions, in
     * which case we expect to have a valid GID.
     */
    Assert(txn->gid != NULL);
    Assert(rbtxn_is_prepared(txn));
    Assert(TransactionIdIsValid(txn->xid));
 
    /* send the flags field */
    pq_sendbyte(out, flags);
 
    /* send fields */
    pq_sendint64(out, prepare_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, txn->prepare_time);
    pq_sendint32(out, txn->xid);
 
    /* send gid */
    pq_sendstring(out, txn->gid);
}

References Assert(), ReorderBufferTXN::end_lsn, ReorderBufferTXN::gid, pq_sendbyte(), pq_sendint32(), pq_sendint64(), pq_sendstring(), ReorderBufferTXN::prepare_time, rbtxn_is_prepared, TransactionIdIsValid, type, and ReorderBufferTXN::xid.

Referenced by logicalrep_write_prepare(), and logicalrep_write_stream_prepare().

logicalrep_write_rel()

void logicalrep_write_rel	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		Bitmapset *	columns,
		PublishGencolsType	include_gencols_type
	)

Definition at line 667 of file proto.c.

{
    char       *relname;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_RELATION);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    /* use Oid as relation identifier */
    pq_sendint32(out, RelationGetRelid(rel));
 
    /* send qualified relation name */
    logicalrep_write_namespace(out, RelationGetNamespace(rel));
    relname = RelationGetRelationName(rel);
    pq_sendstring(out, relname);
 
    /* send replica identity */
    pq_sendbyte(out, rel->rd_rel->relreplident);
 
    /* send the attribute info */
    logicalrep_write_attrs(out, rel, columns, include_gencols_type);
}

References LOGICAL_REP_MSG_RELATION, logicalrep_write_attrs(), logicalrep_write_namespace(), pq_sendbyte(), pq_sendint32(), pq_sendstring(), RelationData::rd_rel, RelationGetNamespace, RelationGetRelationName, RelationGetRelid, relname, and TransactionIdIsValid.

Referenced by send_relation_and_attrs().

logicalrep_write_rollback_prepared()

void logicalrep_write_rollback_prepared	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_end_lsn,
		TimestampTz	prepare_time
	)

Definition at line 293 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_ROLLBACK_PREPARED);
 
    /*
     * This should only ever happen for two-phase commit transactions, in
     * which case we expect to have a valid GID.
     */
    Assert(txn->gid != NULL);
 
    /* send the flags field */
    pq_sendbyte(out, flags);
 
    /* send fields */
    pq_sendint64(out, prepare_end_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, prepare_time);
    pq_sendint64(out, txn->commit_time);
    pq_sendint32(out, txn->xid);
 
    /* send gid */
    pq_sendstring(out, txn->gid);
}

References Assert(), ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, ReorderBufferTXN::gid, LOGICAL_REP_MSG_ROLLBACK_PREPARED, pq_sendbyte(), pq_sendint32(), pq_sendint64(), pq_sendstring(), and ReorderBufferTXN::xid.

Referenced by pgoutput_rollback_prepared_txn().

logicalrep_write_stream_abort()

void logicalrep_write_stream_abort	(	StringInfo	out,
		TransactionId	xid,
		TransactionId	subxid,
		XLogRecPtr	abort_lsn,
		TimestampTz	abort_time,
		bool	write_abort_info
	)

Definition at line 1161 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_ABORT);
 
    Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(subxid));
 
    /* transaction ID */
    pq_sendint32(out, xid);
    pq_sendint32(out, subxid);
 
    if (write_abort_info)
    {
        pq_sendint64(out, abort_lsn);
        pq_sendint64(out, abort_time);
    }
}

References Assert(), LOGICAL_REP_MSG_STREAM_ABORT, pq_sendbyte(), pq_sendint32(), pq_sendint64(), and TransactionIdIsValid.

Referenced by pgoutput_stream_abort().

logicalrep_write_stream_commit()

void logicalrep_write_stream_commit	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	commit_lsn
	)

Definition at line 1107 of file proto.c.

{
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_COMMIT);
 
    Assert(TransactionIdIsValid(txn->xid));
 
    /* transaction ID */
    pq_sendint32(out, txn->xid);
 
    /* send the flags field (unused for now) */
    pq_sendbyte(out, flags);
 
    /* send fields */
    pq_sendint64(out, commit_lsn);
    pq_sendint64(out, txn->end_lsn);
    pq_sendint64(out, txn->commit_time);
}

References Assert(), ReorderBufferTXN::commit_time, ReorderBufferTXN::end_lsn, LOGICAL_REP_MSG_STREAM_COMMIT, pq_sendbyte(), pq_sendint32(), pq_sendint64(), TransactionIdIsValid, and ReorderBufferTXN::xid.

Referenced by pgoutput_stream_commit().

logicalrep_write_stream_prepare()

void logicalrep_write_stream_prepare	(	StringInfo	out,
		ReorderBufferTXN *	txn,
		XLogRecPtr	prepare_lsn
	)

Definition at line 353 of file proto.c.

{
    logicalrep_write_prepare_common(out, LOGICAL_REP_MSG_STREAM_PREPARE,
                                    txn, prepare_lsn);
}

References LOGICAL_REP_MSG_STREAM_PREPARE, and logicalrep_write_prepare_common().

Referenced by pgoutput_stream_prepare_txn().

logicalrep_write_stream_start()

void logicalrep_write_stream_start	(	StringInfo	out,
		TransactionId	xid,
		bool	first_segment
	)

Definition at line 1064 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_START);
 
    Assert(TransactionIdIsValid(xid));
 
    /* transaction ID (we're starting to stream, so must be valid) */
    pq_sendint32(out, xid);
 
    /* 1 if this is the first streaming segment for this xid */
    pq_sendbyte(out, first_segment ? 1 : 0);
}

References Assert(), LOGICAL_REP_MSG_STREAM_START, pq_sendbyte(), pq_sendint32(), and TransactionIdIsValid.

Referenced by pgoutput_stream_start().

logicalrep_write_stream_stop()

void logicalrep_write_stream_stop

(

StringInfo

out

)

Definition at line 1098 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_STOP);
}

References LOGICAL_REP_MSG_STREAM_STOP, and pq_sendbyte().

Referenced by pgoutput_stream_stop().

logicalrep_write_truncate()

void logicalrep_write_truncate	(	StringInfo	out,
		TransactionId	xid,
		int	nrelids,
		Oid	relids[],
		bool	cascade,
		bool	restart_seqs
	)

Definition at line 583 of file proto.c.

{
    int         i;
    uint8       flags = 0;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_TRUNCATE);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    pq_sendint32(out, nrelids);
 
    /* encode and send truncate flags */
    if (cascade)
        flags |= TRUNCATE_CASCADE;
    if (restart_seqs)
        flags |= TRUNCATE_RESTART_SEQS;
    pq_sendint8(out, flags);
 
    for (i = 0; i < nrelids; i++)
        pq_sendint32(out, relids[i]);
}

References i, LOGICAL_REP_MSG_TRUNCATE, pq_sendbyte(), pq_sendint32(), pq_sendint8(), TransactionIdIsValid, TRUNCATE_CASCADE, and TRUNCATE_RESTART_SEQS.

Referenced by pgoutput_truncate().

logicalrep_write_tuple()

static void logicalrep_write_tuple ( StringInfo  out, Relation  rel, TupleTableSlot *  slot, bool  binary, Bitmapset *  columns, PublishGencolsType  include_gencols_type  )

static

Definition at line 770 of file proto.c.

{
    TupleDesc   desc;
    Datum      *values;
    bool       *isnull;
    int         i;
    uint16      nliveatts = 0;
 
    desc = RelationGetDescr(rel);
 
    for (i = 0; i < desc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(desc, i);
 
        if (!logicalrep_should_publish_column(att, columns,
                                              include_gencols_type))
            continue;
 
        nliveatts++;
    }
    pq_sendint16(out, nliveatts);
 
    slot_getallattrs(slot);
    values = slot->tts_values;
    isnull = slot->tts_isnull;
 
    /* Write the values */
    for (i = 0; i < desc->natts; i++)
    {
        HeapTuple   typtup;
        Form_pg_type typclass;
        Form_pg_attribute att = TupleDescAttr(desc, i);
 
        if (!logicalrep_should_publish_column(att, columns,
                                              include_gencols_type))
            continue;
 
        if (isnull[i])
        {
            pq_sendbyte(out, LOGICALREP_COLUMN_NULL);
            continue;
        }
 
        if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(DatumGetPointer(values[i])))
        {
            /*
             * Unchanged toasted datum.  (Note that we don't promise to detect
             * unchanged data in general; this is just a cheap check to avoid
             * sending large values unnecessarily.)
             */
            pq_sendbyte(out, LOGICALREP_COLUMN_UNCHANGED);
            continue;
        }
 
        typtup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(att->atttypid));
        if (!HeapTupleIsValid(typtup))
            elog(ERROR, "cache lookup failed for type %u", att->atttypid);
        typclass = (Form_pg_type) GETSTRUCT(typtup);
 
        /*
         * Send in binary if requested and type has suitable send function.
         */
        if (binary && OidIsValid(typclass->typsend))
        {
            bytea      *outputbytes;
            int         len;
 
            pq_sendbyte(out, LOGICALREP_COLUMN_BINARY);
            outputbytes = OidSendFunctionCall(typclass->typsend, values[i]);
            len = VARSIZE(outputbytes) - VARHDRSZ;
            pq_sendint(out, len, 4);    /* length */
            pq_sendbytes(out, VARDATA(outputbytes), len);   /* data */
            pfree(outputbytes);
        }
        else
        {
            char       *outputstr;
 
            pq_sendbyte(out, LOGICALREP_COLUMN_TEXT);
            outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]);
            pq_sendcountedtext(out, outputstr, strlen(outputstr));
            pfree(outputstr);
        }
 
        ReleaseSysCache(typtup);
    }
}

References DatumGetPointer(), elog, ERROR, GETSTRUCT(), HeapTupleIsValid, i, len, LOGICALREP_COLUMN_BINARY, LOGICALREP_COLUMN_NULL, LOGICALREP_COLUMN_TEXT, LOGICALREP_COLUMN_UNCHANGED, logicalrep_should_publish_column(), TupleDescData::natts, ObjectIdGetDatum(), OidIsValid, OidOutputFunctionCall(), OidSendFunctionCall(), pfree(), pq_sendbyte(), pq_sendbytes(), pq_sendcountedtext(), pq_sendint(), pq_sendint16(), RelationGetDescr, ReleaseSysCache(), SearchSysCache1(), slot_getallattrs(), TupleTableSlot::tts_isnull, TupleTableSlot::tts_values, TupleDescAttr(), values, VARATT_IS_EXTERNAL_ONDISK(), VARDATA(), VARHDRSZ, and VARSIZE().

Referenced by logicalrep_write_delete(), logicalrep_write_insert(), and logicalrep_write_update().

logicalrep_write_typ()

void logicalrep_write_typ	(	StringInfo	out,
		TransactionId	xid,
		Oid	typoid
	)

Definition at line 726 of file proto.c.

{
    Oid         basetypoid = getBaseType(typoid);
    HeapTuple   tup;
    Form_pg_type typtup;
 
    pq_sendbyte(out, LOGICAL_REP_MSG_TYPE);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(basetypoid));
    if (!HeapTupleIsValid(tup))
        elog(ERROR, "cache lookup failed for type %u", basetypoid);
    typtup = (Form_pg_type) GETSTRUCT(tup);
 
    /* use Oid as type identifier */
    pq_sendint32(out, typoid);
 
    /* send qualified type name */
    logicalrep_write_namespace(out, typtup->typnamespace);
    pq_sendstring(out, NameStr(typtup->typname));
 
    ReleaseSysCache(tup);
}

References elog, ERROR, getBaseType(), GETSTRUCT(), HeapTupleIsValid, LOGICAL_REP_MSG_TYPE, logicalrep_write_namespace(), NameStr, ObjectIdGetDatum(), pq_sendbyte(), pq_sendint32(), pq_sendstring(), ReleaseSysCache(), SearchSysCache1(), and TransactionIdIsValid.

Referenced by send_relation_and_attrs().

logicalrep_write_update()

void logicalrep_write_update	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		TupleTableSlot *	oldslot,
		TupleTableSlot *	newslot,
		bool	binary,
		Bitmapset *	columns,
		PublishGencolsType	include_gencols_type
	)

Definition at line 450 of file proto.c.

{
    pq_sendbyte(out, LOGICAL_REP_MSG_UPDATE);
 
    Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);
 
    /* transaction ID (if not valid, we're not streaming) */
    if (TransactionIdIsValid(xid))
        pq_sendint32(out, xid);
 
    /* use Oid as relation identifier */
    pq_sendint32(out, RelationGetRelid(rel));
 
    if (oldslot != NULL)
    {
        if (rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
            pq_sendbyte(out, 'O');  /* old tuple follows */
        else
            pq_sendbyte(out, 'K');  /* old key follows */
        logicalrep_write_tuple(out, rel, oldslot, binary, columns,
                               include_gencols_type);
    }
 
    pq_sendbyte(out, 'N');      /* new tuple follows */
    logicalrep_write_tuple(out, rel, newslot, binary, columns,
                           include_gencols_type);
}

References Assert(), LOGICAL_REP_MSG_UPDATE, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint32(), RelationData::rd_rel, RelationGetRelid, and TransactionIdIsValid.

Referenced by pgoutput_change().