logicalproto.h File Reference

#include "access/xact.h"
#include "replication/reorderbuffer.h"
#include "utils/rel.h"

Include dependency graph for logicalproto.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	LogicalRepTupleData
struct	LogicalRepRelation
struct	LogicalRepTyp
struct	LogicalRepBeginData
struct	LogicalRepCommitData
struct	LogicalRepPreparedTxnData
struct	LogicalRepCommitPreparedTxnData
struct	LogicalRepRollbackPreparedTxnData

Macros
#define	LOGICALREP_PROTO_MIN_VERSION_NUM   1
#define	LOGICALREP_PROTO_VERSION_NUM   1
#define	LOGICALREP_PROTO_STREAM_VERSION_NUM   2
#define	LOGICALREP_PROTO_TWOPHASE_VERSION_NUM   3
#define	LOGICALREP_PROTO_MAX_VERSION_NUM   LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
#define	LOGICALREP_COLUMN_NULL   'n'
#define	LOGICALREP_COLUMN_UNCHANGED   'u'
#define	LOGICALREP_COLUMN_TEXT   't'
#define	LOGICALREP_COLUMN_BINARY   'b' /* added in PG14 */

Typedefs
typedef enum LogicalRepMsgType	LogicalRepMsgType
typedef struct LogicalRepTupleData	LogicalRepTupleData
typedef uint32	LogicalRepRelId
typedef struct LogicalRepRelation	LogicalRepRelation
typedef struct LogicalRepTyp	LogicalRepTyp
typedef struct LogicalRepBeginData	LogicalRepBeginData
typedef struct LogicalRepCommitData	LogicalRepCommitData
typedef struct LogicalRepPreparedTxnData	LogicalRepPreparedTxnData
typedef struct LogicalRepCommitPreparedTxnData	LogicalRepCommitPreparedTxnData
typedef struct LogicalRepRollbackPreparedTxnData	LogicalRepRollbackPreparedTxnData

Enumerations
enum	LogicalRepMsgType {   LOGICAL_REP_MSG_BEGIN = 'B', LOGICAL_REP_MSG_COMMIT = 'C', LOGICAL_REP_MSG_ORIGIN = 'O', LOGICAL_REP_MSG_INSERT = 'I',   LOGICAL_REP_MSG_UPDATE = 'U', LOGICAL_REP_MSG_DELETE = 'D', LOGICAL_REP_MSG_TRUNCATE = 'T', LOGICAL_REP_MSG_RELATION = 'R',   LOGICAL_REP_MSG_TYPE = 'Y', LOGICAL_REP_MSG_MESSAGE = 'M', LOGICAL_REP_MSG_BEGIN_PREPARE = 'b', LOGICAL_REP_MSG_PREPARE = 'P',   LOGICAL_REP_MSG_COMMIT_PREPARED = 'K', LOGICAL_REP_MSG_ROLLBACK_PREPARED = 'r', LOGICAL_REP_MSG_STREAM_START = 'S', LOGICAL_REP_MSG_STREAM_STOP = 'E',   LOGICAL_REP_MSG_STREAM_COMMIT = 'c', LOGICAL_REP_MSG_STREAM_ABORT = 'A', LOGICAL_REP_MSG_STREAM_PREPARE = 'p' }

Functions
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)
void	logicalrep_write_prepare (StringInfo out, ReorderBufferTXN *txn, XLogRecPtr prepare_lsn)
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, HeapTuple newtuple, bool binary)
LogicalRepRelId	logicalrep_read_insert (StringInfo in, LogicalRepTupleData *newtup)
void	logicalrep_write_update (StringInfo out, TransactionId xid, Relation rel, HeapTuple oldtuple, HeapTuple newtuple, bool binary)
LogicalRepRelId	logicalrep_read_update (StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
void	logicalrep_write_delete (StringInfo out, TransactionId xid, Relation rel, HeapTuple oldtuple, bool binary)
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)
LogicalRepRelation *	logicalrep_read_rel (StringInfo in)
void	logicalrep_write_typ (StringInfo out, TransactionId xid, Oid typoid)
void	logicalrep_read_typ (StringInfo out, 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 out, LogicalRepCommitData *commit_data)
void	logicalrep_write_stream_abort (StringInfo out, TransactionId xid, TransactionId subxid)
void	logicalrep_read_stream_abort (StringInfo in, TransactionId *xid, TransactionId *subxid)
char *	logicalrep_message_type (LogicalRepMsgType action)

