Age Owner TLA Line data Source code
1 : /*
2 : * brin_revmap.c
3 : * Range map for BRIN indexes
4 : *
5 : * The range map (revmap) is a translation structure for BRIN indexes: for each
6 : * page range there is one summary tuple, and its location is tracked by the
7 : * revmap. Whenever a new tuple is inserted into a table that violates the
8 : * previously recorded summary values, a new tuple is inserted into the index
9 : * and the revmap is updated to point to it.
10 : *
11 : * The revmap is stored in the first pages of the index, immediately following
12 : * the metapage. When the revmap needs to be expanded, all tuples on the
13 : * regular BRIN page at that block (if any) are moved out of the way.
14 : *
15 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
16 : * Portions Copyright (c) 1994, Regents of the University of California
17 : *
18 : * IDENTIFICATION
19 : * src/backend/access/brin/brin_revmap.c
20 : */
21 : #include "postgres.h"
22 :
23 : #include "access/brin_page.h"
24 : #include "access/brin_pageops.h"
25 : #include "access/brin_revmap.h"
26 : #include "access/brin_tuple.h"
27 : #include "access/brin_xlog.h"
28 : #include "access/rmgr.h"
29 : #include "access/xloginsert.h"
30 : #include "miscadmin.h"
31 : #include "storage/bufmgr.h"
32 : #include "storage/lmgr.h"
33 : #include "utils/rel.h"
34 :
35 :
36 : /*
37 : * In revmap pages, each item stores an ItemPointerData. These defines let one
38 : * find the logical revmap page number and index number of the revmap item for
39 : * the given heap block number.
40 : */
41 : #define HEAPBLK_TO_REVMAP_BLK(pagesPerRange, heapBlk) \
42 : ((heapBlk / pagesPerRange) / REVMAP_PAGE_MAXITEMS)
43 : #define HEAPBLK_TO_REVMAP_INDEX(pagesPerRange, heapBlk) \
44 : ((heapBlk / pagesPerRange) % REVMAP_PAGE_MAXITEMS)
45 :
46 :
47 : struct BrinRevmap
48 : {
49 : Relation rm_irel;
50 : BlockNumber rm_pagesPerRange;
51 : BlockNumber rm_lastRevmapPage; /* cached from the metapage */
52 : Buffer rm_metaBuf;
53 : Buffer rm_currBuf;
54 : };
55 :
56 : /* typedef appears in brin_revmap.h */
57 :
58 :
59 : static BlockNumber revmap_get_blkno(BrinRevmap *revmap,
60 : BlockNumber heapBlk);
61 : static Buffer revmap_get_buffer(BrinRevmap *revmap, BlockNumber heapBlk);
62 : static BlockNumber revmap_extend_and_get_blkno(BrinRevmap *revmap,
63 : BlockNumber heapBlk);
64 : static void revmap_physical_extend(BrinRevmap *revmap);
65 :
66 : /*
67 : * Initialize an access object for a range map. This must be freed by
68 : * brinRevmapTerminate when caller is done with it.
69 : */
70 : BrinRevmap *
2557 kgrittn 71 CBC 40518 : brinRevmapInitialize(Relation idxrel, BlockNumber *pagesPerRange,
72 : Snapshot snapshot)
73 : {
74 : BrinRevmap *revmap;
75 : Buffer meta;
76 : BrinMetaPageData *metadata;
77 : Page page;
78 :
3075 alvherre 79 40518 : meta = ReadBuffer(idxrel, BRIN_METAPAGE_BLKNO);
80 40518 : LockBuffer(meta, BUFFER_LOCK_SHARE);
2545 kgrittn 81 40518 : page = BufferGetPage(meta);
82 40518 : TestForOldSnapshot(snapshot, idxrel, page);
2557 83 40518 : metadata = (BrinMetaPageData *) PageGetContents(page);
84 :
3075 alvherre 85 40518 : revmap = palloc(sizeof(BrinRevmap));
86 40518 : revmap->rm_irel = idxrel;
87 40518 : revmap->rm_pagesPerRange = metadata->pagesPerRange;
88 40518 : revmap->rm_lastRevmapPage = metadata->lastRevmapPage;
89 40518 : revmap->rm_metaBuf = meta;
90 40518 : revmap->rm_currBuf = InvalidBuffer;
91 :
92 40518 : *pagesPerRange = metadata->pagesPerRange;
93 :
94 40518 : LockBuffer(meta, BUFFER_LOCK_UNLOCK);
95 :
96 40518 : return revmap;
97 : }
98 :
99 : /*
100 : * Release resources associated with a revmap access object.
101 : */
102 : void
103 40518 : brinRevmapTerminate(BrinRevmap *revmap)
104 : {
105 40518 : ReleaseBuffer(revmap->rm_metaBuf);
106 40518 : if (revmap->rm_currBuf != InvalidBuffer)
107 39555 : ReleaseBuffer(revmap->rm_currBuf);
108 40518 : pfree(revmap);
109 40518 : }
110 :
111 : /*
112 : * Extend the revmap to cover the given heap block number.
113 : */
114 : void
115 6027 : brinRevmapExtend(BrinRevmap *revmap, BlockNumber heapBlk)
116 : {
117 : BlockNumber mapBlk PG_USED_FOR_ASSERTS_ONLY;
118 :
119 6027 : mapBlk = revmap_extend_and_get_blkno(revmap, heapBlk);
120 :
121 : /* Ensure the buffer we got is in the expected range */
122 6027 : Assert(mapBlk != InvalidBlockNumber &&
123 : mapBlk != BRIN_METAPAGE_BLKNO &&
124 : mapBlk <= revmap->rm_lastRevmapPage);
125 6027 : }
126 :
127 : /*
128 : * Prepare to insert an entry into the revmap; the revmap buffer in which the
129 : * entry is to reside is locked and returned. Most callers should call
130 : * brinRevmapExtend beforehand, as this routine does not extend the revmap if
131 : * it's not long enough.
132 : *
133 : * The returned buffer is also recorded in the revmap struct; finishing that
134 : * releases the buffer, therefore the caller needn't do it explicitly.
135 : */
136 : Buffer
137 2931 : brinLockRevmapPageForUpdate(BrinRevmap *revmap, BlockNumber heapBlk)
138 : {
139 : Buffer rmBuf;
140 :
141 2931 : rmBuf = revmap_get_buffer(revmap, heapBlk);
142 2931 : LockBuffer(rmBuf, BUFFER_LOCK_EXCLUSIVE);
143 :
144 2931 : return rmBuf;
145 : }
146 :
147 : /*
148 : * In the given revmap buffer (locked appropriately by caller), which is used
149 : * in a BRIN index of pagesPerRange pages per range, set the element
150 : * corresponding to heap block number heapBlk to the given TID.
151 : *
152 : * Once the operation is complete, the caller must update the LSN on the
153 : * returned buffer.
154 : *
155 : * This is used both in regular operation and during WAL replay.
156 : */
157 : void
158 3306 : brinSetHeapBlockItemptr(Buffer buf, BlockNumber pagesPerRange,
159 : BlockNumber heapBlk, ItemPointerData tid)
160 : {
161 : RevmapContents *contents;
162 : ItemPointerData *iptr;
163 : Page page;
164 :
165 : /* The correct page should already be pinned and locked */
2545 kgrittn 166 3306 : page = BufferGetPage(buf);
3075 alvherre 167 3306 : contents = (RevmapContents *) PageGetContents(page);
168 3306 : iptr = (ItemPointerData *) contents->rm_tids;
169 3306 : iptr += HEAPBLK_TO_REVMAP_INDEX(pagesPerRange, heapBlk);
170 :
2199 171 3306 : if (ItemPointerIsValid(&tid))
172 3278 : ItemPointerSet(iptr,
173 : ItemPointerGetBlockNumber(&tid),
174 3278 : ItemPointerGetOffsetNumber(&tid));
175 : else
176 28 : ItemPointerSetInvalid(iptr);
3075 177 3306 : }
178 :
179 : /*
180 : * Fetch the BrinTuple for a given heap block.
181 : *
182 : * The buffer containing the tuple is locked, and returned in *buf. The
183 : * returned tuple points to the shared buffer and must not be freed; if caller
184 : * wants to use it after releasing the buffer lock, it must create its own
185 : * palloc'ed copy. As an optimization, the caller can pass a pinned buffer
186 : * *buf on entry, which will avoid a pin-unpin cycle when the next tuple is on
187 : * the same page as a previous one.
188 : *
189 : * If no tuple is found for the given heap range, returns NULL. In that case,
190 : * *buf might still be updated (and pin must be released by caller), but it's
191 : * not locked.
192 : *
193 : * The output tuple offset within the buffer is returned in *off, and its size
194 : * is returned in *size.
195 : */
196 : BrinTuple *
197 142253 : brinGetTupleForHeapBlock(BrinRevmap *revmap, BlockNumber heapBlk,
198 : Buffer *buf, OffsetNumber *off, Size *size, int mode,
199 : Snapshot snapshot)
200 : {
201 142253 : Relation idxRel = revmap->rm_irel;
202 : BlockNumber mapBlk;
203 : RevmapContents *contents;
204 : ItemPointerData *iptr;
205 : BlockNumber blk;
206 : Page page;
207 : ItemId lp;
208 : BrinTuple *tup;
209 : ItemPointerData previptr;
210 :
211 : /* normalize the heap block number to be the first page in the range */
212 142253 : heapBlk = (heapBlk / revmap->rm_pagesPerRange) * revmap->rm_pagesPerRange;
213 :
214 : /*
215 : * Compute the revmap page number we need. If Invalid is returned (i.e.,
216 : * the revmap page hasn't been created yet), the requested page range is
217 : * not summarized.
218 : */
219 142253 : mapBlk = revmap_get_blkno(revmap, heapBlk);
220 142253 : if (mapBlk == InvalidBlockNumber)
221 : {
222 941 : *off = InvalidOffsetNumber;
223 941 : return NULL;
224 : }
225 :
226 141312 : ItemPointerSetInvalid(&previptr);
227 : for (;;)
228 : {
229 141312 : CHECK_FOR_INTERRUPTS();
230 :
231 243221 : if (revmap->rm_currBuf == InvalidBuffer ||
232 101909 : BufferGetBlockNumber(revmap->rm_currBuf) != mapBlk)
233 : {
234 39403 : if (revmap->rm_currBuf != InvalidBuffer)
3075 alvherre 235 UBC 0 : ReleaseBuffer(revmap->rm_currBuf);
236 :
3075 alvherre 237 CBC 39403 : Assert(mapBlk != InvalidBlockNumber);
238 39403 : revmap->rm_currBuf = ReadBuffer(revmap->rm_irel, mapBlk);
239 : }
240 :
241 141312 : LockBuffer(revmap->rm_currBuf, BUFFER_LOCK_SHARE);
242 :
243 : contents = (RevmapContents *)
2545 kgrittn 244 141312 : PageGetContents(BufferGetPage(revmap->rm_currBuf));
3075 alvherre 245 141312 : iptr = contents->rm_tids;
246 141312 : iptr += HEAPBLK_TO_REVMAP_INDEX(revmap->rm_pagesPerRange, heapBlk);
247 :
248 141312 : if (!ItemPointerIsValid(iptr))
249 : {
250 31399 : LockBuffer(revmap->rm_currBuf, BUFFER_LOCK_UNLOCK);
251 31399 : return NULL;
252 : }
253 :
254 : /*
255 : * Check the TID we got in a previous iteration, if any, and save the
256 : * current TID we got from the revmap; if we loop, we can sanity-check
257 : * that the next one we get is different. Otherwise we might be stuck
258 : * looping forever if the revmap is somehow badly broken.
259 : */
260 109913 : if (ItemPointerIsValid(&previptr) && ItemPointerEquals(&previptr, iptr))
3075 alvherre 261 UBC 0 : ereport(ERROR,
262 : (errcode(ERRCODE_INDEX_CORRUPTED),
263 : errmsg_internal("corrupted BRIN index: inconsistent range map")));
3075 alvherre 264 CBC 109913 : previptr = *iptr;
265 :
266 109913 : blk = ItemPointerGetBlockNumber(iptr);
267 109913 : *off = ItemPointerGetOffsetNumber(iptr);
268 :
269 109913 : LockBuffer(revmap->rm_currBuf, BUFFER_LOCK_UNLOCK);
270 :
271 : /* Ok, got a pointer to where the BrinTuple should be. Fetch it. */
272 109913 : if (!BufferIsValid(*buf) || BufferGetBlockNumber(*buf) != blk)
273 : {
274 18250 : if (BufferIsValid(*buf))
275 8794 : ReleaseBuffer(*buf);
276 18250 : *buf = ReadBuffer(idxRel, blk);
277 : }
278 109913 : LockBuffer(*buf, mode);
2545 kgrittn 279 109913 : page = BufferGetPage(*buf);
280 109913 : TestForOldSnapshot(snapshot, idxRel, page);
281 :
282 : /* If we land on a revmap page, start over */
3075 alvherre 283 109913 : if (BRIN_IS_REGULAR_PAGE(page))
284 : {
285 : /*
286 : * If the offset number is greater than what's in the page, it's
287 : * possible that the range was desummarized concurrently. Just
288 : * return NULL to handle that case.
289 : */
2505 290 109913 : if (*off > PageGetMaxOffsetNumber(page))
291 : {
970 alvherre 292 UBC 0 : LockBuffer(*buf, BUFFER_LOCK_UNLOCK);
293 0 : return NULL;
294 : }
295 :
3075 alvherre 296 CBC 109913 : lp = PageGetItemId(page, *off);
297 109913 : if (ItemIdIsUsed(lp))
298 : {
299 109913 : tup = (BrinTuple *) PageGetItem(page, lp);
300 :
301 109913 : if (tup->bt_blkno == heapBlk)
302 : {
303 109913 : if (size)
304 93927 : *size = ItemIdGetLength(lp);
305 : /* found it! */
306 109913 : return tup;
307 : }
308 : }
309 : }
310 :
311 : /*
312 : * No luck. Assume that the revmap was updated concurrently.
313 : */
3075 alvherre 314 UBC 0 : LockBuffer(*buf, BUFFER_LOCK_UNLOCK);
315 : }
316 : /* not reached, but keep compiler quiet */
317 : return NULL;
318 : }
319 :
320 : /*
321 : * Delete an index tuple, marking a page range as unsummarized.
322 : *
323 : * Index must be locked in ShareUpdateExclusiveLock mode.
324 : *
325 : * Return false if caller should retry.
326 : */
327 : bool
2199 alvherre 328 CBC 42 : brinRevmapDesummarizeRange(Relation idxrel, BlockNumber heapBlk)
329 : {
330 : BrinRevmap *revmap;
331 : BlockNumber pagesPerRange;
332 : RevmapContents *contents;
333 : ItemPointerData *iptr;
334 : ItemPointerData invalidIptr;
335 : BlockNumber revmapBlk;
336 : Buffer revmapBuf;
337 : Buffer regBuf;
338 : Page revmapPg;
339 : Page regPg;
340 : OffsetNumber revmapOffset;
341 : OffsetNumber regOffset;
342 : ItemId lp;
343 :
344 42 : revmap = brinRevmapInitialize(idxrel, &pagesPerRange, NULL);
345 :
346 42 : revmapBlk = revmap_get_blkno(revmap, heapBlk);
347 42 : if (!BlockNumberIsValid(revmapBlk))
348 : {
349 : /* revmap page doesn't exist: range not summarized, we're done */
350 9 : brinRevmapTerminate(revmap);
351 9 : return true;
352 : }
353 :
354 : /* Lock the revmap page, obtain the index tuple pointer from it */
355 33 : revmapBuf = brinLockRevmapPageForUpdate(revmap, heapBlk);
356 33 : revmapPg = BufferGetPage(revmapBuf);
357 33 : revmapOffset = HEAPBLK_TO_REVMAP_INDEX(revmap->rm_pagesPerRange, heapBlk);
358 :
359 33 : contents = (RevmapContents *) PageGetContents(revmapPg);
360 33 : iptr = contents->rm_tids;
361 33 : iptr += revmapOffset;
362 :
363 33 : if (!ItemPointerIsValid(iptr))
364 : {
365 : /* no index tuple: range not summarized, we're done */
366 12 : LockBuffer(revmapBuf, BUFFER_LOCK_UNLOCK);
367 12 : brinRevmapTerminate(revmap);
368 12 : return true;
369 : }
370 :
371 21 : regBuf = ReadBuffer(idxrel, ItemPointerGetBlockNumber(iptr));
372 21 : LockBuffer(regBuf, BUFFER_LOCK_EXCLUSIVE);
373 21 : regPg = BufferGetPage(regBuf);
374 :
375 : /*
376 : * We're only removing data, not reading it, so there's no need to
377 : * TestForOldSnapshot here.
378 : */
379 :
380 : /* if this is no longer a regular page, tell caller to start over */
381 21 : if (!BRIN_IS_REGULAR_PAGE(regPg))
382 : {
2199 alvherre 383 UBC 0 : LockBuffer(revmapBuf, BUFFER_LOCK_UNLOCK);
384 0 : LockBuffer(regBuf, BUFFER_LOCK_UNLOCK);
385 0 : brinRevmapTerminate(revmap);
386 0 : return false;
387 : }
388 :
2199 alvherre 389 CBC 21 : regOffset = ItemPointerGetOffsetNumber(iptr);
390 21 : if (regOffset > PageGetMaxOffsetNumber(regPg))
2199 alvherre 391 UBC 0 : ereport(ERROR,
392 : (errcode(ERRCODE_INDEX_CORRUPTED),
393 : errmsg("corrupted BRIN index: inconsistent range map")));
394 :
2199 alvherre 395 CBC 21 : lp = PageGetItemId(regPg, regOffset);
396 21 : if (!ItemIdIsUsed(lp))
2199 alvherre 397 UBC 0 : ereport(ERROR,
398 : (errcode(ERRCODE_INDEX_CORRUPTED),
399 : errmsg("corrupted BRIN index: inconsistent range map")));
400 :
401 : /*
402 : * Placeholder tuples only appear during unfinished summarization, and we
403 : * hold ShareUpdateExclusiveLock, so this function cannot run concurrently
404 : * with that. So any placeholder tuples that exist are leftovers from a
405 : * crashed or aborted summarization; remove them silently.
406 : */
407 :
2199 alvherre 408 CBC 21 : START_CRIT_SECTION();
409 :
410 21 : ItemPointerSetInvalid(&invalidIptr);
411 21 : brinSetHeapBlockItemptr(revmapBuf, revmap->rm_pagesPerRange, heapBlk,
412 : invalidIptr);
413 21 : PageIndexTupleDeleteNoCompact(regPg, regOffset);
414 : /* XXX record free space in FSM? */
415 :
416 21 : MarkBufferDirty(regBuf);
417 21 : MarkBufferDirty(revmapBuf);
418 :
419 21 : if (RelationNeedsWAL(idxrel))
420 : {
421 : xl_brin_desummarize xlrec;
422 : XLogRecPtr recptr;
423 :
2193 424 21 : xlrec.pagesPerRange = revmap->rm_pagesPerRange;
2199 425 21 : xlrec.heapBlk = heapBlk;
426 21 : xlrec.regOffset = regOffset;
427 :
428 21 : XLogBeginInsert();
429 21 : XLogRegisterData((char *) &xlrec, SizeOfBrinDesummarize);
430 21 : XLogRegisterBuffer(0, revmapBuf, 0);
431 21 : XLogRegisterBuffer(1, regBuf, REGBUF_STANDARD);
432 21 : recptr = XLogInsert(RM_BRIN_ID, XLOG_BRIN_DESUMMARIZE);
433 21 : PageSetLSN(revmapPg, recptr);
434 21 : PageSetLSN(regPg, recptr);
435 : }
436 :
437 21 : END_CRIT_SECTION();
438 :
439 21 : UnlockReleaseBuffer(regBuf);
440 21 : LockBuffer(revmapBuf, BUFFER_LOCK_UNLOCK);
441 21 : brinRevmapTerminate(revmap);
442 :
443 21 : return true;
444 : }
445 :
446 : /*
447 : * Given a heap block number, find the corresponding physical revmap block
448 : * number and return it. If the revmap page hasn't been allocated yet, return
449 : * InvalidBlockNumber.
450 : */
451 : static BlockNumber
3075 452 145226 : revmap_get_blkno(BrinRevmap *revmap, BlockNumber heapBlk)
453 : {
454 : BlockNumber targetblk;
455 :
456 : /* obtain revmap block number, skip 1 for metapage block */
457 145226 : targetblk = HEAPBLK_TO_REVMAP_BLK(revmap->rm_pagesPerRange, heapBlk) + 1;
458 :
459 : /* Normal case: the revmap page is already allocated */
460 145226 : if (targetblk <= revmap->rm_lastRevmapPage)
461 144276 : return targetblk;
462 :
463 950 : return InvalidBlockNumber;
464 : }
465 :
466 : /*
467 : * Obtain and return a buffer containing the revmap page for the given heap
468 : * page. The revmap must have been previously extended to cover that page.
469 : * The returned buffer is also recorded in the revmap struct; finishing that
470 : * releases the buffer, therefore the caller needn't do it explicitly.
471 : */
472 : static Buffer
473 2931 : revmap_get_buffer(BrinRevmap *revmap, BlockNumber heapBlk)
474 : {
475 : BlockNumber mapBlk;
476 :
477 : /* Translate the heap block number to physical index location. */
478 2931 : mapBlk = revmap_get_blkno(revmap, heapBlk);
479 :
480 2931 : if (mapBlk == InvalidBlockNumber)
3075 alvherre 481 UBC 0 : elog(ERROR, "revmap does not cover heap block %u", heapBlk);
482 :
483 : /* Ensure the buffer we got is in the expected range */
3075 alvherre 484 CBC 2931 : Assert(mapBlk != BRIN_METAPAGE_BLKNO &&
485 : mapBlk <= revmap->rm_lastRevmapPage);
486 :
487 : /*
488 : * Obtain the buffer from which we need to read. If we already have the
489 : * correct buffer in our access struct, use that; otherwise, release that,
490 : * (if valid) and read the one we need.
491 : */
492 5710 : if (revmap->rm_currBuf == InvalidBuffer ||
493 2779 : mapBlk != BufferGetBlockNumber(revmap->rm_currBuf))
494 : {
495 153 : if (revmap->rm_currBuf != InvalidBuffer)
496 1 : ReleaseBuffer(revmap->rm_currBuf);
497 :
498 153 : revmap->rm_currBuf = ReadBuffer(revmap->rm_irel, mapBlk);
499 : }
500 :
501 2931 : return revmap->rm_currBuf;
502 : }
503 :
504 : /*
505 : * Given a heap block number, find the corresponding physical revmap block
506 : * number and return it. If the revmap page hasn't been allocated yet, extend
507 : * the revmap until it is.
508 : */
509 : static BlockNumber
510 6027 : revmap_extend_and_get_blkno(BrinRevmap *revmap, BlockNumber heapBlk)
511 : {
512 : BlockNumber targetblk;
513 :
514 : /* obtain revmap block number, skip 1 for metapage block */
515 6027 : targetblk = HEAPBLK_TO_REVMAP_BLK(revmap->rm_pagesPerRange, heapBlk) + 1;
516 :
517 : /* Extend the revmap, if necessary */
518 6268 : while (targetblk > revmap->rm_lastRevmapPage)
519 : {
520 241 : CHECK_FOR_INTERRUPTS();
521 241 : revmap_physical_extend(revmap);
522 : }
523 :
524 6027 : return targetblk;
525 : }
526 :
527 : /*
528 : * Try to extend the revmap by one page. This might not happen for a number of
529 : * reasons; caller is expected to retry until the expected outcome is obtained.
530 : */
531 : static void
532 241 : revmap_physical_extend(BrinRevmap *revmap)
533 : {
534 : Buffer buf;
535 : Page page;
536 : Page metapage;
537 : BrinMetaPageData *metadata;
538 : BlockNumber mapBlk;
539 : BlockNumber nblocks;
540 241 : Relation irel = revmap->rm_irel;
541 :
542 : /*
543 : * Lock the metapage. This locks out concurrent extensions of the revmap,
544 : * but note that we still need to grab the relation extension lock because
545 : * another backend can extend the index with regular BRIN pages.
3075 alvherre 546 ECB : */
3075 alvherre 547 CBC 241 : LockBuffer(revmap->rm_metaBuf, BUFFER_LOCK_EXCLUSIVE);
2545 kgrittn 548 241 : metapage = BufferGetPage(revmap->rm_metaBuf);
3075 alvherre 549 GIC 241 : metadata = (BrinMetaPageData *) PageGetContents(metapage);
550 :
551 : /*
552 : * Check that our cached lastRevmapPage value was up-to-date; if it
553 : * wasn't, update the cached copy and have caller start over.
3075 alvherre 554 ECB : */
3075 alvherre 555 GIC 241 : if (metadata->lastRevmapPage != revmap->rm_lastRevmapPage)
3075 alvherre 556 ECB : {
3075 alvherre 557 CBC 120 : revmap->rm_lastRevmapPage = metadata->lastRevmapPage;
558 120 : LockBuffer(revmap->rm_metaBuf, BUFFER_LOCK_UNLOCK);
3075 alvherre 559 GIC 120 : return;
3075 alvherre 560 ECB : }
3075 alvherre 561 GIC 121 : mapBlk = metadata->lastRevmapPage + 1;
3075 alvherre 562 ECB :
3075 alvherre 563 CBC 121 : nblocks = RelationGetNumberOfBlocks(irel);
3075 alvherre 564 GIC 121 : if (mapBlk < nblocks)
3075 alvherre 565 ECB : {
3075 alvherre 566 CBC 2 : buf = ReadBuffer(irel, mapBlk);
567 2 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
2545 kgrittn 568 GIC 2 : page = BufferGetPage(buf);
569 : }
570 : else
3075 alvherre 571 ECB : {
4 andres 572 GNC 119 : buf = ExtendBufferedRel(EB_REL(irel), MAIN_FORKNUM, NULL,
573 : EB_LOCK_FIRST);
3075 alvherre 574 GIC 119 : if (BufferGetBlockNumber(buf) != mapBlk)
575 : {
576 : /*
577 : * Very rare corner case: somebody extended the relation
578 : * concurrently after we read its length. If this happens, give
3075 alvherre 579 EUB : * up and have caller start over. We will have to evacuate that
580 : * page from under whoever is using it.
581 : */
3075 alvherre 582 UIC 0 : LockBuffer(revmap->rm_metaBuf, BUFFER_LOCK_UNLOCK);
4 andres 583 UNC 0 : UnlockReleaseBuffer(buf);
3075 alvherre 584 UIC 0 : return;
3075 alvherre 585 ECB : }
2545 kgrittn 586 GIC 119 : page = BufferGetPage(buf);
587 : }
588 :
589 : /* Check that it's a regular block (or an empty page) */
3075 alvherre 590 CBC 121 : if (!PageIsNew(page) && !BRIN_IS_REGULAR_PAGE(page))
3075 alvherre 591 UIC 0 : ereport(ERROR,
3075 alvherre 592 ECB : (errcode(ERRCODE_INDEX_CORRUPTED),
2118 tgl 593 : errmsg("unexpected page type 0x%04X in BRIN index \"%s\" block %u",
594 : BrinPageType(page),
595 : RelationGetRelationName(irel),
596 : BufferGetBlockNumber(buf))));
597 :
598 : /* If the page is in use, evacuate it and restart */
3075 alvherre 599 GIC 121 : if (brin_start_evacuating_page(irel, buf))
600 : {
601 1 : LockBuffer(revmap->rm_metaBuf, BUFFER_LOCK_UNLOCK);
602 1 : brin_evacuate_page(irel, revmap->rm_pagesPerRange, revmap, buf);
3075 alvherre 603 ECB :
604 : /* have caller start over */
3075 alvherre 605 GIC 1 : return;
3075 alvherre 606 ECB : }
607 :
608 : /*
609 : * Ok, we have now locked the metapage and the target block. Re-initialize
610 : * the target block as a revmap page, and update the metapage.
611 : */
3075 alvherre 612 GIC 120 : START_CRIT_SECTION();
613 :
614 : /* the rm_tids array is initialized to all invalid by PageInit */
615 120 : brin_page_init(page, BRIN_PAGETYPE_REVMAP);
616 120 : MarkBufferDirty(buf);
617 :
3075 alvherre 618 CBC 120 : metadata->lastRevmapPage = mapBlk;
1984 tgl 619 ECB :
620 : /*
621 : * Set pd_lower just past the end of the metadata. This is essential,
622 : * because without doing so, metadata will be lost if xlog.c compresses
623 : * the page. (We must do this here because pre-v11 versions of PG did not
624 : * set the metapage's pd_lower correctly, so a pg_upgraded index might
625 : * contain the wrong value.)
626 : */
1984 tgl 627 GIC 120 : ((PageHeader) metapage)->pd_lower =
1984 tgl 628 CBC 120 : ((char *) metadata + sizeof(BrinMetaPageData)) - (char *) metapage;
629 :
3075 alvherre 630 120 : MarkBufferDirty(revmap->rm_metaBuf);
3075 alvherre 631 ECB :
3075 alvherre 632 CBC 120 : if (RelationNeedsWAL(revmap->rm_irel))
633 : {
3075 alvherre 634 ECB : xl_brin_revmap_extend xlrec;
635 : XLogRecPtr recptr;
636 :
3075 alvherre 637 CBC 52 : xlrec.targetBlk = mapBlk;
3062 heikki.linnakangas 638 ECB :
3062 heikki.linnakangas 639 GIC 52 : XLogBeginInsert();
640 52 : XLogRegisterData((char *) &xlrec, SizeOfBrinRevmapExtend);
1984 tgl 641 CBC 52 : XLogRegisterBuffer(0, revmap->rm_metaBuf, REGBUF_STANDARD);
642 :
3062 heikki.linnakangas 643 52 : XLogRegisterBuffer(1, buf, REGBUF_WILL_INIT);
644 :
645 52 : recptr = XLogInsert(RM_BRIN_ID, XLOG_BRIN_REVMAP_EXTEND);
3075 alvherre 646 GIC 52 : PageSetLSN(metapage, recptr);
647 52 : PageSetLSN(page, recptr);
648 : }
649 :
650 120 : END_CRIT_SECTION();
651 :
652 120 : LockBuffer(revmap->rm_metaBuf, BUFFER_LOCK_UNLOCK);
653 :
654 120 : UnlockReleaseBuffer(buf);
655 : }
|
