Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit c6ff84b

Browse filesBrowse files
anarazelanarazel
committed
Emit invalidations to standby for transactions without xid.
So far, when a transaction with pending invalidations, but without an assigned xid, committed, we simply ignored those invalidation messages. That's problematic, because those are actually sent for a reason. Known symptoms of this include that existing sessions on a hot-standby replica sometimes fail to notice new concurrently built indexes and visibility map updates. The solution is to WAL log such invalidations in transactions without an xid. We considered to alternatively force-assign an xid, but that'd be problematic for vacuum, which might be run in systems with few xids. Important: This adds a new WAL record, but as the patch has to be back-patched, we can't bump the WAL page magic. This means that standbys have to be updated before primaries; otherwise "PANIC: standby_redo: unknown op code 32" errors can be encountered. XXX: Reported-By: Васильев Дмитрий, Masahiko Sawada Discussion: CAB-SwXY6oH=9twBkXJtgR4UC1NqT-vpYAtxCseME62ADwyK5OA@mail.gmail.com CAD21AoDpZ6Xjg=gFrGPnSn4oTRRcwK1EBrWCq9OqOHuAcMMC=w@mail.gmail.com
1 parent 2ac3be2 commit c6ff84b
Copy full SHA for c6ff84b

File tree

Expand file treeCollapse file tree

10 files changed

+181
-48
lines changed
Open diff view settings
Filter options
Expand file treeCollapse file tree

10 files changed

+181
-48
lines changed
Open diff view settings
Collapse file

‎src/backend/access/rmgrdesc/standbydesc.c‎

Copy file name to clipboardExpand all lines: src/backend/access/rmgrdesc/standbydesc.c
+54Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,14 @@ standby_desc(StringInfo buf, XLogReaderState *record)
5858

5959
standby_desc_running_xacts(buf, xlrec);
6060
}
61+
else if (info == XLOG_INVALIDATIONS)
62+
{
63+
xl_invalidations *xlrec = (xl_invalidations *) rec;
64+
65+
standby_desc_invalidations(buf, xlrec->nmsgs, xlrec->msgs,
66+
xlrec->dbId, xlrec->tsId,
67+
xlrec->relcacheInitFileInval);
68+
}
6169
}
6270

6371
const char *
@@ -73,7 +81,53 @@ standby_identify(uint8 info)
7381
case XLOG_RUNNING_XACTS:
7482
id = "RUNNING_XACTS";
7583
break;
84+
case XLOG_INVALIDATIONS:
85+
id = "INVALIDATIONS";
86+
break;
7687
}
7788

7889
return id;
7990
}
91+
92+
/*
93+
* This routine is used by both standby_desc and xact_desc, because
94+
* transaction commits and XLOG_INVALIDATIONS messages contain invalidations;
95+
* it seems pointless to duplicate the code.
96+
*/
97+
void
98+
standby_desc_invalidations(StringInfo buf,
99+
int nmsgs, SharedInvalidationMessage *msgs,
100+
Oid dbId, Oid tsId,
101+
bool relcacheInitFileInval)
102+
{
103+
int i;
104+
105+
if (relcacheInitFileInval)
106+
appendStringInfo(buf, "; relcache init file inval dbid %u tsid %u",
107+
dbId, tsId);
108+
109+
appendStringInfoString(buf, "; inval msgs:");
110+
for (i = 0; i < nmsgs; i++)
111+
{
112+
SharedInvalidationMessage *msg = &msgs[i];
113+
114+
if (msg->id >= 0)
115+
appendStringInfo(buf, " catcache %d", msg->id);
116+
else if (msg->id == SHAREDINVALCATALOG_ID)
117+
appendStringInfo(buf, " catalog %u", msg->cat.catId);
118+
else if (msg->id == SHAREDINVALRELCACHE_ID)
119+
appendStringInfo(buf, " relcache %u", msg->rc.relId);
120+
/* not expected, but print something anyway */
121+
else if (msg->id == SHAREDINVALSMGR_ID)
122+
appendStringInfoString(buf, " smgr");
123+
/* not expected, but print something anyway */
124+
else if (msg->id == SHAREDINVALRELMAP_ID)
125+
appendStringInfoString(buf, " relmap");
126+
else if (msg->id == SHAREDINVALRELMAP_ID)
127+
appendStringInfo(buf, " relmap db %u", msg->rm.dbId);
128+
else if (msg->id == SHAREDINVALSNAPSHOT_ID)
129+
appendStringInfo(buf, " snapshot %u", msg->sn.relId);
130+
else
131+
appendStringInfo(buf, " unknown id %d", msg->id);
132+
}
133+
}
Collapse file