Macro Definition Documentation

LOGICALREP_COLUMN_BINARY

#define LOGICALREP_COLUMN_BINARY   'b' /* added in PG14 */

Definition at line 93 of file logicalproto.h.

Referenced by logicalrep_read_tuple(), logicalrep_write_tuple(), slot_modify_data(), and slot_store_data().

LOGICALREP_COLUMN_NULL

#define LOGICALREP_COLUMN_NULL   'n'

Definition at line 90 of file logicalproto.h.

Referenced by logicalrep_read_tuple(), and logicalrep_write_tuple().

LOGICALREP_COLUMN_TEXT

#define LOGICALREP_COLUMN_TEXT   't'

Definition at line 92 of file logicalproto.h.

Referenced by logicalrep_read_tuple(), logicalrep_write_tuple(), slot_modify_data(), and slot_store_data().

LOGICALREP_COLUMN_UNCHANGED

#define LOGICALREP_COLUMN_UNCHANGED   'u'

Definition at line 91 of file logicalproto.h.

Referenced by apply_handle_update(), logicalrep_read_tuple(), logicalrep_write_tuple(), and slot_modify_data().

LOGICALREP_PROTO_MAX_VERSION_NUM

#define LOGICALREP_PROTO_MAX_VERSION_NUM   LOGICALREP_PROTO_TWOPHASE_VERSION_NUM

Definition at line 39 of file logicalproto.h.

Referenced by pgoutput_startup().

LOGICALREP_PROTO_MIN_VERSION_NUM

#define LOGICALREP_PROTO_MIN_VERSION_NUM   1

Definition at line 35 of file logicalproto.h.

Referenced by pgoutput_startup().

LOGICALREP_PROTO_STREAM_VERSION_NUM

#define LOGICALREP_PROTO_STREAM_VERSION_NUM   2

Definition at line 37 of file logicalproto.h.

Referenced by ApplyWorkerMain(), and pgoutput_startup().

LOGICALREP_PROTO_TWOPHASE_VERSION_NUM

#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM   3

Definition at line 38 of file logicalproto.h.

Referenced by ApplyWorkerMain(), and pgoutput_startup().

LOGICALREP_PROTO_VERSION_NUM

#define LOGICALREP_PROTO_VERSION_NUM   1

Definition at line 36 of file logicalproto.h.

Referenced by ApplyWorkerMain().

Typedef Documentation

LogicalRepBeginData

typedef struct LogicalRepBeginData LogicalRepBeginData

LogicalRepCommitData

typedef struct LogicalRepCommitData LogicalRepCommitData

LogicalRepCommitPreparedTxnData

typedef struct LogicalRepCommitPreparedTxnData LogicalRepCommitPreparedTxnData

LogicalRepMsgType

typedef enum LogicalRepMsgType LogicalRepMsgType

LogicalRepPreparedTxnData

typedef struct LogicalRepPreparedTxnData LogicalRepPreparedTxnData

LogicalRepRelation

typedef struct LogicalRepRelation LogicalRepRelation

LogicalRepRelId

typedef uint32 LogicalRepRelId

Definition at line 95 of file logicalproto.h.

LogicalRepRollbackPreparedTxnData

typedef struct LogicalRepRollbackPreparedTxnData LogicalRepRollbackPreparedTxnData

LogicalRepTupleData

typedef struct LogicalRepTupleData LogicalRepTupleData

LogicalRepTyp

typedef struct LogicalRepTyp LogicalRepTyp

Enumeration Type Documentation

LogicalRepMsgType

enum LogicalRepMsgType

Enumerator
LOGICAL_REP_MSG_BEGIN
LOGICAL_REP_MSG_COMMIT
LOGICAL_REP_MSG_ORIGIN
LOGICAL_REP_MSG_INSERT
LOGICAL_REP_MSG_UPDATE
LOGICAL_REP_MSG_DELETE
LOGICAL_REP_MSG_TRUNCATE
LOGICAL_REP_MSG_RELATION
LOGICAL_REP_MSG_TYPE
LOGICAL_REP_MSG_MESSAGE
LOGICAL_REP_MSG_BEGIN_PREPARE
LOGICAL_REP_MSG_PREPARE
LOGICAL_REP_MSG_COMMIT_PREPARED
LOGICAL_REP_MSG_ROLLBACK_PREPARED
LOGICAL_REP_MSG_STREAM_START
LOGICAL_REP_MSG_STREAM_STOP
LOGICAL_REP_MSG_STREAM_COMMIT
LOGICAL_REP_MSG_STREAM_ABORT
LOGICAL_REP_MSG_STREAM_PREPARE

