logicalproto.h File Reference

#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

Macros
#define	LOGICALREP_PROTO_MIN_VERSION_NUM   1
#define	LOGICALREP_PROTO_VERSION_NUM   1

Typedefs
typedef struct LogicalRepTupleData	LogicalRepTupleData
typedef uint32	LogicalRepRelId
typedef struct LogicalRepRelation	LogicalRepRelation
typedef struct LogicalRepTyp	LogicalRepTyp
typedef struct LogicalRepBeginData	LogicalRepBeginData
typedef struct LogicalRepCommitData	LogicalRepCommitData

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_origin (StringInfo out, const char *origin, XLogRecPtr origin_lsn)
char *	logicalrep_read_origin (StringInfo in, XLogRecPtr *origin_lsn)
void	logicalrep_write_insert (StringInfo out, Relation rel, HeapTuple newtuple)
LogicalRepRelId	logicalrep_read_insert (StringInfo in, LogicalRepTupleData *newtup)
void	logicalrep_write_update (StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple)
LogicalRepRelId	logicalrep_read_update (StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup)
void	logicalrep_write_delete (StringInfo out, Relation rel, HeapTuple oldtuple)
LogicalRepRelId	logicalrep_read_delete (StringInfo in, LogicalRepTupleData *oldtup)
void	logicalrep_write_rel (StringInfo out, Relation rel)
LogicalRepRelation *	logicalrep_read_rel (StringInfo in)
void	logicalrep_write_typ (StringInfo out, Oid typoid)
void	logicalrep_read_typ (StringInfo out, LogicalRepTyp *ltyp)

Macro Definition Documentation

#define LOGICALREP_PROTO_MIN_VERSION_NUM   1

Definition at line 27 of file logicalproto.h.

Referenced by pgoutput_startup().

#define LOGICALREP_PROTO_VERSION_NUM   1

Definition at line 28 of file logicalproto.h.

Referenced by ApplyWorkerMain(), and pgoutput_startup().

Typedef Documentation

typedef struct LogicalRepBeginData LogicalRepBeginData

typedef struct LogicalRepCommitData LogicalRepCommitData

typedef struct LogicalRepRelation LogicalRepRelation

typedef uint32 LogicalRepRelId

Definition at line 39 of file logicalproto.h.

typedef struct LogicalRepTupleData LogicalRepTupleData

typedef struct LogicalRepTyp LogicalRepTyp

Function Documentation

void logicalrep_read_begin	(	StringInfo	in,
		LogicalRepBeginData *	begin_data
	)

Definition at line 57 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);
}

void logicalrep_read_commit	(	StringInfo	in,
		LogicalRepCommitData *	commit_data
	)

Definition at line 92 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);
}

LogicalRepRelId logicalrep_read_delete	(	StringInfo	in,
		LogicalRepTupleData *	oldtup
	)

Definition at line 277 of file proto.c.

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

LogicalRepRelId logicalrep_read_insert	(	StringInfo	in,
		LogicalRepTupleData *	newtup
	)

Definition at line 160 of file proto.c.

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

char* logicalrep_read_origin	(	StringInfo	in,
		XLogRecPtr *	origin_lsn
	)

Definition at line 126 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));
}

LogicalRepRelation* logicalrep_read_rel

(

StringInfo

in

)

Definition at line 324 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;
}

void logicalrep_read_typ	(	StringInfo	out,
		LogicalRepTyp *	ltyp
	)

Definition at line 376 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));
}

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

Definition at line 211 of file proto.c.

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

void logicalrep_write_begin	(	StringInfo	out,
		ReorderBufferTXN *	txn
	)

Definition at line 43 of file proto.c.

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

Referenced by pgoutput_begin_txn().

{
    pq_sendbyte(out, 'B');      /* BEGIN */

    /* fixed fields */
    pq_sendint64(out, txn->final_lsn);
    pq_sendint64(out, txn->commit_time);
    pq_sendint(out, txn->xid, 4);
}

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

Definition at line 72 of file proto.c.

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

Referenced by pgoutput_commit_txn().

{
    uint8       flags = 0;

    pq_sendbyte(out, 'C');      /* sending 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);
}

void logicalrep_write_delete	(	StringInfo	out,
		Relation	rel,
		HeapTuple	oldtuple
	)

Definition at line 252 of file proto.c.

References Assert, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint(), RelationData::rd_rel, RelationGetRelid, REPLICA_IDENTITY_DEFAULT, REPLICA_IDENTITY_FULL, and REPLICA_IDENTITY_INDEX.

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, 'D');      /* action DELETE */

    /* use Oid as relation identifier */
    pq_sendint(out, RelationGetRelid(rel), 4);

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

void logicalrep_write_insert	(	StringInfo	out,
		Relation	rel,
		HeapTuple	newtuple
	)

Definition at line 139 of file proto.c.

References Assert, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint(), RelationData::rd_rel, RelationGetRelid, REPLICA_IDENTITY_DEFAULT, REPLICA_IDENTITY_FULL, and REPLICA_IDENTITY_INDEX.

Referenced by pgoutput_change().

{
    pq_sendbyte(out, 'I');      /* action INSERT */

    Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);

    /* use Oid as relation identifier */
    pq_sendint(out, RelationGetRelid(rel), 4);

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

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

Definition at line 110 of file proto.c.

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

Referenced by pgoutput_begin_txn().

{
    pq_sendbyte(out, 'O');      /* ORIGIN */

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

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

void logicalrep_write_rel	(	StringInfo	out,
		Relation	rel
	)

Definition at line 299 of file proto.c.

References logicalrep_write_attrs(), logicalrep_write_namespace(), pq_sendbyte(), pq_sendint(), pq_sendstring(), RelationData::rd_rel, RelationGetNamespace, RelationGetRelationName, and RelationGetRelid.

Referenced by pgoutput_change().

{
    char       *relname;

    pq_sendbyte(out, 'R');      /* sending RELATION */

    /* use Oid as relation identifier */
    pq_sendint(out, RelationGetRelid(rel), 4);

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

void logicalrep_write_typ	(	StringInfo	out,
		Oid	typoid
	)

Definition at line 349 of file proto.c.

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

Referenced by pgoutput_change().

{
    Oid         basetypoid = getBaseType(typoid);
    HeapTuple   tup;
    Form_pg_type typtup;

    pq_sendbyte(out, 'Y');      /* sending TYPE */

    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_sendint(out, typoid, 4);

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

    ReleaseSysCache(tup);
}

void logicalrep_write_update	(	StringInfo	out,
		Relation	rel,
		HeapTuple	oldtuple,
		HeapTuple	newtuple
	)

Definition at line 182 of file proto.c.

References Assert, logicalrep_write_tuple(), pq_sendbyte(), pq_sendint(), RelationData::rd_rel, RelationGetRelid, REPLICA_IDENTITY_DEFAULT, REPLICA_IDENTITY_FULL, and REPLICA_IDENTITY_INDEX.

Referenced by pgoutput_change().

{
    pq_sendbyte(out, 'U');      /* action UPDATE */

    Assert(rel->rd_rel->relreplident == REPLICA_IDENTITY_DEFAULT ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_FULL ||
           rel->rd_rel->relreplident == REPLICA_IDENTITY_INDEX);

    /* use Oid as relation identifier */
    pq_sendint(out, RelationGetRelid(rel), 4);

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

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