‎src/backend/access/rmgrdesc/xactdesc.c‎

Copy file name to clipboardExpand all lines: src/backend/access/rmgrdesc/xactdesc.c
+4-26Lines changed: 4 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
#include "access/xact.h"
1919
#include "catalog/catalog.h"
2020
#include "storage/sinval.h"
21+
#include "storage/standbydefs.h"
2122
#include "utils/timestamp.h"
2223

2324
/*
@@ -203,32 +204,9 @@ xact_desc_commit(StringInfo buf, uint8 info, xl_xact_commit *xlrec, RepOriginId
203204
}
204205
if (parsed.nmsgs > 0)
205206
{
206-
if (XactCompletionRelcacheInitFileInval(parsed.xinfo))
207-
appendStringInfo(buf, "; relcache init file inval dbid %u tsid %u",
208-
parsed.dbId, parsed.tsId);
209-
210-
appendStringInfoString(buf, "; inval msgs:");
211-
for (i = 0; i < parsed.nmsgs; i++)
212-
{
213-
SharedInvalidationMessage *msg = &parsed.msgs[i];
214-
215-
if (msg->id >= 0)
216-
appendStringInfo(buf, " catcache %d", msg->id);
217-
else if (msg->id == SHAREDINVALCATALOG_ID)
218-
appendStringInfo(buf, " catalog %u", msg->cat.catId);
219-
else if (msg->id == SHAREDINVALRELCACHE_ID)
220-
appendStringInfo(buf, " relcache %u", msg->rc.relId);
221-
/* not expected, but print something anyway */
222-
else if (msg->id == SHAREDINVALSMGR_ID)
223-
appendStringInfoString(buf, " smgr");
224-
/* not expected, but print something anyway */
225-
else if (msg->id == SHAREDINVALRELMAP_ID)
226-
appendStringInfoString(buf, " relmap");
227-
else if (msg->id == SHAREDINVALSNAPSHOT_ID)
228-
appendStringInfo(buf, " snapshot %u", msg->sn.relId);
229-
else
230-
appendStringInfo(buf, " unknown id %d", msg->id);
231-
}
207+
standby_desc_invalidations(
208+
buf, parsed.nmsgs, parsed.msgs, parsed.dbId, parsed.tsId,
209+
XactCompletionRelcacheInitFileInval(parsed.xinfo));
232210
}
233211

234212
if (XactCompletionForceSyncCommit(parsed.xinfo))
Collapse file

‎src/backend/access/transam/xact.c‎

Copy file name to clipboardExpand all lines: src/backend/access/transam/xact.c
+18Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1163,6 +1163,24 @@ RecordTransactionCommit(void)
11631163
/* Can't have child XIDs either; AssignTransactionId enforces this */
11641164
Assert(nchildren == 0);
11651165

1166+
/*
1167+
* Transactions without an assigned xid can contain invalidation
1168+
* messages (e.g. explicit relcache invalidations or catcache
1169+
* invalidations for inplace updates); standbys need to process
1170+
* those. We can't emit a commit record without an xid, and we don't
1171+
* want to force assigning an xid, because that'd be problematic for
1172+
* e.g. vacuum. Hence we emit a bespoke record for the
1173+
* invalidations. We don't want to use that in case a commit record is
1174+
* emitted, so they happen synchronously with commits (besides not
1175+
* wanting to emit more WAL recoreds).
1176+
*/
1177+
if (nmsgs != 0)
1178+
{
1179+
LogStandbyInvalidations(nmsgs, invalMessages,
1180+
RelcacheInitFileInval);
1181+
wrote_xlog = true; /* not strictly necessary */
1182+
}
1183+
11661184
/*
11671185
* If we didn't create XLOG entries, we're done here; otherwise we
11681186
* should trigger flushing those entries the same as a commit record
Collapse file

‎src/backend/replication/logical/decode.c‎

Copy file name to clipboardExpand all lines: src/backend/replication/logical/decode.c
+9Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -327,6 +327,15 @@ DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
327327
break;
328328
case XLOG_STANDBY_LOCK:
329329
break;
330+
case XLOG_INVALIDATIONS:
331+
{
332+
xl_invalidations *invalidations =
333+
(xl_invalidations *) XLogRecGetData(r);
334+
335+
ReorderBufferImmediateInvalidation(
336+
ctx->reorder, invalidations->nmsgs, invalidations->msgs);
337+
}
338+
break;
330339
default:
331340
elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
332341
}
Collapse file

‎src/backend/replication/logical/reorderbuffer.c‎

Copy file name to clipboardExpand all lines: src/backend/replication/logical/reorderbuffer.c
+33-20Lines changed: 33 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1810,33 +1810,46 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
18101810
* catalog and we need to update the caches according to that.
18111811
*/
18121812
if (txn->base_snapshot != NULL && txn->ninvalidations > 0)
1813-
{
1814-
bool use_subtxn = IsTransactionOrTransactionBlock();
1815-
1816-
if (use_subtxn)
1817-
BeginInternalSubTransaction("replay");
1818-
1819-
/*
1820-
* Force invalidations to happen outside of a valid transaction - that
1821-
* way entries will just be marked as invalid without accessing the
1822-
* catalog. That's advantageous because we don't need to setup the
1823-
* full state necessary for catalog access.
1824-
*/
1825-
if (use_subtxn)
1826-
AbortCurrentTransaction();
1827-
1828-
ReorderBufferExecuteInvalidations(rb, txn);
1829-
1830-
if (use_subtxn)
1831-
RollbackAndReleaseCurrentSubTransaction();
1832-
}
1813+
ReorderBufferImmediateInvalidation(rb, txn->ninvalidations,
1814+
txn->invalidations);
18331815
else
18341816
Assert(txn->ninvalidations == 0);
18351817

