logicalproto.h File Reference

#include "access/xact.h"
#include "executor/tuptable.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
struct	LogicalRepStreamAbortData

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_STREAM_PARALLEL_VERSION_NUM   4
#define	LOGICALREP_PROTO_MAX_VERSION_NUM   LOGICALREP_PROTO_STREAM_PARALLEL_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
typedef struct LogicalRepStreamAbortData	LogicalRepStreamAbortData

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, 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_COLUMN_BINARY

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

Definition at line 99 of file logicalproto.h.

LOGICALREP_COLUMN_NULL

#define LOGICALREP_COLUMN_NULL   'n'

Definition at line 96 of file logicalproto.h.

LOGICALREP_COLUMN_TEXT

#define LOGICALREP_COLUMN_TEXT   't'

Definition at line 98 of file logicalproto.h.

LOGICALREP_COLUMN_UNCHANGED

#define LOGICALREP_COLUMN_UNCHANGED   'u'

Definition at line 97 of file logicalproto.h.

LOGICALREP_PROTO_MAX_VERSION_NUM

#define LOGICALREP_PROTO_MAX_VERSION_NUM   LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM

Definition at line 45 of file logicalproto.h.

LOGICALREP_PROTO_MIN_VERSION_NUM

#define LOGICALREP_PROTO_MIN_VERSION_NUM   1

Definition at line 40 of file logicalproto.h.

LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM

#define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM   4

Definition at line 44 of file logicalproto.h.

LOGICALREP_PROTO_STREAM_VERSION_NUM

#define LOGICALREP_PROTO_STREAM_VERSION_NUM   2

Definition at line 42 of file logicalproto.h.

LOGICALREP_PROTO_TWOPHASE_VERSION_NUM

#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM   3

Definition at line 43 of file logicalproto.h.

LOGICALREP_PROTO_VERSION_NUM

#define LOGICALREP_PROTO_VERSION_NUM   1

Definition at line 41 of file logicalproto.h.

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 101 of file logicalproto.h.

LogicalRepRollbackPreparedTxnData

typedef struct LogicalRepRollbackPreparedTxnData LogicalRepRollbackPreparedTxnData

LogicalRepStreamAbortData

typedef struct LogicalRepStreamAbortData LogicalRepStreamAbortData

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

const char * logicalrep_message_type ( LogicalRepMsgType  action )

extern

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 fb(), 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_begin()

void logicalrep_read_begin ( StringInfo  in, LogicalRepBeginData *  begin_data  )

extern

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 elog, ERROR, fb(), pq_getmsgint(), pq_getmsgint64(), and XLogRecPtrIsValid.

Referenced by apply_handle_begin().

logicalrep_read_begin_prepare()

void logicalrep_read_begin_prepare ( StringInfo  in, LogicalRepPreparedTxnData *  begin_data  )

extern

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, ERROR, fb(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), strlcpy(), and XLogRecPtrIsValid.

Referenced by apply_handle_begin_prepare().

logicalrep_read_commit()

void logicalrep_read_commit ( StringInfo  in, LogicalRepCommitData *  commit_data  )

extern

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 elog, ERROR, fb(), pq_getmsgbyte(), and pq_getmsgint64().

Referenced by apply_handle_commit().

logicalrep_read_commit_prepared()

void logicalrep_read_commit_prepared ( StringInfo  in, LogicalRepCommitPreparedTxnData *  prepare_data  )

extern

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 elog, ERROR, fb(), pq_getmsgbyte(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), strlcpy(), and XLogRecPtrIsValid.

Referenced by apply_handle_commit_prepared().

logicalrep_read_delete()

LogicalRepRelId logicalrep_read_delete ( StringInfo  in, LogicalRepTupleData *  oldtup  )

extern

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 elog, ERROR, fb(), logicalrep_read_tuple(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_delete().

logicalrep_read_insert()

LogicalRepRelId logicalrep_read_insert ( StringInfo  in, LogicalRepTupleData *  newtup  )

extern

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 elog, ERROR, fb(), logicalrep_read_tuple(), pq_getmsgbyte(), and pq_getmsgint().

Referenced by apply_handle_insert().

logicalrep_read_origin()

char * logicalrep_read_origin ( StringInfo  in, XLogRecPtr *  origin_lsn  )

extern

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  )

extern

Definition at line 228 of file proto.c.

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

References fb(), and logicalrep_read_prepare_common().

Referenced by apply_handle_prepare().

logicalrep_read_rel()

LogicalRepRelation * logicalrep_read_rel ( StringInfo  in )

extern

Definition at line 698 of file proto.c.

{
    LogicalRepRelation *rel = palloc_object(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_object, 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  )

extern

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, fb(), pq_getmsgbyte(), pq_getmsgint(), pq_getmsgint64(), pq_getmsgstring(), strlcpy(), 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  )

extern

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 Assert, fb(), InvalidXLogRecPtr, pq_getmsgint(), and pq_getmsgint64().

Referenced by apply_handle_stream_abort().

logicalrep_read_stream_commit()

TransactionId logicalrep_read_stream_commit ( StringInfo  in, LogicalRepCommitData *  commit_data  )

extern

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 elog, ERROR, fb(), 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  )

extern

Definition at line 365 of file proto.c.

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

References fb(), and logicalrep_read_prepare_common().

Referenced by apply_handle_stream_prepare().

logicalrep_read_stream_start()

TransactionId logicalrep_read_stream_start ( StringInfo  in, bool *  first_segment  )

extern

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, fb(), 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  )

extern

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

void logicalrep_read_typ ( StringInfo  in, LogicalRepTyp *  ltyp  )

extern

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 fb(), logicalrep_read_namespace(), pq_getmsgint(), pq_getmsgstring(), and pstrdup().

Referenced by apply_handle_type().

logicalrep_read_update()

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

extern

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 elog, ERROR, fb(), 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  )

extern

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(), and fb().

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

logicalrep_write_begin()

void logicalrep_write_begin ( StringInfo  out, ReorderBufferTXN *  txn  )

extern

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  )

extern

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  )

extern

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  )

extern

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, fb(), 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  )

extern

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, fb(), 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  )

extern

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 fb(), 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  )

extern

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 fb(), 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_origin()

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

extern

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  )

extern

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

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

extern

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  )

extern

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, fb(), 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  )

extern

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, fb(), 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  )

extern

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  )

extern

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  )

extern

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, fb(), 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 )

extern

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  )

extern

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

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

extern

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, fb(), 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  )

extern

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, fb(), LOGICAL_REP_MSG_UPDATE, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint32(), RelationData::rd_rel, RelationGetRelid, and TransactionIdIsValid.

Referenced by pgoutput_change().