Definition at line 51 of file logicalproto.h.

{
    LOGICAL_REP_MSG_BEGIN = 'B',
    LOGICAL_REP_MSG_COMMIT = 'C',
    LOGICAL_REP_MSG_ORIGIN = 'O',
    LOGICAL_REP_MSG_INSERT = 'I',
    LOGICAL_REP_MSG_UPDATE = 'U',
    LOGICAL_REP_MSG_DELETE = 'D',
    LOGICAL_REP_MSG_TRUNCATE = 'T',
    LOGICAL_REP_MSG_RELATION = 'R',
    LOGICAL_REP_MSG_TYPE = 'Y',
    LOGICAL_REP_MSG_MESSAGE = 'M',
    LOGICAL_REP_MSG_BEGIN_PREPARE = 'b',
    LOGICAL_REP_MSG_PREPARE = 'P',
    LOGICAL_REP_MSG_COMMIT_PREPARED = 'K',
    LOGICAL_REP_MSG_ROLLBACK_PREPARED = 'r',
    LOGICAL_REP_MSG_STREAM_START = 'S',
    LOGICAL_REP_MSG_STREAM_STOP = 'E',
    LOGICAL_REP_MSG_STREAM_COMMIT = 'c',
    LOGICAL_REP_MSG_STREAM_ABORT = 'A',
    LOGICAL_REP_MSG_STREAM_PREPARE = 'p'
} LogicalRepMsgType;

Function Documentation

logicalrep_message_type()

char* logicalrep_message_type

(

LogicalRepMsgType

action

)

Definition at line 1164 of file proto.c.

References elog, ERROR, 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, and LOGICAL_REP_MSG_UPDATE.

Referenced by apply_error_callback().

{
    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";
    }

    elog(ERROR, "invalid logical replication message type \"%c\"", action);

    return NULL;                /* keep compiler quiet */
}

logicalrep_read_begin()

void logicalrep_read_begin	(	StringInfo	in,
		LogicalRepBeginData *	begin_data
	)

Definition at line 60 of file proto.c.

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

Referenced by apply_handle_begin().

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

logicalrep_read_begin_prepare()

void logicalrep_read_begin_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	begin_data
	)

Definition at line 131 of file proto.c.

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

Referenced by apply_handle_begin_prepare().

{
    /* read fields */
    begin_data->prepare_lsn = pq_getmsgint64(in);
    if (begin_data->prepare_lsn == InvalidXLogRecPtr)
        elog(ERROR, "prepare_lsn not set in begin prepare message");
    begin_data->end_lsn = pq_getmsgint64(in);
    if (begin_data->end_lsn == InvalidXLogRecPtr)
        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));
}

logicalrep_read_commit()

void logicalrep_read_commit	(	StringInfo	in,
		LogicalRepCommitData *	commit_data
	)

Definition at line 95 of file proto.c.

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

Referenced by apply_handle_commit().

{
    /* 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);
}

logicalrep_read_commit_prepared()

void logicalrep_read_commit_prepared	(	StringInfo	in,
		LogicalRepCommitPreparedTxnData *	prepare_data
	)

Definition at line 264 of file proto.c.

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

Referenced by apply_handle_commit_prepared().

{
    /* 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 (prepare_data->commit_lsn == InvalidXLogRecPtr)
        elog(ERROR, "commit_lsn is not set in commit prepared message");
    prepare_data->end_lsn = pq_getmsgint64(in);
    if (prepare_data->end_lsn == InvalidXLogRecPtr)
        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));
}

logicalrep_read_delete()

LogicalRepRelId logicalrep_read_delete	(	StringInfo	in,
		LogicalRepTupleData *	oldtup
	)

Definition at line 548 of file proto.c.

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

Referenced by apply_handle_delete().

{
    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;
}

logicalrep_read_insert()

LogicalRepRelId logicalrep_read_insert	(	StringInfo	in,
		LogicalRepTupleData *	newtup
	)

Definition at line 422 of file proto.c.

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

Referenced by apply_handle_insert().

{
    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;
}

logicalrep_read_origin()

char* logicalrep_read_origin	(	StringInfo	in,
		XLogRecPtr *	origin_lsn
	)

Definition at line 387 of file proto.c.

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

{
    /* fixed fields */
    *origin_lsn = pq_getmsgint64(in);

    /* return origin */
    return pstrdup(pq_getmsgstring(in));
}