18361818
/* remove potential on-disk data, and deallocate */
18371819
ReorderBufferCleanupTXN(rb, txn);
18381820
}
18391821

1822+
/*
1823+
* Execute invalidations happening outside the context of a decoded
1824+
* transaction. That currently happens either for xid-less commits
1825+
* (c.f. RecordTransactionCommit()) or for invalidations in uninteresting
1826+
* transactions (via ReorderBufferForget()).
1827+
*/
1828+
void
1829+
ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations,
1830+
SharedInvalidationMessage *invalidations)
1831+
{
1832+
bool use_subtxn = IsTransactionOrTransactionBlock();
1833+
int i;
1834+
1835+
if (use_subtxn)
1836+
BeginInternalSubTransaction("replay");
1837+
1838+
/*
1839+
* Force invalidations to happen outside of a valid transaction - that
1840+
* way entries will just be marked as invalid without accessing the
1841+
* catalog. That's advantageous because we don't need to setup the
1842+
* full state necessary for catalog access.
1843+
*/
1844+
if (use_subtxn)
1845+
AbortCurrentTransaction();
1846+
1847+
for (i = 0; i < ninvalidations; i++)
1848+
LocalExecuteInvalidationMessage(&invalidations[i]);
1849+
1850+
if (use_subtxn)
1851+
RollbackAndReleaseCurrentSubTransaction();
1852+
}
18401853

18411854
/*
18421855
* Tell reorderbuffer about an xid seen in the WAL stream. Has to be called at
Collapse file

‎src/backend/storage/ipc/standby.c‎

Copy file name to clipboardExpand all lines: src/backend/storage/ipc/standby.c
+35Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -825,6 +825,16 @@ standby_redo(XLogReaderState *record)
825825

826826
ProcArrayApplyRecoveryInfo(&running);
827827
}
828+
else if (info == XLOG_INVALIDATIONS)
829+
{
830+
xl_invalidations *xlrec = (xl_invalidations *) XLogRecGetData(record);
831+
832+
ProcessCommittedInvalidationMessages(xlrec->msgs,
833+
xlrec->nmsgs,
834+
xlrec->relcacheInitFileInval,
835+
xlrec->dbId,
836+
xlrec->tsId);
837+
}
828838
else
829839
elog(PANIC, "standby_redo: unknown op code %u", info);
830840
}
@@ -1068,3 +1078,28 @@ LogAccessExclusiveLockPrepare(void)
10681078
*/
10691079
(void) GetTopTransactionId();
10701080
}
1081+
1082+
/*
1083+
* Emit WAL for invalidations. This currently is only used for commits without
1084+
* an xid but which contain invalidations.
1085+
*/
1086+
void
1087+
LogStandbyInvalidations(int nmsgs, SharedInvalidationMessage *msgs,
1088+
bool relcacheInitFileInval)
1089+
{
1090+
xl_invalidations xlrec;
1091+
1092+
/* prepare record */
1093+
memset(&xlrec, 0, sizeof(xlrec));
1094+
xlrec.dbId = MyDatabaseId;
1095+
xlrec.tsId = MyDatabaseTableSpace;
1096+
xlrec.relcacheInitFileInval = relcacheInitFileInval;
1097+
xlrec.nmsgs = nmsgs;
1098+
1099+
/* perform insertion */
1100+
XLogBeginInsert();
1101+
XLogRegisterData((char *) (&xlrec), MinSizeOfInvalidations);
1102+
XLogRegisterData((char *) msgs,
1103+
nmsgs * sizeof(SharedInvalidationMessage));
1104+
XLogInsert(RM_STANDBY_ID, XLOG_INVALIDATIONS);
1105+
}
Collapse file

