>From d04cd9cca5e6e1425bf69dfe9417e684021f8960 Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Fri, 22 Feb 2013 17:43:27 +0100 Subject: [PATCH] Introduce "replication_identifiers" to keep track of remote nodes. Replication identifiers can be used to track & lookup remote nodes identified via (sysid, tlid, remote_dbid, local_dbid, name) and map that tuple to a local uint16. Keyed by that replication identifier the progress of replication from that system is tracked in a crashsafe manner. Support for tracking that via output plugins is added as well. Needs a catversion bump. --- src/backend/access/rmgrdesc/xactdesc.c | 17 +- src/backend/access/transam/xact.c | 68 +- src/backend/access/transam/xlog.c | 8 + src/backend/catalog/Makefile | 2 +- src/backend/catalog/catalog.c | 8 +- src/backend/catalog/system_views.sql | 7 + src/backend/replication/logical/Makefile | 3 +- src/backend/replication/logical/decode.c | 6 +- src/backend/replication/logical/logical.c | 7 +- src/backend/replication/logical/reorderbuffer.c | 3 +- .../replication/logical/replication_identifier.c | 898 +++++++++++++++++++++ src/backend/storage/ipc/ipci.c | 3 + src/backend/utils/cache/syscache.c | 23 + src/backend/utils/misc/guc.c | 1 - src/bin/initdb/initdb.c | 1 + src/bin/pg_resetxlog/pg_resetxlog.c | 2 + src/include/access/xact.h | 9 +- src/include/access/xlog.h | 1 + src/include/access/xlogdefs.h | 6 + src/include/catalog/indexing.h | 6 + src/include/catalog/pg_proc.h | 21 + src/include/catalog/pg_replication_identifier.h | 75 ++ src/include/replication/logical.h | 6 + src/include/replication/reorderbuffer.h | 10 +- src/include/replication/replication_identifier.h | 45 ++ src/include/utils/syscache.h | 2 + src/test/regress/expected/rules.out | 5 + src/test/regress/expected/sanity_check.out | 1 + 28 files changed, 1222 insertions(+), 22 deletions(-) create mode 100644 src/backend/replication/logical/replication_identifier.c create mode 100644 src/include/catalog/pg_replication_identifier.h create mode 100644 src/include/replication/replication_identifier.h diff --git a/src/backend/access/rmgrdesc/xactdesc.c b/src/backend/access/rmgrdesc/xactdesc.c index 2caf5a0..7e7e40e 100644 --- a/src/backend/access/rmgrdesc/xactdesc.c +++ b/src/backend/access/rmgrdesc/xactdesc.c @@ -26,9 +26,12 @@ xact_desc_commit(StringInfo buf, xl_xact_commit *xlrec) { int i; TransactionId *subxacts; + SharedInvalidationMessage *msgs; subxacts = (TransactionId *) &xlrec->xnodes[xlrec->nrels]; + msgs = (SharedInvalidationMessage *) &subxacts[xlrec->nsubxacts]; + appendStringInfoString(buf, timestamptz_to_str(xlrec->xact_time)); if (xlrec->nrels > 0) @@ -50,9 +53,6 @@ xact_desc_commit(StringInfo buf, xl_xact_commit *xlrec) } if (xlrec->nmsgs > 0) { - SharedInvalidationMessage *msgs; - - msgs = (SharedInvalidationMessage *) &subxacts[xlrec->nsubxacts]; if (XactCompletionRelcacheInitFileInval(xlrec->xinfo)) appendStringInfo(buf, "; relcache init file inval dbid %u tsid %u", @@ -81,6 +81,17 @@ xact_desc_commit(StringInfo buf, xl_xact_commit *xlrec) appendStringInfo(buf, " unknown id %d", msg->id); } } + if (xlrec->xinfo & XACT_CONTAINS_ORIGIN) + { + xl_xact_origin *origin = (xl_xact_origin *) &(msgs[xlrec->nmsgs]); + + appendStringInfo(buf, " origin %u, lsn %X/%X, at %s", + origin->origin_node_id, + (uint32)(origin->origin_lsn >> 32), + (uint32)origin->origin_lsn, + timestamptz_to_str(origin->origin_timestamp)); + } + } static void diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 9140b3a..02e163a 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -38,8 +38,10 @@ #include "libpq/pqsignal.h" #include "miscadmin.h" #include "pgstat.h" +#include "replication/logical.h" #include "replication/walsender.h" #include "replication/syncrep.h" +#include "replication/replication_identifier.h" #include "storage/fd.h" #include "storage/lmgr.h" #include "storage/predicate.h" @@ -1076,11 +1078,13 @@ RecordTransactionCommit(void) /* * Do we need the long commit record? If not, use the compact format. */ - if (nrels > 0 || nmsgs > 0 || RelcacheInitFileInval || forceSyncCommit) + if (nrels > 0 || nmsgs > 0 || RelcacheInitFileInval || + forceSyncCommit || replication_origin_id != InvalidRepNodeId) { - XLogRecData rdata[4]; + XLogRecData rdata[5]; int lastrdata = 0; xl_xact_commit xlrec; + xl_xact_origin origin; /* * Set flags required for recovery processing of commits. @@ -1128,6 +1132,21 @@ RecordTransactionCommit(void) rdata[3].buffer = InvalidBuffer; lastrdata = 3; } + /* dump transaction origin information */ + if (replication_origin_id != InvalidRepNodeId) + { + Assert(replication_origin_lsn != InvalidXLogRecPtr); + xlrec.xinfo |= XACT_CONTAINS_ORIGIN; + origin.origin_node_id = replication_origin_id; + origin.origin_lsn = replication_origin_lsn; + origin.origin_timestamp = replication_origin_timestamp; + + rdata[lastrdata].next = &(rdata[4]); + rdata[4].data = (char *) &origin; + rdata[4].len = sizeof(xl_xact_origin); + rdata[4].buffer = InvalidBuffer; + lastrdata = 4; + } rdata[lastrdata].next = NULL; (void) XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT, rdata); @@ -1158,13 +1177,21 @@ RecordTransactionCommit(void) } } + /* record plain commit ts if not replaying remote actions */ + if (replication_origin_id == InvalidRepNodeId) + replication_origin_timestamp = xactStopTimestamp; + else + AdvanceCachedReplicationIdentifier(replication_origin_lsn, XactLastRecEnd); + /* * We don't need to log the commit timestamp separately since the commit * record logged above has all the necessary action to set the timestamp * again. */ TransactionTreeSetCommitTimestamp(xid, nchildren, children, - xactStopTimestamp, 0, false); + replication_origin_timestamp, + replication_origin_id, + false); /* * Check if we want to commit asynchronously. We can allow the XLOG flush @@ -1248,7 +1275,6 @@ RecordTransactionCommit(void) /* Reset XactLastRecEnd until the next transaction writes something */ XactLastRecEnd = 0; - cleanup: /* Clean up local data */ if (rels) @@ -4675,10 +4701,12 @@ xact_redo_commit_internal(TransactionId xid, XLogRecPtr lsn, SharedInvalidationMessage *inval_msgs, int nmsgs, RelFileNode *xnodes, int nrels, Oid dbId, Oid tsId, - uint32 xinfo) + uint32 xinfo, + xl_xact_origin *origin) { TransactionId max_xid; int i; + RepNodeId origin_node_id = InvalidRepNodeId; max_xid = TransactionIdLatest(xid, nsubxacts, sub_xids); @@ -4698,9 +4726,26 @@ xact_redo_commit_internal(TransactionId xid, XLogRecPtr lsn, LWLockRelease(XidGenLock); } + Assert(!!(xinfo & XACT_CONTAINS_ORIGIN) == (origin != NULL)); + + if (xinfo & XACT_CONTAINS_ORIGIN) + { + origin_node_id = origin->origin_node_id; + commit_time = origin->origin_timestamp; + } + /* Set the transaction commit time */ TransactionTreeSetCommitTimestamp(xid, nsubxacts, sub_xids, - commit_time, 0, false); + commit_time, + origin_node_id, false); + + if (xinfo & XACT_CONTAINS_ORIGIN) + { + /* recover apply progress */ + AdvanceReplicationIdentifier(origin->origin_node_id, + origin->origin_lsn, + lsn); + } if (standbyState == STANDBY_DISABLED) { @@ -4815,19 +4860,23 @@ xact_redo_commit(xl_xact_commit *xlrec, { TransactionId *subxacts; SharedInvalidationMessage *inval_msgs; - + xl_xact_origin *origin = NULL; /* subxid array follows relfilenodes */ subxacts = (TransactionId *) &(xlrec->xnodes[xlrec->nrels]); /* invalidation messages array follows subxids */ inval_msgs = (SharedInvalidationMessage *) &(subxacts[xlrec->nsubxacts]); + if (xlrec->xinfo & XACT_CONTAINS_ORIGIN) + origin = (xl_xact_origin *) &(inval_msgs[xlrec->nmsgs]); + xact_redo_commit_internal(xid, lsn, xlrec->xact_time, subxacts, xlrec->nsubxacts, inval_msgs, xlrec->nmsgs, xlrec->xnodes, xlrec->nrels, xlrec->dbId, xlrec->tsId, - xlrec->xinfo); + xlrec->xinfo, + origin); } /* @@ -4843,7 +4892,8 @@ xact_redo_commit_compact(xl_xact_commit_compact *xlrec, NULL, 0, /* relfilenodes */ InvalidOid, /* dbId */ InvalidOid, /* tsId */ - 0); /* xinfo */ + 0, /* xinfo */ + NULL /* origin */); } /* diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 7a7d306..235fe75 100755 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -42,6 +42,7 @@ #include "postmaster/bgwriter.h" #include "postmaster/startup.h" #include "replication/logical.h" +#include "replication/replication_identifier.h" #include "replication/snapbuild.h" #include "replication/walreceiver.h" #include "replication/walsender.h" @@ -1055,6 +1056,7 @@ begin:; rechdr->xl_len = len; /* doesn't include backup blocks */ rechdr->xl_info = info; rechdr->xl_rmid = rmid; + rechdr->xl_origin_id = replication_origin_id; rechdr->xl_prev = InvalidXLogRecPtr; COMP_CRC32(rdata_crc, ((char *) rechdr), offsetof(XLogRecord, xl_prev)); @@ -6361,6 +6363,11 @@ StartupXLOG(void) StartupMultiXact(); /* + * Recover knowledge about replay progress of known replication partners. + */ + StartupReplicationIdentifier(checkPoint.redo); + + /* * Initialize unlogged LSN. On a clean shutdown, it's restored from the * control file. On recovery, all unlogged relations are blown away, so * the unlogged LSN counter can be reset too. @@ -8449,6 +8456,7 @@ CheckPointGuts(XLogRecPtr checkPointRedo, int flags) CheckPointSnapBuild(); CheckpointLogicalRewriteHeap(); CheckPointBuffers(flags); /* performs all required fsyncs */ + CheckPointReplicationIdentifier(checkPointRedo); /* We deliberately delay 2PC checkpointing as long as possible */ CheckPointTwoPhase(checkPointRedo); } diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile index a974bd5..3a20672 100644 --- a/src/backend/catalog/Makefile +++ b/src/backend/catalog/Makefile @@ -39,7 +39,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\ pg_ts_config.h pg_ts_config_map.h pg_ts_dict.h \ pg_ts_parser.h pg_ts_template.h pg_extension.h \ pg_foreign_data_wrapper.h pg_foreign_server.h pg_user_mapping.h \ - pg_foreign_table.h \ + pg_foreign_table.h pg_replication_identifier.h \ pg_default_acl.h pg_seclabel.h pg_shseclabel.h pg_collation.h pg_range.h \ toasting.h indexing.h \ ) diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c index 7719798..d2ea0d5 100644 --- a/src/backend/catalog/catalog.c +++ b/src/backend/catalog/catalog.c @@ -32,6 +32,7 @@ #include "catalog/pg_namespace.h" #include "catalog/pg_pltemplate.h" #include "catalog/pg_db_role_setting.h" +#include "catalog/pg_replication_identifier.h" #include "catalog/pg_shdepend.h" #include "catalog/pg_shdescription.h" #include "catalog/pg_shseclabel.h" @@ -275,7 +276,8 @@ IsSharedRelation(Oid relationId) relationId == SharedDependRelationId || relationId == SharedSecLabelRelationId || relationId == TableSpaceRelationId || - relationId == DbRoleSettingRelationId) + relationId == DbRoleSettingRelationId || + relationId == ReplicationIdentifierRelationId) return true; /* These are their indexes (see indexing.h) */ if (relationId == AuthIdRolnameIndexId || @@ -291,7 +293,9 @@ IsSharedRelation(Oid relationId) relationId == SharedSecLabelObjectIndexId || relationId == TablespaceOidIndexId || relationId == TablespaceNameIndexId || - relationId == DbRoleSettingDatidRolidIndexId) + relationId == DbRoleSettingDatidRolidIndexId || + relationId == ReplicationLocalIdentIndex || + relationId == ReplicationExternalIdentIndex) return true; /* These are their toast tables and toast indexes (see toasting.h) */ if (relationId == PgShdescriptionToastTable || diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 49aae10..ed13748 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -720,6 +720,13 @@ CREATE VIEW pg_user_mappings AS REVOKE ALL on pg_user_mapping FROM public; + +CREATE VIEW pg_replication_identifier_progress AS + SELECT * + FROM pg_get_replication_identifier_progress(); + +REVOKE ALL on pg_user_mapping FROM public; + -- -- We have a few function definitions in here, too. -- At some point there might be enough to justify breaking them out into diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile index 6fae278..f24dbbe 100644 --- a/src/backend/replication/logical/Makefile +++ b/src/backend/replication/logical/Makefile @@ -14,7 +14,8 @@ include $(top_builddir)/src/Makefile.global override CPPFLAGS := -I$(srcdir) $(CPPFLAGS) -OBJS = decode.o logical.o logicalfuncs.o reorderbuffer.o snapbuild.o +OBJS = decode.o logical.o logicalfuncs.o reorderbuffer.o replication_identifier.o \ + snapbuild.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 655feca..50ccef9 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -496,7 +496,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, /* replay actions of all transaction + subtransactions in order */ ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr, - commit_time); + commit_time, buf->record.xl_origin_id); } /* @@ -540,6 +540,7 @@ DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) change = ReorderBufferGetChange(ctx->reorder); change->action = REORDER_BUFFER_CHANGE_INSERT; + change->origin_id = r->xl_origin_id; memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); if (xlrec->flags & XLOG_HEAP_CONTAINS_NEW_TUPLE) @@ -580,6 +581,7 @@ DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) change = ReorderBufferGetChange(ctx->reorder); change->action = REORDER_BUFFER_CHANGE_UPDATE; + change->origin_id = r->xl_origin_id; memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); data = (char *) &xlhdr->header; @@ -634,6 +636,7 @@ DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) change = ReorderBufferGetChange(ctx->reorder); change->action = REORDER_BUFFER_CHANGE_DELETE; + change->origin_id = r->xl_origin_id; memcpy(&change->tp.relnode, &xlrec->target.node, sizeof(RelFileNode)); @@ -689,6 +692,7 @@ DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) change = ReorderBufferGetChange(ctx->reorder); change->action = REORDER_BUFFER_CHANGE_INSERT; + change->origin_id = r->xl_origin_id; memcpy(&change->tp.relnode, &xlrec->node, sizeof(RelFileNode)); /* diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 10041cd..3c1d65a 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -71,6 +71,10 @@ LogicalDecodingSlot *MyLogicalDecodingSlot = NULL; /* user settable parameters */ int max_logical_slots = 0; /* the maximum number of logical slots */ +RepNodeId replication_origin_id = InvalidRepNodeId; /* assumed identity */ +XLogRecPtr replication_origin_lsn; +TimestampTz replication_origin_timestamp; + static void LogicalSlotKill(int code, Datum arg); /* persistency functions */ @@ -809,7 +813,8 @@ StartupLogicalDecoding(XLogRecPtr checkPointRedo) /* one of our own directories */ if (strcmp(logical_de->d_name, "snapshots") == 0 || - strcmp(logical_de->d_name, "mappings") == 0) + strcmp(logical_de->d_name, "mappings") == 0 || + strcmp(logical_de->d_name, "checkpoints") == 0) continue; /* we crashed while a slot was being setup or deleted, clean up */ diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 05a5020..979fa23 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -1261,7 +1261,7 @@ ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap) void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, - TimestampTz commit_time) + TimestampTz commit_time, RepNodeId origin) { ReorderBufferTXN *txn; ReorderBufferIterTXNState *iterstate = NULL; @@ -1282,6 +1282,7 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, txn->final_lsn = commit_lsn; txn->end_lsn = end_lsn; txn->commit_time = commit_time; + txn->origin_id = origin; /* serialize the last bunch of changes if we need start earlier anyway */ if (txn->nentries_mem != txn->nentries) diff --git a/src/backend/replication/logical/replication_identifier.c b/src/backend/replication/logical/replication_identifier.c new file mode 100644 index 0000000..67684b2 --- /dev/null +++ b/src/backend/replication/logical/replication_identifier.c @@ -0,0 +1,898 @@ +/*------------------------------------------------------------------------- + * + * replication_identifier.c + * Logical Replication Node Identifier and replication progress persistency + * support. + * + * Copyright (c) 2013, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/logical/replication_identifier.c + * + */ + +#include "postgres.h" + +#include + +#include "funcapi.h" +#include "miscadmin.h" + +#include "access/genam.h" +#include "access/heapam.h" +#include "access/htup_details.h" +#include "access/xact.h" + +#include "catalog/indexing.h" + +#include "nodes/execnodes.h" + +#include "replication/replication_identifier.h" +#include "replication/logical.h" + +#include "storage/fd.h" +#include "storage/copydir.h" + +#include "utils/builtins.h" +#include "utils/fmgroids.h" +#include "utils/rel.h" +#include "utils/syscache.h" +#include "utils/tqual.h" + +/* + * Replay progress of a single remote node. + */ +typedef struct ReplicationState +{ + /* + * Local identifier for the remote node. + */ + RepNodeId local_identifier; + + /* + * Location of the latest commit from the remote side. + */ + XLogRecPtr remote_lsn; + + /* + * Remember the local lsn of the commit record so we can XLogFlush() to it + * during a checkpoint so we know the commit record actually is safe on + * disk. + */ + XLogRecPtr local_lsn; +} ReplicationState; + +/* + * Base address into a shared memory array of replication states of size + * max_logical_slots. + * XXX: Should we use a separate variable to size this than max_logical_slots? + */ +static ReplicationState *ReplicationStates; + +/* + * Backend-local, cached element from ReplicationStates for use in a backend + * replaying remote commits, so we don't have to search ReplicationStates for + * the backends current RepNodeId. + */ +static ReplicationState *local_replication_state = NULL; + +/* Magic for on disk files. */ +#define REPLICATION_STATE_MAGIC (uint32)0x1257DADE + +/* XXX: move to c.h? */ +#ifndef UINT16_MAX +#define UINT16_MAX ((1<<16) - 1) +#else +#if UINT16_MAX != ((1<<16) - 1) +#error "uh, wrong UINT16_MAX?" +#endif +#endif + +/* + * Check for a persistent repication identifier identified by remotesysid, + * remotetli, remotedb, riname, rilocaldb. + * + * Returns InvalidOid if the node isn't known yet. + */ +RepNodeId +GetReplicationIdentifier(char *riname, bool missing_ok) +{ + Form_pg_replication_identifier ident; + Oid riident = InvalidOid; + HeapTuple tuple; + Datum riname_d; + + riname_d = CStringGetTextDatum(riname); + + tuple = SearchSysCache1(REPLIDREMOTE, riname_d); + if (HeapTupleIsValid(tuple)) + { + ident = (Form_pg_replication_identifier)GETSTRUCT(tuple); + riident = ident->riident; + ReleaseSysCache(tuple); + } + else if (!missing_ok) + elog(ERROR, "cache lookup failed for replication identifier named %s", + riname); + + return riident; +} + +/* + * Create a persistent replication identifier. + * + * Needs to be called in a transaction. + */ +RepNodeId +CreateReplicationIdentifier(char *riname) +{ + Oid riident; + HeapTuple tuple = NULL; + Relation rel; + Datum riname_d; + SnapshotData SnapshotDirty; + SysScanDesc scan; + ScanKeyData key; + + riname_d = CStringGetTextDatum(riname); + + Assert(IsTransactionState()); + + /* + * We need the numeric replication identifiers to be 16bit wide, so we + * cannot rely on the normal oid allocation. So we simply scan + * pg_replication_identifier for the first unused id. That's not + * particularly efficient, but this should be an fairly infrequent + * operation - we can easily spend a bit more code when it turns out it + * should be faster. + * + * We handle concurrency by taking an exclusive lock (allowing reads!) + * over the table for the duration of the search. Because we use a "dirty + * snapshot" we can read rows that other in-progress sessions have + * written, even though they would be invisible with normal snapshots. Due + * to the exclusive lock there's no danger that new rows can appear while + * we're checking. + */ + InitDirtySnapshot(SnapshotDirty); + + rel = heap_open(ReplicationIdentifierRelationId, ExclusiveLock); + + for (riident = InvalidOid + 1; riident <= UINT16_MAX; riident++) + { + bool nulls[Natts_pg_replication_identifier]; + Datum values[Natts_pg_replication_identifier]; + bool collides; + CHECK_FOR_INTERRUPTS(); + + ScanKeyInit(&key, + Anum_pg_replication_riident, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(riident)); + + scan = systable_beginscan(rel, ReplicationLocalIdentIndex, + true /* indexOK */, + &SnapshotDirty, + 1, &key); + + collides = HeapTupleIsValid(systable_getnext(scan)); + + systable_endscan(scan); + + if (!collides) + { + /* + * Ok, found an unused riident, insert the new row and do a CCI, + * so our callers can look it up if they want to. + */ + memset(&nulls, 0, sizeof(nulls)); + + values[Anum_pg_replication_riident -1] = ObjectIdGetDatum(riident); + values[Anum_pg_replication_riname - 1] = riname_d; + + tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls); + simple_heap_insert(rel, tuple); + CatalogUpdateIndexes(rel, tuple); + CommandCounterIncrement(); + break; + } + } + + /* now release lock again, */ + heap_close(rel, ExclusiveLock); + + if (tuple == NULL) + ereport(ERROR, + (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED), + errmsg("no free replication id could be found"))); + + heap_freetuple(tuple); + return riident; +} + + +/* + * Lookup pg_replication_identifier tuple via its riident. + * + * The result needs to be ReleaseSysCache'ed and is an invalid HeapTuple if + * the lookup failed. + */ +HeapTuple +GetReplicationInfoByIdentifier(RepNodeId riident, bool missing_ok) +{ + HeapTuple tuple; + + Assert(OidIsValid((Oid) riident)); + Assert(riident < UINT16_MAX); + tuple = SearchSysCache1(REPLIDIDENT, + ObjectIdGetDatum((Oid) riident)); + + if (!HeapTupleIsValid(tuple) && !missing_ok) + elog(ERROR, "cache lookup failed for replication identifier id: %u", + riident); + + return tuple; +} + +static void +CheckReplicationIdentifierPrerequisites(bool check_slots) +{ + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("only superusers can query or manipulate replication identifiers"))); + + if (check_slots && max_logical_slots == 0) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot query or manipulate replication identifiers when max_logical_slots = 0"))); + +} + +Datum +pg_replication_identifier_get(PG_FUNCTION_ARGS) +{ + char *name; + RepNodeId riident; + + CheckReplicationIdentifierPrerequisites(false); + + name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0))); + riident = GetReplicationIdentifier(name, true); + + pfree(name); + + if (OidIsValid(riident)) + PG_RETURN_OID(riident); + PG_RETURN_NULL(); +} + + +Datum +pg_replication_identifier_create(PG_FUNCTION_ARGS) +{ + char *name; + RepNodeId riident; + + CheckReplicationIdentifierPrerequisites(false); + + name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0))); + riident = CreateReplicationIdentifier(name); + + pfree(name); + + PG_RETURN_OID(riident); +} + +Datum +pg_replication_identifier_setup_replaying_from(PG_FUNCTION_ARGS) +{ + char *name; + RepNodeId origin; + + CheckReplicationIdentifierPrerequisites(true); + + name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0))); + origin = GetReplicationIdentifier(name, false); + SetupCachedReplicationIdentifier(origin); + + replication_origin_id = origin; + + pfree(name); + + PG_RETURN_VOID(); +} + +Datum +pg_replication_identifier_reset_replaying_from(PG_FUNCTION_ARGS) +{ + CheckReplicationIdentifierPrerequisites(true); + + TeardownCachedReplicationIdentifier(); + + replication_origin_id = InvalidRepNodeId; + + PG_RETURN_VOID(); +} + + +Datum +pg_replication_identifier_setup_tx_origin(PG_FUNCTION_ARGS) +{ + text *location = PG_GETARG_TEXT_P(0); + char *locationstr; + uint32 hi, + lo; + + CheckReplicationIdentifierPrerequisites(true); + + locationstr = text_to_cstring(location); + + if (sscanf(locationstr, "%X/%X", &hi, &lo) != 2) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("could not parse transaction log location \"%s\"", + locationstr))); + + if (local_replication_state == NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("need to setup the origin id first"))); + + replication_origin_lsn = ((uint64) hi) << 32 | lo; + replication_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1); + + pfree(locationstr); + + PG_RETURN_VOID(); +} + +Datum +pg_get_replication_identifier_progress(PG_FUNCTION_ARGS) +{ + ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; + TupleDesc tupdesc; + Tuplestorestate *tupstore; + MemoryContext per_query_ctx; + MemoryContext oldcontext; + int i; +#define REPLICATION_IDENTIFIER_PROGRESS_COLS 4 + + CheckReplicationIdentifierPrerequisites(true); + + if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("set-valued function called in context that cannot accept a set"))); + if (!(rsinfo->allowedModes & SFRM_Materialize)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("materialize mode required, but it is not allowed in this context"))); + if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) + elog(ERROR, "return type must be a row type"); + + if (tupdesc->natts != REPLICATION_IDENTIFIER_PROGRESS_COLS) + elog(ERROR, "wrong function definition"); + + per_query_ctx = rsinfo->econtext->ecxt_per_query_memory; + oldcontext = MemoryContextSwitchTo(per_query_ctx); + + tupstore = tuplestore_begin_heap(true, false, work_mem); + rsinfo->returnMode = SFRM_Materialize; + rsinfo->setResult = tupstore; + rsinfo->setDesc = tupdesc; + + MemoryContextSwitchTo(oldcontext); + + /* + * Iterate through all possible ReplicationStates, display if they are + * filled. Note that we do not take any locks, so slightly corrupted/out + * of date values are a possibility. + */ + for (i = 0; i < max_logical_slots; i++) + { + ReplicationState *state; + Datum values[REPLICATION_IDENTIFIER_PROGRESS_COLS]; + bool nulls[REPLICATION_IDENTIFIER_PROGRESS_COLS]; + char location[MAXFNAMELEN]; + HeapTuple ri; + Form_pg_replication_identifier ric; + + state = &ReplicationStates[i]; + + /* unused slot, nothing to display */ + if (state->local_identifier == InvalidRepNodeId) + continue; + + memset(values, 0, sizeof(values)); + memset(nulls, 0, sizeof(nulls)); + + values[ 0] = ObjectIdGetDatum(state->local_identifier); + + ri = GetReplicationInfoByIdentifier(state->local_identifier, true); + + /* + * We're not preventing the identifier to be dropped concurrently, so + * silently accept that it might be gone. + */ + if (!OidIsValid(ri)) + continue; + + ric = (Form_pg_replication_identifier) GETSTRUCT(ri); + values[ 1] = PointerGetDatum(&ric->riname); + + snprintf(location, sizeof(location), "%X/%X", + (uint32) (state->remote_lsn >> 32), (uint32) state->remote_lsn); + values[ 2] = CStringGetTextDatum(location); + snprintf(location, sizeof(location), "%X/%X", + (uint32) (state->local_lsn >> 32), (uint32) state->local_lsn); + values[ 3] = CStringGetTextDatum(location); + + tuplestore_putvalues(tupstore, tupdesc, values, nulls); + + /* free the strings we just allocated */ + pfree(DatumGetPointer(values[ 2])); + pfree(DatumGetPointer(values[ 3])); + ReleaseSysCache(ri); + } + + tuplestore_donestoring(tupstore); + +#undef REPLICATION_IDENTIFIER_PROGRESS_COLS + + return (Datum) 0; +} + +Size +ReplicationIdentifierShmemSize(void) +{ + Size size = 0; + + /* + * FIXME: max_logical_slots is the wrong thing to use here, here we keep + * the replay state of *remote* transactions. + */ + if (max_logical_slots == 0) + return size; + + size = add_size(size, + mul_size(max_logical_slots, sizeof(ReplicationState))); + return size; +} + +void +ReplicationIdentifierShmemInit(void) +{ + bool found; + + if (max_logical_slots == 0) + return; + + ReplicationStates = (ReplicationState *) + ShmemInitStruct("ReplicationIdentifierState", + ReplicationIdentifierShmemSize(), + &found); + + if (!found) + { + MemSet(ReplicationStates, 0, ReplicationIdentifierShmemSize()); + } +} + +/* --------------------------------------------------------------------------- + * Perform a checkpoint of replication identifier's progress with respect to + * the replayed remote_lsn. Make sure that all transactions we refer to in the + * checkpoint (local_lsn) are actually on-disk. This might not yet be the case + * if the transactions were originally committed asynchronously. + * + * We store checkpoints in the following format: + * +-------+-------------------------+-------------------------+-----+ + * | MAGIC | struct ReplicationState | struct ReplicationState | ... | EOF + * +-------+-------------------------+-------------------------+-----+ + * + * So its just the magic, followed by the statically sized + * ReplicationStates. Note that the maximum number of ReplicationStates is + * determined by max_logical_slots. + * + * FIXME: Add a CRC32 to the end. + * --------------------------------------------------------------------------- + */ +void +CheckPointReplicationIdentifier(XLogRecPtr ckpt) +{ + char tmppath[MAXPGPATH]; + char path[MAXPGPATH]; + int fd; + int tmpfd; + int i; + uint32 magic = REPLICATION_STATE_MAGIC; + + if (max_logical_slots == 0) + return; + + /* + * Write to a filename a LSN of the checkpoint's REDO pointer, so we can + * deal with the checkpoint failing after + * CheckPointReplicationIdentifier() finishing. + */ + sprintf(path, "pg_llog/checkpoints/%X-%X.ckpt", + (uint32)(ckpt >> 32), (uint32)ckpt); + sprintf(tmppath, "pg_llog/checkpoints/%X-%X.ckpt.tmp", + (uint32)(ckpt >> 32), (uint32)ckpt); + + /* check whether file already exists */ + fd = OpenTransientFile(path, + O_RDONLY | PG_BINARY, + 0); + + /* usual case, no checkpoint performed yet */ + if (fd < 0 && errno == ENOENT) + ; + else if (fd < 0) + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not check replication state checkpoint \"%s\": %m", + path))); + /* already checkpointed before crash during a checkpoint or so */ + else + { + CloseTransientFile(fd); + return; + } + + /* make sure no old temp file is remaining */ + if (unlink(tmppath) < 0 && errno != ENOENT) + ereport(PANIC, (errmsg("failed while unlinking %s", path))); + + /* + * no other backend can perform this at the same time, we're protected by + * CheckpointLock. + */ + tmpfd = OpenTransientFile(tmppath, + O_CREAT | O_EXCL | O_WRONLY | PG_BINARY, + S_IRUSR | S_IWUSR); + if (tmpfd < 0) + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not create replication identifier checkpoint \"%s\": %m", + tmppath))); + + /* write magic */ + if ((write(tmpfd, &magic, sizeof(magic))) != + sizeof(magic)) + { + CloseTransientFile(tmpfd); + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not write replication identifier checkpoint \"%s\": %m", + tmppath))); + } + + /* write actual data */ + for (i = 0; i < max_logical_slots; i++) + { + ReplicationState local_state; + + if (ReplicationStates[i].local_identifier == InvalidRepNodeId) + continue; + + local_state.local_identifier = ReplicationStates[i].local_identifier; + local_state.remote_lsn = ReplicationStates[i].remote_lsn; + local_state.local_lsn = InvalidXLogRecPtr; + + /* make sure we only write out a commit that's persistent */ + XLogFlush(ReplicationStates[i].local_lsn); + + if ((write(tmpfd, &local_state, sizeof(ReplicationState))) != + sizeof(ReplicationState)) + { + CloseTransientFile(tmpfd); + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not write replication identifier checkpoint \"%s\": %m", + tmppath))); + } + } + + /* fsync the file */ + if (pg_fsync(tmpfd) != 0) + { + CloseTransientFile(tmpfd); + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not fsync replication identifier checkpoint \"%s\": %m", + tmppath))); + } + + CloseTransientFile(tmpfd); + + /* rename to permanent file, fsync file and directory */ + if (rename(tmppath, path) != 0) + { + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not rename replication identifier checkpoint from \"%s\" to \"%s\": %m", + tmppath, path))); + } + + fsync_fname("pg_llog/checkpoints", true); + fsync_fname(path, false); +} + +/* + * Recover replication replay status from checkpoint data saved earlier by + * CheckPointReplicationIdentifier. + * + * This only needs to be called at startup and *not* during every checkpoint + * read during recovery (e.g. in HS or PITR from a base backup) afterwards. All + * state thereafter can be recovered by looking at commit records. + */ +void +StartupReplicationIdentifier(XLogRecPtr ckpt) +{ + char path[MAXPGPATH]; + int fd; + int readBytes; + uint32 magic = REPLICATION_STATE_MAGIC; + int last_state = 0; + + /* don't want to overwrite already existing state */ +#ifdef USE_ASSERT_CHECKING + static bool already_started = false; + Assert(!already_started); + already_started = true; +#endif + + if (max_logical_slots == 0) + return; + + elog(LOG, "starting up replication identifier with ckpt at %X/%X", + (uint32)(ckpt >> 32), (uint32)ckpt); + + sprintf(path, "pg_llog/checkpoints/%X-%X.ckpt", + (uint32)(ckpt >> 32), (uint32)ckpt); + + fd = OpenTransientFile(path, O_RDONLY | PG_BINARY, 0); + + /* + * might have had max_logical_slots == 0 last run, or we just brought up a + * standby. + */ + if (fd < 0 && errno == ENOENT) + return; + else if (fd < 0) + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not open replication state checkpoint \"%s\": %m", + path))); + + /* verify magic, thats written even if nothing was active */ + readBytes = read(fd, &magic, sizeof(magic)); + if (readBytes != sizeof(magic)) + ereport(PANIC, + (errmsg("could not read replication state checkpoint magic \"%s\": %m", + path))); + + if (magic != REPLICATION_STATE_MAGIC) + ereport(PANIC, + (errmsg("replication checkpoint has wrong magic %u instead of %u", + magic, REPLICATION_STATE_MAGIC))); + + /* recover individual states, until there are no more to be found */ + while (true) + { + ReplicationState local_state; + readBytes = read(fd, &local_state, sizeof(local_state)); + + /* no further data */ + if (readBytes == 0) + break; + + if (readBytes < 0) + { + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not read replication checkpoint file \"%s\": %m", + path))); + } + + if (readBytes != sizeof(local_state)) + { + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not read replication checkpoint file \"%s\": read %d of %zu", + path, readBytes, sizeof(local_state)))); + } + + if (last_state == max_logical_slots) + ereport(PANIC, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("no free replication state could be found, increase max_logical_slots"))); + + /* copy data shared memory */ + ReplicationStates[last_state++] = local_state; + + elog(LOG, "recovered replication state of node %u to %X/%X", + local_state.local_identifier, + (uint32)(local_state.remote_lsn >> 32), + (uint32)local_state.remote_lsn); + } + + CloseTransientFile(fd); +} + +/* + * Tell the replication identifier machinery that a commit from 'node' that + * originated at the LSN remote_commit on the remote node was replayed + * successfully and that we don't need to do so again. In combination with + * setting up replication_origin_lsn and replication_origin_id that ensures we + * won't loose knowledge about that after a crash if the the transaction had a + * persistent effect (think of asynchronous commits). + * + * local_commit needs to be a local LSN of the commit so that we can make sure + * uppon a checkpoint that enough WAL has been persisted to disk. + */ +void +AdvanceReplicationIdentifier(RepNodeId node, + XLogRecPtr remote_commit, + XLogRecPtr local_commit) +{ + int i; + int free_slot = -1; + ReplicationState *replication_state = NULL; + + /* + * XXX: should we restore into a hashtable and dump into shmem only after + * recovery finished? + */ + + /* check whether slot already exists */ + for (i = 0; i < max_logical_slots; i++) + { + /* remember where to insert if necessary */ + if (ReplicationStates[i].local_identifier == InvalidRepNodeId && + free_slot == -1) + { + free_slot = i; + continue; + } + + /* not our slot */ + if (ReplicationStates[i].local_identifier != node) + continue; + + /* ok, found slot */ + replication_state = &ReplicationStates[i]; + break; + } + + if (replication_state == NULL && free_slot == -1) + ereport(ERROR, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("no free replication state could be found for %u, increase max_logical_slots", + node))); + /* initialize new slot */ + else if (replication_state == NULL) + { + replication_state = &ReplicationStates[free_slot]; + Assert(replication_state->remote_lsn == InvalidXLogRecPtr); + Assert(replication_state->local_lsn == InvalidXLogRecPtr); + replication_state->local_identifier = node; + } + + /* + * Due to - harmless - race conditions during a checkpoint we could see + * values here that are older than the ones we already have in + * memory. Don't overwrite those. + */ + if (replication_state->remote_lsn < remote_commit) + replication_state->remote_lsn = remote_commit; + if (replication_state->local_lsn < local_commit) + replication_state->local_lsn = local_commit; +} + + +/* + * Setup a replication identifier in the shared memory struct if it doesn't + * already exists and cache access to the specific ReplicationSlot so the + * array doesn't have to be searched when calling + * AdvanceCachedReplicationIdentifier(). + * + * Obviously only one such cached identifier can exist per process and the + * current cached value can only be set again after the prvious value is torn + * down with TeardownCachedReplicationIdentifier. + */ +void +SetupCachedReplicationIdentifier(RepNodeId node) +{ + int i; + int free_slot = -1; + + Assert(max_logical_slots > 0); + + if (local_replication_state != NULL) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot setup replication origin when one is already setup"))); + + /* + * Search for either an existing slot for that identifier or a free one we + * can use. + */ + for (i = 0; i < max_logical_slots; i++) + { + /* remember where to insert if necessary */ + if (ReplicationStates[i].local_identifier == InvalidRepNodeId && + free_slot == -1) + { + free_slot = i; + continue; + } + + /* not our slot */ + if (ReplicationStates[i].local_identifier != node) + continue; + + local_replication_state = &ReplicationStates[i]; + } + + + if (local_replication_state == NULL && free_slot == -1) + ereport(ERROR, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("no free replication state could be found for %u, increase max_logical_slots", + node))); + else if (local_replication_state == NULL) + { + local_replication_state = &ReplicationStates[free_slot]; + local_replication_state->local_identifier = node; + Assert(local_replication_state->remote_lsn == InvalidXLogRecPtr); + Assert(local_replication_state->local_lsn == InvalidXLogRecPtr); + } +} + +/* + * Make currently cached replication identifier unavailable so a new one can + * be setup with SetupCachedReplicationIdentifier(). + * + * This function may only be called if a previous identifier was setup with + * SetupCachedReplicationIdentifier(). + */ +void +TeardownCachedReplicationIdentifier(void) +{ + Assert(max_logical_slots != 0); + + if (local_replication_state == NULL) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot setup replication origin when one is already setup"))); + + local_replication_state = NULL; +} + +/* + * Do the same work AdvanceReplicationIdentifier() does, just on a pre-cached + * identifier. This is noticeably cheaper if you only ever work on a single + * replication identifier. + */ +void +AdvanceCachedReplicationIdentifier(XLogRecPtr remote_commit, + XLogRecPtr local_commit) +{ + Assert(local_replication_state != NULL); + if (local_replication_state->local_lsn < local_commit) + local_replication_state->local_lsn = local_commit; + if (local_replication_state->remote_lsn < remote_commit) + local_replication_state->remote_lsn = remote_commit; +} + +/* + * Ask the machinery about the point up to which we successfully replayed + * changes from a already setup & cached replication identifier. + */ +XLogRecPtr +RemoteCommitFromCachedReplicationIdentifier(void) +{ + Assert(local_replication_state != NULL); + return local_replication_state->remote_lsn; +} diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index ed2d69f..868ebb4 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -31,6 +31,7 @@ #include "replication/logical.h" #include "replication/walreceiver.h" #include "replication/walsender.h" +#include "replication/replication_identifier.h" #include "storage/bufmgr.h" #include "storage/dsm.h" #include "storage/ipc.h" @@ -129,6 +130,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port) size = add_size(size, CheckpointerShmemSize()); size = add_size(size, AutoVacuumShmemSize()); size = add_size(size, LogicalDecodingShmemSize()); + size = add_size(size, ReplicationIdentifierShmemSize()); size = add_size(size, WalSndShmemSize()); size = add_size(size, WalRcvShmemSize()); size = add_size(size, BTreeShmemSize()); @@ -237,6 +239,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port) CheckpointerShmemInit(); AutoVacuumShmemInit(); LogicalDecodingShmemInit(); + ReplicationIdentifierShmemInit(); WalSndShmemInit(); WalRcvShmemInit(); diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c index e9bdfea..727405d 100644 --- a/src/backend/utils/cache/syscache.c +++ b/src/backend/utils/cache/syscache.c @@ -54,6 +54,7 @@ #include "catalog/pg_shdepend.h" #include "catalog/pg_shdescription.h" #include "catalog/pg_shseclabel.h" +#include "catalog/pg_replication_identifier.h" #include "catalog/pg_statistic.h" #include "catalog/pg_tablespace.h" #include "catalog/pg_ts_config.h" @@ -620,6 +621,28 @@ static const struct cachedesc cacheinfo[] = { }, 128 }, + {ReplicationIdentifierRelationId, /* REPLIDIDENT */ + ReplicationLocalIdentIndex, + 1, + { + Anum_pg_replication_riident, + 0, + 0, + 0 + }, + 16 + }, + {ReplicationIdentifierRelationId, /* REPLIDREMOTE */ + ReplicationExternalIdentIndex, + 1, + { + Anum_pg_replication_riname, + 0, + 0, + 0 + }, + 16 + }, {RewriteRelationId, /* RULERELNAME */ RewriteRelRulenameIndexId, 2, diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index a4146f0..1aacd6b 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -474,7 +474,6 @@ static bool data_checksums; static int wal_segment_size; static bool integer_datetimes; static int effective_io_concurrency; - /* should be static, but commands/variable.c needs to get at this */ char *role_string; diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index b7424e7..f1e991a 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -202,6 +202,7 @@ const char *subdirs[] = { "pg_stat", "pg_stat_tmp", "pg_llog", + "pg_llog/checkpoints", "pg_llog/snapshots", "pg_llog/mappings" }; diff --git a/src/bin/pg_resetxlog/pg_resetxlog.c b/src/bin/pg_resetxlog/pg_resetxlog.c index f1b5d6d..dbc5b74 100644 --- a/src/bin/pg_resetxlog/pg_resetxlog.c +++ b/src/bin/pg_resetxlog/pg_resetxlog.c @@ -55,6 +55,7 @@ #include "catalog/catversion.h" #include "catalog/pg_control.h" #include "common/fe_memutils.h" +#include "replication/logical.h" extern int optind; extern char *optarg; @@ -970,6 +971,7 @@ WriteEmptyXLOG(void) record->xl_len = sizeof(CheckPoint); record->xl_info = XLOG_CHECKPOINT_SHUTDOWN; record->xl_rmid = RM_XLOG_ID; + record->xl_origin_id = InvalidRepNodeId; memcpy(XLogRecGetData(record), &ControlFile.checkPointCopy, sizeof(CheckPoint)); diff --git a/src/include/access/xact.h b/src/include/access/xact.h index 9632378..9592a2d 100644 --- a/src/include/access/xact.h +++ b/src/include/access/xact.h @@ -146,6 +146,13 @@ typedef struct xl_xact_commit /* ARRAY OF SHARED INVALIDATION MESSAGES FOLLOWS */ } xl_xact_commit; +typedef struct xl_xact_origin +{ + XLogRecPtr origin_lsn; + RepNodeId origin_node_id; + TimestampTz origin_timestamp; +} xl_xact_origin; + #define MinSizeOfXactCommit offsetof(xl_xact_commit, xnodes) /* @@ -158,7 +165,7 @@ typedef struct xl_xact_commit */ #define XACT_COMPLETION_UPDATE_RELCACHE_FILE 0x01 #define XACT_COMPLETION_FORCE_SYNC_COMMIT 0x02 - +#define XACT_CONTAINS_ORIGIN 0x04 /* Access macros for above flags */ #define XactCompletionRelcacheInitFileInval(xinfo) (xinfo & XACT_COMPLETION_UPDATE_RELCACHE_FILE) #define XactCompletionForceSyncCommit(xinfo) (xinfo & XACT_COMPLETION_FORCE_SYNC_COMMIT) diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 7415a26..54e5dca 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -48,6 +48,7 @@ typedef struct XLogRecord /* 2 bytes of padding here, initialize to zero */ XLogRecPtr xl_prev; /* ptr to previous record in log */ pg_crc32 xl_crc; /* CRC for this record */ + RepNodeId xl_origin_id; /* what node did originally cause this record to be written */ /* If MAXALIGN==8, there are 4 wasted bytes here */ diff --git a/src/include/access/xlogdefs.h b/src/include/access/xlogdefs.h index bca166e..60fcc31 100644 --- a/src/include/access/xlogdefs.h +++ b/src/include/access/xlogdefs.h @@ -45,6 +45,12 @@ typedef uint64 XLogSegNo; typedef uint32 TimeLineID; /* + * Denotes the node on which the action causing a wal record to be logged + * originated on. + */ +typedef uint16 RepNodeId; + +/* * Because O_DIRECT bypasses the kernel buffers, and because we never * read those buffers except during crash recovery or if wal_level != minimal, * it is a win to use it in all cases where we sync on each write(). We could diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h index 4860e98..fd6f88e 100644 --- a/src/include/catalog/indexing.h +++ b/src/include/catalog/indexing.h @@ -313,6 +313,12 @@ DECLARE_UNIQUE_INDEX(pg_extension_name_index, 3081, on pg_extension using btree( DECLARE_UNIQUE_INDEX(pg_range_rngtypid_index, 3542, on pg_range using btree(rngtypid oid_ops)); #define RangeTypidIndexId 3542 +DECLARE_UNIQUE_INDEX(pg_replication_identifier_riiident_index, 3195, on pg_replication_identifier using btree(riident oid_ops)); +#define ReplicationLocalIdentIndex 3195 + +DECLARE_UNIQUE_INDEX(pg_replication_identifier_riname_index, 3196, on pg_replication_identifier using btree(riname varchar_pattern_ops)); +#define ReplicationExternalIdentIndex 3196 + /* last step of initialization script: build the indexes declared above */ BUILD_INDICES diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h index 815c8ae..fcce1fe 100644 --- a/src/include/catalog/pg_proc.h +++ b/src/include/catalog/pg_proc.h @@ -4778,6 +4778,27 @@ DESCR("stop logical replication"); /* event triggers */ DATA(insert OID = 3566 ( pg_event_trigger_dropped_objects PGNSP PGUID 12 10 100 0 0 f f f f t t s 0 0 2249 "" "{26,26,23,25,25,25,25}" "{o,o,o,o,o,o,o}" "{classid, objid, objsubid, object_type, schema_name, object_name, object_identity}" _null_ pg_event_trigger_dropped_objects _null_ _null_ _null_ )); DESCR("list objects dropped by the current command"); + +/* replication_identifier.h */ +DATA(insert OID = 3197 ( pg_replication_identifier_create PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 26 "25" _null_ _null_ _null_ _null_ pg_replication_identifier_create _null_ _null_ _null_ )); +DESCR("create local replication identifier for the passed external one"); + +DATA(insert OID = 3198 ( pg_replication_identifier_get PGNSP PGUID 12 1 0 0 0 f f f f t f s 1 0 26 "25" _null_ _null_ _null_ _null_ pg_replication_identifier_get _null_ _null_ _null_ )); +DESCR("translate the external node identifier to a local one"); + +DATA(insert OID = 3199 ( pg_replication_identifier_setup_replaying_from PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "25" _null_ _null_ _null_ _null_ pg_replication_identifier_setup_replaying_from _null_ _null_ _null_ )); +DESCR("setup from which node we are replaying transactions from currently"); + +DATA(insert OID = 3200 ( pg_replication_identifier_reset_replaying_from PGNSP PGUID 12 1 0 0 0 f f f f t f v 0 0 2278 "" _null_ _null_ _null_ _null_ pg_replication_identifier_reset_replaying_from _null_ _null_ _null_ )); +DESCR("reset replay mode"); + +DATA(insert OID = 3201 ( pg_replication_identifier_setup_tx_origin PGNSP PGUID 12 1 0 0 0 f f f f t f v 2 0 2278 "25 1184" _null_ _null_ _null_ _null_ pg_replication_identifier_setup_tx_origin _null_ _null_ _null_ )); +DESCR("setup transaction timestamp and origin lsn"); + +DATA(insert OID = 3202 ( pg_get_replication_identifier_progress PGNSP PGUID 12 1 100 0 0 f f f f f t v 0 0 2249 "" "{26,25,25,25}" "{o,o,o,o}" "{local_id, external_id, remote_lsn, local_lsn}" _null_ pg_get_replication_identifier_progress _null_ _null_ _null_ )); +DESCR("replication identifier progress"); + + /* * Symbolic values for provolatile column: these indicate whether the result * of a function is dependent *only* on the values of its explicit arguments, diff --git a/src/include/catalog/pg_replication_identifier.h b/src/include/catalog/pg_replication_identifier.h new file mode 100644 index 0000000..918adff --- /dev/null +++ b/src/include/catalog/pg_replication_identifier.h @@ -0,0 +1,75 @@ +/*------------------------------------------------------------------------- + * + * pg_replication_identifier.h + * Persistent Replication Node Identifiers + * + * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/catalog/pg_replication_identifier.h + * + * NOTES + * the genbki.pl script reads this file and generates .bki + * information from the DATA() statements. + * + *------------------------------------------------------------------------- + */ +#ifndef PG_REPLICATION_IDENTIFIER_H +#define PG_REPLICATION_IDENTIFIER_H + +#include "catalog/genbki.h" +#include "access/xlogdefs.h" + +/* ---------------- + * pg_replication_identifier. cpp turns this into + * typedef struct FormData_pg_replication_identifier + * ---------------- + */ +#define ReplicationIdentifierRelationId 3465 + +CATALOG(pg_replication_identifier,3465) BKI_SHARED_RELATION BKI_WITHOUT_OIDS +{ + /* + * locally known identifier that gets included into wal. + * + * This should never leave the system. + * + * Needs to fit into a uint16, so we don't waste too much space in WAL + * records. For this reason we don't use a normal Oid column here, since + * we need to handle allocation of new values manually. + */ + Oid riident; + + /* + * Variable-length fields start here, but we allow direct access to + * riname. + */ + + /* external, free-format, identifier */ + text riname; +#ifdef CATALOG_VARLEN /* further variable-length fields */ +#endif +} FormData_pg_replication_identifier; + +/* ---------------- + * Form_pg_extension corresponds to a pointer to a tuple with + * the format of pg_extension relation. + * ---------------- + */ +typedef FormData_pg_replication_identifier *Form_pg_replication_identifier; + +/* ---------------- + * compiler constants for pg_replication_identifier + * ---------------- + */ + +#define Natts_pg_replication_identifier 2 +#define Anum_pg_replication_riident 1 +#define Anum_pg_replication_riname 2 + +/* ---------------- + * pg_replication_identifier has no initial contents + * ---------------- + */ + +#endif /* PG_REPLICTION_IDENTIFIER_H */ diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h index 273b98f..eae7efc 100644 --- a/src/include/replication/logical.h +++ b/src/include/replication/logical.h @@ -160,6 +160,12 @@ typedef struct LogicalDecodingContext /* GUCs */ extern PGDLLIMPORT int max_logical_slots; +#define InvalidRepNodeId 0 +extern PGDLLIMPORT RepNodeId replication_origin_id; +extern PGDLLIMPORT XLogRecPtr replication_origin_lsn; +extern PGDLLIMPORT TimestampTz replication_origin_timestamp; + + extern Size LogicalDecodingShmemSize(void); extern void LogicalDecodingShmemInit(void); diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 4181ccf..ff5d29c 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -60,6 +60,8 @@ typedef struct ReorderBufferChange int action_internal; }; + RepNodeId origin_id; + /* * Context data for the change, which part of the union is valid depends * on action/action_internal. @@ -148,6 +150,12 @@ typedef struct ReorderBufferTXN */ XLogRecPtr restart_decoding_lsn; + /* origin of the change that caused this transaction */ + RepNodeId origin_id; + + /* did the TX have catalog changes */ + bool does_timetravel; + /* * Commit time, only known when we read the actual commit record. */ @@ -320,7 +328,7 @@ void ReorderBufferReturnChange(ReorderBuffer *, ReorderBufferChange *); void ReorderBufferQueueChange(ReorderBuffer *, TransactionId, XLogRecPtr lsn, ReorderBufferChange *); void ReorderBufferCommit(ReorderBuffer *, TransactionId, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, - TimestampTz commit_time); + TimestampTz commit_time, RepNodeId origin_id); void ReorderBufferAssignChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn); void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn, XLogRecPtr end_lsn); diff --git a/src/include/replication/replication_identifier.h b/src/include/replication/replication_identifier.h new file mode 100644 index 0000000..866009d --- /dev/null +++ b/src/include/replication/replication_identifier.h @@ -0,0 +1,45 @@ +/*------------------------------------------------------------------------- + * replication_identifier.h + * XXX + * + * Copyright (c) 2013, PostgreSQL Global Development Group + * + *------------------------------------------------------------------------- + */ +#ifndef REPLICATION_IDENTIFIER_H +#define REPLICATION_IDENTIFIER_H + +#include "catalog/pg_replication_identifier.h" +#include "replication/logical.h" + +/* API for querying & manipulating replication identifiers */ +extern RepNodeId GetReplicationIdentifier(char *name, bool missing_ok); +extern RepNodeId CreateReplicationIdentifier(char *name); +extern HeapTuple GetReplicationInfoByIdentifier(RepNodeId riident, bool missing_ok); + +extern void AdvanceReplicationIdentifier(RepNodeId node, + XLogRecPtr remote_commit, + XLogRecPtr local_commit); +extern void AdvanceCachedReplicationIdentifier(XLogRecPtr remote_commit, + XLogRecPtr local_commit); +extern void SetupCachedReplicationIdentifier(RepNodeId node); +extern void TeardownCachedReplicationIdentifier(void); +extern XLogRecPtr RemoteCommitFromCachedReplicationIdentifier(void); + +/* crash recovery support */ +extern void CheckPointReplicationIdentifier(XLogRecPtr ckpt); +extern void StartupReplicationIdentifier(XLogRecPtr ckpt); + +/* internals */ +extern Size ReplicationIdentifierShmemSize(void); +extern void ReplicationIdentifierShmemInit(void); + +/* SQL callable functions */ +extern Datum pg_replication_identifier_get(PG_FUNCTION_ARGS); +extern Datum pg_replication_identifier_create(PG_FUNCTION_ARGS); +extern Datum pg_replication_identifier_setup_replaying_from(PG_FUNCTION_ARGS); +extern Datum pg_replication_identifier_reset_replaying_from(PG_FUNCTION_ARGS); +extern Datum pg_replication_identifier_setup_tx_origin(PG_FUNCTION_ARGS); +extern Datum pg_get_replication_identifier_progress(PG_FUNCTION_ARGS); + +#endif diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h index e41b3d2..357f344 100644 --- a/src/include/utils/syscache.h +++ b/src/include/utils/syscache.h @@ -77,6 +77,8 @@ enum SysCacheIdentifier RANGETYPE, RELNAMENSP, RELOID, + REPLIDIDENT, + REPLIDREMOTE, RULERELNAME, STATRELATTINH, TABLESPACEOID, diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 72b0f48..446bbfe 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1367,6 +1367,11 @@ pg_prepared_xacts| SELECT p.transaction, FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid) LEFT JOIN pg_authid u ON ((p.ownerid = u.oid))) LEFT JOIN pg_database d ON ((p.dbid = d.oid))); +pg_replication_identifier_progress| SELECT pg_get_replication_identifier_progress.local_id, + pg_get_replication_identifier_progress.external_id, + pg_get_replication_identifier_progress.remote_lsn, + pg_get_replication_identifier_progress.local_lsn + FROM pg_get_replication_identifier_progress() pg_get_replication_identifier_progress(local_id, external_id, remote_lsn, local_lsn); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, pg_authid.rolinherit, diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out index a62a3e3..d5e40d6 100644 --- a/src/test/regress/expected/sanity_check.out +++ b/src/test/regress/expected/sanity_check.out @@ -120,6 +120,7 @@ pg_opfamily|t pg_pltemplate|t pg_proc|t pg_range|t +pg_replication_identifier|t pg_rewrite|t pg_seclabel|t pg_shdepend|t -- 1.8.5.rc2.dirty