logicalrep_read_prepare()

void logicalrep_read_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	prepare_data
	)

Definition at line 225 of file proto.c.

References logicalrep_read_prepare_common().

Referenced by apply_handle_prepare().

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

logicalrep_read_rel()

LogicalRepRelation* logicalrep_read_rel

(

StringInfo

in

)

Definition at line 683 of file proto.c.

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

Referenced by apply_handle_relation().

{
    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);

    /* Get attribute description */
    logicalrep_read_attrs(in, rel);

    return rel;
}

logicalrep_read_rollback_prepared()

void logicalrep_read_rollback_prepared	(	StringInfo	in,
		LogicalRepRollbackPreparedTxnData *	rollback_data
	)

Definition at line 322 of file proto.c.

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

Referenced by apply_handle_rollback_prepared().

{
    /* 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 (rollback_data->prepare_end_lsn == InvalidXLogRecPtr)
        elog(ERROR, "prepare_end_lsn is not set in rollback prepared message");
    rollback_data->rollback_end_lsn = pq_getmsgint64(in);
    if (rollback_data->rollback_end_lsn == InvalidXLogRecPtr)
        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));
}

logicalrep_read_stream_abort()

void logicalrep_read_stream_abort	(	StringInfo	in,
		TransactionId *	xid,
		TransactionId *	subxid
	)

Definition at line 1151 of file proto.c.

References Assert, and pq_getmsgint().

Referenced by apply_handle_stream_abort().

{
    Assert(xid && subxid);

    *xid = pq_getmsgint(in, 4);
    *subxid = pq_getmsgint(in, 4);
}

logicalrep_read_stream_commit()

TransactionId logicalrep_read_stream_commit	(	StringInfo	out,
		LogicalRepCommitData *	commit_data
	)

Definition at line 1109 of file proto.c.

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

Referenced by apply_handle_stream_commit().

{
    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;
}

logicalrep_read_stream_prepare()

void logicalrep_read_stream_prepare	(	StringInfo	in,
		LogicalRepPreparedTxnData *	prepare_data
	)

Definition at line 362 of file proto.c.

References logicalrep_read_prepare_common().

Referenced by apply_handle_stream_prepare().

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

logicalrep_read_stream_start()

TransactionId logicalrep_read_stream_start	(	StringInfo	in,
		bool *	first_segment
	)

Definition at line 1059 of file proto.c.

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

Referenced by apply_handle_stream_start().

{
    TransactionId xid;

    Assert(first_segment);

    xid = pq_getmsgint(in, 4);
    *first_segment = (pq_getmsgbyte(in) == 1);

    return xid;
}

logicalrep_read_truncate()

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

Definition at line 602 of file proto.c.

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

Referenced by apply_handle_truncate().

{
    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;
}

logicalrep_read_typ()

void logicalrep_read_typ	(	StringInfo	out,
		LogicalRepTyp *	ltyp
	)

Definition at line 739 of file proto.c.

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

Referenced by apply_handle_type().

{
    ltyp->remoteid = pq_getmsgint(in, 4);

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

logicalrep_read_update()

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

Definition at line 477 of file proto.c.

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

Referenced by apply_handle_update().

{
    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;
}

logicalrep_write_begin()

void logicalrep_write_begin	(	StringInfo	out,
		ReorderBufferTXN *	txn
	)

Definition at line 46 of file proto.c.

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

Referenced by pgoutput_begin_txn().

{
    pq_sendbyte(out, LOGICAL_REP_MSG_BEGIN);

    /* fixed fields */
    pq_sendint64(out, txn->final_lsn);
    pq_sendint64(out, txn->xact_time.commit_time);
    pq_sendint32(out, txn->xid);
}

logicalrep_write_begin_prepare()

void logicalrep_write_begin_prepare	(	StringInfo	out,
		ReorderBufferTXN *	txn
	)

Definition at line 113 of file proto.c.

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, ReorderBufferTXN::xact_time, and ReorderBufferTXN::xid.

Referenced by pgoutput_begin_prepare_txn().

{
    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->xact_time.prepare_time);
    pq_sendint32(out, txn->xid);

    /* send gid */
    pq_sendstring(out, txn->gid);
}