‎src/backend/utils/cache/inval.c‎

Copy file name to clipboardExpand all lines: src/backend/utils/cache/inval.c
+3-2Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -842,8 +842,9 @@ xactGetCommittedInvalidationMessages(SharedInvalidationMessage **msgs,
842842
}
843843

844844
/*
845-
* ProcessCommittedInvalidationMessages is executed by xact_redo_commit()
846-
* to process invalidation messages added to commit records.
845+
* ProcessCommittedInvalidationMessages is executed by xact_redo_commit() or
846+
* standby_redo() to process invalidation messages. Currently that happens
847+
* only at end-of-xact.
847848
*
848849
* Relcache init file invalidation requires processing both
849850
* before and after we send the SI messages. See AtEOXact_Inval()
Collapse file

‎src/include/replication/reorderbuffer.h‎

Copy file name to clipboardExpand all lines: src/include/replication/reorderbuffer.h
+2Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -391,6 +391,8 @@ void ReorderBufferAddNewTupleCids(ReorderBuffer *, TransactionId, XLogRecPtr lsn
391391
CommandId cmin, CommandId cmax, CommandId combocid);
392392
void ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
393393
Size nmsgs, SharedInvalidationMessage *msgs);
394+
void ReorderBufferImmediateInvalidation(ReorderBuffer *, uint32 ninvalidations,
395+
SharedInvalidationMessage *invalidations);
394396
void ReorderBufferProcessXid(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);
395397
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);
396398
bool ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
Collapse file

‎src/include/storage/standby.h‎

Copy file name to clipboardExpand all lines: src/include/storage/standby.h
+2Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,5 +85,7 @@ extern void LogAccessExclusiveLock(Oid dbOid, Oid relOid);
8585
extern void LogAccessExclusiveLockPrepare(void);
8686

8787
extern XLogRecPtr LogStandbySnapshot(void);
88+
extern void LogStandbyInvalidations(int nmsgs, SharedInvalidationMessage *msgs,
89+
bool relcacheInitFileInval);
8890

8991
#endif /* STANDBY_H */
Collapse file

‎src/include/storage/standbydefs.h‎

Copy file name to clipboardExpand all lines: src/include/storage/standbydefs.h
+21Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,17 +17,23 @@
1717
#include "access/xlogreader.h"
1818
#include "lib/stringinfo.h"
1919
#include "storage/lockdefs.h"
20+
#include "storage/sinval.h"
2021

2122
/* Recovery handlers for the Standby Rmgr (RM_STANDBY_ID) */
2223
extern void standby_redo(XLogReaderState *record);
2324
extern void standby_desc(StringInfo buf, XLogReaderState *record);
2425
extern const char *standby_identify(uint8 info);
26+
extern void standby_desc_invalidations(StringInfo buf,
27+
int nmsgs, SharedInvalidationMessage *msgs,
28+
Oid dbId, Oid tsId,
29+
bool relcacheInitFileInval);
2530

2631
/*
2732
* XLOG message types
2833
*/
2934
#define XLOG_STANDBY_LOCK 0x00
3035
#define XLOG_RUNNING_XACTS 0x10
36+
#define XLOG_INVALIDATIONS 0x20
3137

3238
typedef struct xl_standby_locks
3339
{
@@ -50,4 +56,19 @@ typedef struct xl_running_xacts
5056
TransactionId xids[FLEXIBLE_ARRAY_MEMBER];
5157
} xl_running_xacts;
5258

59+
/*
60+
* Invalidations for standby, currently only when transactions without an
61+
* assigned xid commit.
62+
*/
63+
typedef struct xl_invalidations
64+
{
65+
Oid dbId; /* MyDatabaseId */
66+
Oid tsId; /* MyDatabaseTableSpace */
67+
bool relcacheInitFileInval; /* invalidate relcache init file */
68+
int nmsgs; /* number of shared inval msgs */
69+
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER];
70+
} xl_invalidations;
71+
72+
#define MinSizeOfInvalidations offsetof(xl_invalidations, msgs)
73+
5374
#endif /* STANDBYDEFS_H */

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.