TLA Line data Source code
1 : /*
2 : * brin_xlog.c
3 : * XLog replay routines for BRIN indexes
4 : *
5 : * Portions Copyright (c) 1996-2023, 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
24 CBC 22 : brin_xlog_createidx(XLogReaderState *record)
25 : {
26 22 : XLogRecPtr lsn = record->EndRecPtr;
27 22 : xl_brin_createidx *xlrec = (xl_brin_createidx *) XLogRecGetData(record);
28 : Buffer buf;
29 : Page page;
30 :
31 : /* create the index' metapage */
32 22 : buf = XLogInitBufferForRedo(record, 0);
33 22 : Assert(BufferIsValid(buf));
34 22 : page = (Page) BufferGetPage(buf);
35 22 : brin_metapage_init(page, xlrec->pagesPerRange, xlrec->version);
36 22 : PageSetLSN(page, lsn);
37 22 : MarkBufferDirty(buf);
38 22 : UnlockReleaseBuffer(buf);
39 22 : }
40 :
41 : /*
42 : * Common part of an insert or update. Inserts the new tuple and updates the
43 : * revmap.
44 : */
45 : static void
46 380 : brin_xlog_insert_update(XLogReaderState *record,
47 : xl_brin_insert *xlrec)
48 : {
49 380 : 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 380 : if (XLogRecGetInfo(record) & XLOG_BRIN_INIT_PAGE)
60 : {
61 43 : buffer = XLogInitBufferForRedo(record, 0);
62 43 : page = BufferGetPage(buffer);
63 43 : brin_page_init(page, BRIN_PAGETYPE_REGULAR);
64 43 : action = BLK_NEEDS_REDO;
65 : }
66 : else
67 : {
68 337 : action = XLogReadBufferForRedo(record, 0, &buffer);
69 : }
70 :
71 : /* need this page's blkno to store in revmap */
72 380 : regpgno = BufferGetBlockNumber(buffer);
73 :
74 : /* insert the index item into the page */
75 380 : if (action == BLK_NEEDS_REDO)
76 : {
77 : OffsetNumber offnum;
78 : BrinTuple *tuple;
79 : Size tuplen;
80 :
81 380 : tuple = (BrinTuple *) XLogRecGetBlockData(record, 0, &tuplen);
82 :
83 380 : Assert(tuple->bt_blkno == xlrec->heapBlk);
84 :
85 380 : page = (Page) BufferGetPage(buffer);
86 380 : offnum = xlrec->offnum;
87 380 : if (PageGetMaxOffsetNumber(page) + 1 < offnum)
88 UBC 0 : elog(PANIC, "brin_xlog_insert_update: invalid max offset number");
89 :
90 CBC 380 : offnum = PageAddItem(page, (Item) tuple, tuplen, offnum, true, false);
91 380 : if (offnum == InvalidOffsetNumber)
92 UBC 0 : elog(PANIC, "brin_xlog_insert_update: failed to add tuple");
93 :
94 CBC 380 : PageSetLSN(page, lsn);
95 380 : MarkBufferDirty(buffer);
96 : }
97 380 : if (BufferIsValid(buffer))
98 380 : UnlockReleaseBuffer(buffer);
99 :
100 : /* update the revmap */
101 380 : action = XLogReadBufferForRedo(record, 1, &buffer);
102 380 : if (action == BLK_NEEDS_REDO)
103 : {
104 : ItemPointerData tid;
105 :
106 380 : ItemPointerSet(&tid, regpgno, xlrec->offnum);
107 380 : page = (Page) BufferGetPage(buffer);
108 :
109 380 : brinSetHeapBlockItemptr(buffer, xlrec->pagesPerRange, xlrec->heapBlk,
110 : tid);
111 380 : PageSetLSN(page, lsn);
112 380 : MarkBufferDirty(buffer);
113 : }
114 380 : if (BufferIsValid(buffer))
115 380 : UnlockReleaseBuffer(buffer);
116 :
117 : /* XXX no FSM updates here ... */
118 380 : }
119 :
120 : /*
121 : * replay a BRIN index insertion
122 : */
123 : static void
124 376 : brin_xlog_insert(XLogReaderState *record)
125 : {
126 376 : xl_brin_insert *xlrec = (xl_brin_insert *) XLogRecGetData(record);
127 :
128 376 : brin_xlog_insert_update(record, xlrec);
129 376 : }
130 :
131 : /*
132 : * replay a BRIN index update
133 : */
134 : static void
135 4 : brin_xlog_update(XLogReaderState *record)
136 : {
137 4 : XLogRecPtr lsn = record->EndRecPtr;
138 4 : xl_brin_update *xlrec = (xl_brin_update *) XLogRecGetData(record);
139 : Buffer buffer;
140 : XLogRedoAction action;
141 :
142 : /* First remove the old tuple */
143 4 : action = XLogReadBufferForRedo(record, 2, &buffer);
144 4 : if (action == BLK_NEEDS_REDO)
145 : {
146 : Page page;
147 : OffsetNumber offnum;
148 :
149 4 : page = (Page) BufferGetPage(buffer);
150 :
151 4 : offnum = xlrec->oldOffnum;
152 :
153 4 : PageIndexTupleDeleteNoCompact(page, offnum);
154 :
155 4 : PageSetLSN(page, lsn);
156 4 : MarkBufferDirty(buffer);
157 : }
158 :
159 : /* Then insert the new tuple and update revmap, like in an insertion. */
160 4 : brin_xlog_insert_update(record, &xlrec->insert);
161 :
162 4 : if (BufferIsValid(buffer))
163 4 : UnlockReleaseBuffer(buffer);
164 4 : }
165 :
166 : /*
167 : * Update a tuple on a single page.
168 : */
169 : static void
170 524 : brin_xlog_samepage_update(XLogReaderState *record)
171 : {
172 524 : XLogRecPtr lsn = record->EndRecPtr;
173 : xl_brin_samepage_update *xlrec;
174 : Buffer buffer;
175 : XLogRedoAction action;
176 :
177 524 : xlrec = (xl_brin_samepage_update *) XLogRecGetData(record);
178 524 : action = XLogReadBufferForRedo(record, 0, &buffer);
179 524 : if (action == BLK_NEEDS_REDO)
180 : {
181 : Size tuplen;
182 : BrinTuple *brintuple;
183 : Page page;
184 : OffsetNumber offnum;
185 :
186 524 : brintuple = (BrinTuple *) XLogRecGetBlockData(record, 0, &tuplen);
187 :
188 524 : page = (Page) BufferGetPage(buffer);
189 :
190 524 : offnum = xlrec->offnum;
191 :
192 524 : if (!PageIndexTupleOverwrite(page, offnum, (Item) brintuple, tuplen))
193 UBC 0 : elog(PANIC, "brin_xlog_samepage_update: failed to replace tuple");
194 :
195 CBC 524 : PageSetLSN(page, lsn);
196 524 : MarkBufferDirty(buffer);
197 : }
198 524 : if (BufferIsValid(buffer))
199 524 : UnlockReleaseBuffer(buffer);
200 :
201 : /* XXX no FSM updates here ... */
202 524 : }
203 :
204 : /*
205 : * Replay a revmap page extension
206 : */
207 : static void
208 22 : brin_xlog_revmap_extend(XLogReaderState *record)
209 : {
210 22 : 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 22 : xlrec = (xl_brin_revmap_extend *) XLogRecGetData(record);
219 :
220 22 : XLogRecGetBlockTag(record, 1, NULL, NULL, &targetBlk);
221 22 : Assert(xlrec->targetBlk == targetBlk);
222 :
223 : /* Update the metapage */
224 22 : action = XLogReadBufferForRedo(record, 0, &metabuf);
225 22 : if (action == BLK_NEEDS_REDO)
226 : {
227 : Page metapg;
228 : BrinMetaPageData *metadata;
229 :
230 22 : metapg = BufferGetPage(metabuf);
231 22 : metadata = (BrinMetaPageData *) PageGetContents(metapg);
232 :
233 22 : Assert(metadata->lastRevmapPage == xlrec->targetBlk - 1);
234 22 : metadata->lastRevmapPage = xlrec->targetBlk;
235 :
236 22 : 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 22 : ((PageHeader) metapg)->pd_lower =
246 22 : ((char *) metadata + sizeof(BrinMetaPageData)) - (char *) metapg;
247 :
248 22 : 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 22 : buf = XLogInitBufferForRedo(record, 1);
257 22 : page = (Page) BufferGetPage(buf);
258 22 : brin_page_init(page, BRIN_PAGETYPE_REVMAP);
259 :
260 22 : PageSetLSN(page, lsn);
261 22 : MarkBufferDirty(buf);
262 :
263 22 : UnlockReleaseBuffer(buf);
264 22 : if (BufferIsValid(metabuf))
265 22 : UnlockReleaseBuffer(metabuf);
266 22 : }
267 :
268 : static void
269 7 : brin_xlog_desummarize_page(XLogReaderState *record)
270 : {
271 7 : XLogRecPtr lsn = record->EndRecPtr;
272 : xl_brin_desummarize *xlrec;
273 : Buffer buffer;
274 : XLogRedoAction action;
275 :
276 7 : xlrec = (xl_brin_desummarize *) XLogRecGetData(record);
277 :
278 : /* Update the revmap */
279 7 : action = XLogReadBufferForRedo(record, 0, &buffer);
280 7 : if (action == BLK_NEEDS_REDO)
281 : {
282 : ItemPointerData iptr;
283 :
284 7 : ItemPointerSetInvalid(&iptr);
285 7 : brinSetHeapBlockItemptr(buffer, xlrec->pagesPerRange, xlrec->heapBlk, iptr);
286 :
287 7 : PageSetLSN(BufferGetPage(buffer), lsn);
288 7 : MarkBufferDirty(buffer);
289 : }
290 7 : if (BufferIsValid(buffer))
291 7 : UnlockReleaseBuffer(buffer);
292 :
293 : /* remove the leftover entry from the regular page */
294 7 : action = XLogReadBufferForRedo(record, 1, &buffer);
295 7 : if (action == BLK_NEEDS_REDO)
296 : {
297 7 : Page regPg = BufferGetPage(buffer);
298 :
299 7 : PageIndexTupleDeleteNoCompact(regPg, xlrec->regOffset);
300 :
301 7 : PageSetLSN(regPg, lsn);
302 7 : MarkBufferDirty(buffer);
303 : }
304 7 : if (BufferIsValid(buffer))
305 7 : UnlockReleaseBuffer(buffer);
306 7 : }
307 :
308 : void
309 955 : brin_redo(XLogReaderState *record)
310 : {
311 955 : uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
312 :
313 955 : switch (info & XLOG_BRIN_OPMASK)
314 : {
315 22 : case XLOG_BRIN_CREATE_INDEX:
316 22 : brin_xlog_createidx(record);
317 22 : break;
318 376 : case XLOG_BRIN_INSERT:
319 376 : brin_xlog_insert(record);
320 376 : break;
321 4 : case XLOG_BRIN_UPDATE:
322 4 : brin_xlog_update(record);
323 4 : break;
324 524 : case XLOG_BRIN_SAMEPAGE_UPDATE:
325 524 : brin_xlog_samepage_update(record);
326 524 : break;
327 22 : case XLOG_BRIN_REVMAP_EXTEND:
328 22 : brin_xlog_revmap_extend(record);
329 22 : break;
330 7 : case XLOG_BRIN_DESUMMARIZE:
331 7 : brin_xlog_desummarize_page(record);
332 7 : break;
333 UBC 0 : default:
334 0 : elog(PANIC, "brin_redo: unknown op code %u", info);
335 : }
336 CBC 955 : }
337 :
338 : /*
339 : * Mask a BRIN page before doing consistency checks.
340 : */
341 : void
342 UBC 0 : brin_mask(char *pagedata, BlockNumber blkno)
343 : {
344 0 : Page page = (Page) pagedata;
345 0 : PageHeader pagehdr = (PageHeader) page;
346 :
347 0 : mask_page_lsn_and_checksum(page);
348 :
349 0 : 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 0 : if (BRIN_IS_REGULAR_PAGE(page) ||
357 0 : (BRIN_IS_META_PAGE(page) && pagehdr->pd_lower > SizeOfPageHeaderData))
358 : {
359 0 : mask_unused_space(page);
360 : }
361 :
362 : /*
363 : * BRIN_EVACUATE_PAGE is not WAL-logged, since it's of no use in recovery.
364 : * Mask it. See brin_start_evacuating_page() for details.
365 : */
366 0 : BrinPageFlags(page) &= ~BRIN_EVACUATE_PAGE;
367 0 : }
|