logicalrep_write_commit()

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

Definition at line 75 of file proto.c.

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

Referenced by pgoutput_commit_txn().

{
    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->xact_time.commit_time);
}

logicalrep_write_commit_prepared()

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

Definition at line 234 of file proto.c.

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

Referenced by pgoutput_commit_prepared_txn().

{
    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->xact_time.commit_time);
    pq_sendint32(out, txn->xid);

    /* send gid */
    pq_sendstring(out, txn->gid);
}

logicalrep_write_delete()

void logicalrep_write_delete	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		HeapTuple	oldtuple,
		bool	binary
	)

Definition at line 518 of file proto.c.

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

Referenced by pgoutput_change().

{
    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, oldtuple, binary);
}

logicalrep_write_insert()

void logicalrep_write_insert	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		HeapTuple	newtuple,
		bool	binary
	)

Definition at line 400 of file proto.c.

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

Referenced by pgoutput_change().

{
    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, newtuple, binary);
}

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 627 of file proto.c.

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().

{
    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);
}

logicalrep_write_origin()

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

Definition at line 371 of file proto.c.

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

Referenced by send_repl_origin().

{
    pq_sendbyte(out, LOGICAL_REP_MSG_ORIGIN);

    /* fixed fields */
    pq_sendint64(out, origin_lsn);

    /* origin string */
    pq_sendstring(out, origin);
}

logicalrep_write_prepare()

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

Definition at line 184 of file proto.c.

References LOGICAL_REP_MSG_PREPARE, and logicalrep_write_prepare_common().

Referenced by pgoutput_prepare_txn().

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

logicalrep_write_rel()

void logicalrep_write_rel	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel
	)

Definition at line 654 of file proto.c.

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().

{
    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);
}

logicalrep_write_rollback_prepared()

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

Definition at line 290 of file proto.c.

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

Referenced by pgoutput_rollback_prepared_txn().

{
    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->xact_time.commit_time);
    pq_sendint32(out, txn->xid);

    /* send gid */
    pq_sendstring(out, txn->gid);
}

logicalrep_write_stream_abort()

void logicalrep_write_stream_abort	(	StringInfo	out,
		TransactionId	xid,
		TransactionId	subxid
	)

Definition at line 1135 of file proto.c.

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

Referenced by pgoutput_stream_abort().

{
    pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_ABORT);

    Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(subxid));

    /* transaction ID */
    pq_sendint32(out, xid);
    pq_sendint32(out, subxid);
}

logicalrep_write_stream_commit()

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

Definition at line 1084 of file proto.c.

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

Referenced by pgoutput_stream_commit().

{
    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->xact_time.commit_time);
}

logicalrep_write_stream_prepare()

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

Definition at line 350 of file proto.c.

References LOGICAL_REP_MSG_STREAM_PREPARE, and logicalrep_write_prepare_common().

Referenced by pgoutput_stream_prepare_txn().

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

logicalrep_write_stream_start()

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

Definition at line 1041 of file proto.c.

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

Referenced by pgoutput_stream_start().

{
    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);
}

logicalrep_write_stream_stop()

void logicalrep_write_stream_stop

(

StringInfo

out

)

Definition at line 1075 of file proto.c.

References LOGICAL_REP_MSG_STREAM_STOP, and pq_sendbyte().

Referenced by pgoutput_stream_stop().

{
    pq_sendbyte(out, LOGICAL_REP_MSG_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 570 of file proto.c.

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

Referenced by pgoutput_truncate().

{
    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]);
}

logicalrep_write_typ()

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

Definition at line 708 of file proto.c.

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

Referenced by send_relation_and_attrs().

{
    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 relation identifier */
    pq_sendint32(out, typoid);

    /* send qualified type name */
    logicalrep_write_namespace(out, typtup->typnamespace);
    pq_sendstring(out, NameStr(typtup->typname));

    ReleaseSysCache(tup);
}

logicalrep_write_update()

void logicalrep_write_update	(	StringInfo	out,
		TransactionId	xid,
		Relation	rel,
		HeapTuple	oldtuple,
		HeapTuple	newtuple,
		bool	binary
	)

Definition at line 444 of file proto.c.

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

Referenced by pgoutput_change().

{
    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 (oldtuple != 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, oldtuple, binary);
    }

    pq_sendbyte(out, 'N');      /* new tuple follows */
    logicalrep_write_tuple(out, rel, newtuple, binary);
}