diff options
Diffstat (limited to 'src/backend/replication/logical/proto.c')
| -rw-r--r-- | src/backend/replication/logical/proto.c | 52 |
1 files changed, 52 insertions, 0 deletions
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index c9b0eeefd7e..3dbe85d61a2 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -649,6 +649,56 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, } /* + * Write SEQUENCE to stream + */ +void +logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid, + XLogRecPtr lsn, bool transactional, + int64 last_value, int64 log_cnt, bool is_called) +{ + uint8 flags = 0; + char *relname; + + pq_sendbyte(out, LOGICAL_REP_MSG_SEQUENCE); + + /* transaction ID (if not valid, we're not streaming) */ + if (TransactionIdIsValid(xid)) + pq_sendint32(out, xid); + + pq_sendint8(out, flags); + pq_sendint64(out, lsn); + + logicalrep_write_namespace(out, RelationGetNamespace(rel)); + relname = RelationGetRelationName(rel); + pq_sendstring(out, relname); + + pq_sendint8(out, transactional); + pq_sendint64(out, last_value); + pq_sendint64(out, log_cnt); + pq_sendint8(out, is_called); +} + +/* + * Read SEQUENCE from the stream. + */ +void +logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata) +{ + /* XXX skipping flags and lsn */ + pq_getmsgint(in, 1); + pq_getmsgint64(in); + + /* Read relation name from stream */ + seqdata->nspname = pstrdup(logicalrep_read_namespace(in)); + seqdata->seqname = pstrdup(pq_getmsgstring(in)); + + seqdata->transactional = pq_getmsgint(in, 1); + seqdata->last_value = pq_getmsgint64(in); + seqdata->log_cnt = pq_getmsgint64(in); + seqdata->is_called = pq_getmsgint(in, 1); +} + +/* * Write relation description to the output stream. */ void @@ -1203,6 +1253,8 @@ logicalrep_message_type(LogicalRepMsgType action) return "STREAM ABORT"; case LOGICAL_REP_MSG_STREAM_PREPARE: return "STREAM PREPARE"; + case LOGICAL_REP_MSG_SEQUENCE: + return "SEQUENCE"; } elog(ERROR, "invalid logical replication message type \"%c\"", action); |