1 /*
2 * brin_xlog.c
3 * XLog replay routines for BRIN indexes
4 *
5 * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
6 * Portions Copyright (c) 1994, Regents of the University of California
7 *
8 * IDENTIFICATION
9 * src/backend/access/brin/brin_xlog.c
10 */
11 #include "postgres.h"
12
13 #include "access/brin_page.h"
14 #include "access/brin_pageops.h"
15 #include "access/brin_xlog.h"
16 #include "access/bufmask.h"
17 #include "access/xlogutils.h"
18
19
20 /*
21 * xlog replay routines
22 */
23 static void
brin_xlog_createidx(XLogReaderState * record)24 brin_xlog_createidx(XLogReaderState *record)
25 {
26 XLogRecPtr lsn = record->EndRecPtr;
27 xl_brin_createidx *xlrec = (xl_brin_createidx *) XLogRecGetData(record);
28 Buffer buf;
29 Page page;
30
31 /* create the index' metapage */
32 buf = XLogInitBufferForRedo(record, 0);
33 Assert(BufferIsValid(buf));
34 page = (Page) BufferGetPage(buf);
35 brin_metapage_init(page, xlrec->pagesPerRange, xlrec->version);
36 PageSetLSN(page, lsn);
37 MarkBufferDirty(buf);
38 UnlockReleaseBuffer(buf);
39 }
40
41 /*
42 * Common part of an insert or update. Inserts the new tuple and updates the
43 * revmap.
44 */
45 static void
brin_xlog_insert_update(XLogReaderState * record,xl_brin_insert * xlrec)46 brin_xlog_insert_update(XLogReaderState *record,
47 xl_brin_insert *xlrec)
48 {
49 XLogRecPtr lsn = record->EndRecPtr;
50 Buffer buffer;
51 BlockNumber regpgno;
52 Page page;
53 XLogRedoAction action;
54
55 /*
56 * If we inserted the first and only tuple on the page, re-initialize the
57 * page from scratch.
58 */
59 if (XLogRecGetInfo(record) & XLOG_BRIN_INIT_PAGE)
60 {
61 buffer = XLogInitBufferForRedo(record, 0);
62 page = BufferGetPage(buffer);
63 brin_page_init(page, BRIN_PAGETYPE_REGULAR);
64 action = BLK_NEEDS_REDO;
65 }
66 else
67 {
68 action = XLogReadBufferForRedo(record, 0, &buffer);
69 }
70
71 /* need this page's blkno to store in revmap */
72 regpgno = BufferGetBlockNumber(buffer);
73
74 /* insert the index item into the page */
75 if (action == BLK_NEEDS_REDO)
76 {
77 OffsetNumber offnum;
78 BrinTuple *tuple;
79 Size tuplen;
80
81 tuple = (BrinTuple *) XLogRecGetBlockData(record, 0, &tuplen);
82
83 Assert(tuple->bt_blkno == xlrec->heapBlk);
84
85 page = (Page) BufferGetPage(buffer);
86 offnum = xlrec->offnum;
87 if (PageGetMaxOffsetNumber(page) + 1 < offnum)
88 elog(PANIC, "brin_xlog_insert_update: invalid max offset number");
89
90 offnum = PageAddItem(page, (Item) tuple, tuplen, offnum, true, false);
91 if (offnum == InvalidOffsetNumber)
92 elog(PANIC, "brin_xlog_insert_update: failed to add tuple");
93
94 PageSetLSN(page, lsn);
95 MarkBufferDirty(buffer);
96 }
97 if (BufferIsValid(buffer))
98 UnlockReleaseBuffer(buffer);
99
100 /* update the revmap */
101 action = XLogReadBufferForRedo(record, 1, &buffer);
102 if (action == BLK_NEEDS_REDO)
103 {
104 ItemPointerData tid;
105
106 ItemPointerSet(&tid, regpgno, xlrec->offnum);
107 page = (Page) BufferGetPage(buffer);
108
109 brinSetHeapBlockItemptr(buffer, xlrec->pagesPerRange, xlrec->heapBlk,
110 tid);
111 PageSetLSN(page, lsn);
112 MarkBufferDirty(buffer);
113 }
114 if (BufferIsValid(buffer))
115 UnlockReleaseBuffer(buffer);
116
117 /* XXX no FSM updates here ... */
118 }
119
120 /*
121 * replay a BRIN index insertion
122 */
123 static void
brin_xlog_insert(XLogReaderState * record)124 brin_xlog_insert(XLogReaderState *record)
125 {
126 xl_brin_insert *xlrec = (xl_brin_insert *) XLogRecGetData(record);
127
128 brin_xlog_insert_update(record, xlrec);
129 }
130
131 /*
132 * replay a BRIN index update
133 */
134 static void
brin_xlog_update(XLogReaderState * record)135 brin_xlog_update(XLogReaderState *record)
136 {
137 XLogRecPtr lsn = record->EndRecPtr;
138 xl_brin_update *xlrec = (xl_brin_update *) XLogRecGetData(record);
139 Buffer buffer;
140 XLogRedoAction action;
141
142 /* First remove the old tuple */
143 action = XLogReadBufferForRedo(record, 2, &buffer);
144 if (action == BLK_NEEDS_REDO)
145 {
146 Page page;
147 OffsetNumber offnum;
148
149 page = (Page) BufferGetPage(buffer);
150
151 offnum = xlrec->oldOffnum;
152
153 PageIndexTupleDeleteNoCompact(page, offnum);
154
155 PageSetLSN(page, lsn);
156 MarkBufferDirty(buffer);
157 }
158
159 /* Then insert the new tuple and update revmap, like in an insertion. */
160 brin_xlog_insert_update(record, &xlrec->insert);
161
162 if (BufferIsValid(buffer))
163 UnlockReleaseBuffer(buffer);
164 }
165
166 /*
167 * Update a tuple on a single page.
168 */
169 static void
brin_xlog_samepage_update(XLogReaderState * record)170 brin_xlog_samepage_update(XLogReaderState *record)
171 {
172 XLogRecPtr lsn = record->EndRecPtr;
173 xl_brin_samepage_update *xlrec;
174 Buffer buffer;
175 XLogRedoAction action;
176
177 xlrec = (xl_brin_samepage_update *) XLogRecGetData(record);
178 action = XLogReadBufferForRedo(record, 0, &buffer);
179 if (action == BLK_NEEDS_REDO)
180 {
181 Size tuplen;
182 BrinTuple *brintuple;
183 Page page;
184 OffsetNumber offnum;
185
186 brintuple = (BrinTuple *) XLogRecGetBlockData(record, 0, &tuplen);
187
188 page = (Page) BufferGetPage(buffer);
189
190 offnum = xlrec->offnum;
191
192 if (!PageIndexTupleOverwrite(page, offnum, (Item) brintuple, tuplen))
193 elog(PANIC, "brin_xlog_samepage_update: failed to replace tuple");
194
195 PageSetLSN(page, lsn);
196 MarkBufferDirty(buffer);
197 }
198 if (BufferIsValid(buffer))
199 UnlockReleaseBuffer(buffer);
200
201 /* XXX no FSM updates here ... */
202 }
203
204 /*
205 * Replay a revmap page extension
206 */
207 static void
brin_xlog_revmap_extend(XLogReaderState * record)208 brin_xlog_revmap_extend(XLogReaderState *record)
209 {
210 XLogRecPtr lsn = record->EndRecPtr;
211 xl_brin_revmap_extend *xlrec;
212 Buffer metabuf;
213 Buffer buf;
214 Page page;
215 BlockNumber targetBlk;
216 XLogRedoAction action;
217
218 xlrec = (xl_brin_revmap_extend *) XLogRecGetData(record);
219
220 XLogRecGetBlockTag(record, 1, NULL, NULL, &targetBlk);
221 Assert(xlrec->targetBlk == targetBlk);
222
223 /* Update the metapage */
224 action = XLogReadBufferForRedo(record, 0, &metabuf);
225 if (action == BLK_NEEDS_REDO)
226 {
227 Page metapg;
228 BrinMetaPageData *metadata;
229
230 metapg = BufferGetPage(metabuf);
231 metadata = (BrinMetaPageData *) PageGetContents(metapg);
232
233 Assert(metadata->lastRevmapPage == xlrec->targetBlk - 1);
234 metadata->lastRevmapPage = xlrec->targetBlk;
235
236 PageSetLSN(metapg, lsn);
237
238 /*
239 * Set pd_lower just past the end of the metadata. This is essential,
240 * because without doing so, metadata will be lost if xlog.c
241 * compresses the page. (We must do this here because pre-v11
242 * versions of PG did not set the metapage's pd_lower correctly, so a
243 * pg_upgraded index might contain the wrong value.)
244 */
245 ((PageHeader) metapg)->pd_lower =
246 ((char *) metadata + sizeof(BrinMetaPageData)) - (char *) metapg;
247
248 MarkBufferDirty(metabuf);
249 }
250
251 /*
252 * Re-init the target block as a revmap page. There's never a full- page
253 * image here.
254 */
255
256 buf = XLogInitBufferForRedo(record, 1);
257 page = (Page) BufferGetPage(buf);
258 brin_page_init(page, BRIN_PAGETYPE_REVMAP);
259
260 PageSetLSN(page, lsn);
261 MarkBufferDirty(buf);
262
263 UnlockReleaseBuffer(buf);
264 if (BufferIsValid(metabuf))
265 UnlockReleaseBuffer(metabuf);
266 }
267
268 static void
brin_xlog_desummarize_page(XLogReaderState * record)269 brin_xlog_desummarize_page(XLogReaderState *record)
270 {
271 XLogRecPtr lsn = record->EndRecPtr;
272 xl_brin_desummarize *xlrec;
273 Buffer buffer;
274 XLogRedoAction action;
275
276 xlrec = (xl_brin_desummarize *) XLogRecGetData(record);
277
278 /* Update the revmap */
279 action = XLogReadBufferForRedo(record, 0, &buffer);
280 if (action == BLK_NEEDS_REDO)
281 {
282 ItemPointerData iptr;
283
284 ItemPointerSetInvalid(&iptr);
285 brinSetHeapBlockItemptr(buffer, xlrec->pagesPerRange, xlrec->heapBlk, iptr);
286
287 PageSetLSN(BufferGetPage(buffer), lsn);
288 MarkBufferDirty(buffer);
289 }
290 if (BufferIsValid(buffer))
291 UnlockReleaseBuffer(buffer);
292
293 /* remove the leftover entry from the regular page */
294 action = XLogReadBufferForRedo(record, 1, &buffer);
295 if (action == BLK_NEEDS_REDO)
296 {
297 Page regPg = BufferGetPage(buffer);
298
299 PageIndexTupleDeleteNoCompact(regPg, xlrec->regOffset);
300
301 PageSetLSN(regPg, lsn);
302 MarkBufferDirty(buffer);
303 }
304 if (BufferIsValid(buffer))
305 UnlockReleaseBuffer(buffer);
306 }
307
308 void
brin_redo(XLogReaderState * record)309 brin_redo(XLogReaderState *record)
310 {
311 uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
312
313 switch (info & XLOG_BRIN_OPMASK)
314 {
315 case XLOG_BRIN_CREATE_INDEX:
316 brin_xlog_createidx(record);
317 break;
318 case XLOG_BRIN_INSERT:
319 brin_xlog_insert(record);
320 break;
321 case XLOG_BRIN_UPDATE:
322 brin_xlog_update(record);
323 break;
324 case XLOG_BRIN_SAMEPAGE_UPDATE:
325 brin_xlog_samepage_update(record);
326 break;
327 case XLOG_BRIN_REVMAP_EXTEND:
328 brin_xlog_revmap_extend(record);
329 break;
330 case XLOG_BRIN_DESUMMARIZE:
331 brin_xlog_desummarize_page(record);
332 break;
333 default:
334 elog(PANIC, "brin_redo: unknown op code %u", info);
335 }
336 }
337
338 /*
339 * Mask a BRIN page before doing consistency checks.
340 */
341 void
brin_mask(char * pagedata,BlockNumber blkno)342 brin_mask(char *pagedata, BlockNumber blkno)
343 {
344 Page page = (Page) pagedata;
345 PageHeader pagehdr = (PageHeader) page;
346
347 mask_page_lsn_and_checksum(page);
348
349 mask_page_hint_bits(page);
350
351 /*
352 * Regular brin pages contain unused space which needs to be masked.
353 * Similarly for meta pages, but mask it only if pd_lower appears to have
354 * been set correctly.
355 */
356 if (BRIN_IS_REGULAR_PAGE(page) ||
357 (BRIN_IS_META_PAGE(page) && pagehdr->pd_lower > SizeOfPageHeaderData))
358 {
359 mask_unused_space(page);
360 }
361 }
362