Line data Source code
1 : /*
2 : * brin_pageops.c
3 : * Page-handling routines for BRIN indexes
4 : *
5 : * Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
6 : * Portions Copyright (c) 1994, Regents of the University of California
7 : *
8 : * IDENTIFICATION
9 : * src/backend/access/brin/brin_pageops.c
10 : */
11 : #include "postgres.h"
12 :
13 : #include "access/brin_page.h"
14 : #include "access/brin_pageops.h"
15 : #include "access/brin_revmap.h"
16 : #include "access/brin_xlog.h"
17 : #include "access/xloginsert.h"
18 : #include "miscadmin.h"
19 : #include "storage/bufmgr.h"
20 : #include "storage/freespace.h"
21 : #include "storage/lmgr.h"
22 : #include "utils/rel.h"
23 :
24 : /*
25 : * Maximum size of an entry in a BRIN_PAGETYPE_REGULAR page. We can tolerate
26 : * a single item per page, unlike other index AMs.
27 : */
28 : #define BrinMaxItemSize \
29 : MAXALIGN_DOWN(BLCKSZ - \
30 : (MAXALIGN(SizeOfPageHeaderData + \
31 : sizeof(ItemIdData)) + \
32 : MAXALIGN(sizeof(BrinSpecialSpace))))
33 :
34 : static Buffer brin_getinsertbuffer(Relation irel, Buffer oldbuf, Size itemsz,
35 : bool *extended);
36 : static Size br_page_get_freespace(Page page);
37 : static void brin_initialize_empty_new_buffer(Relation idxrel, Buffer buffer);
38 :
39 :
40 : /*
41 : * Update tuple origtup (size origsz), located in offset oldoff of buffer
42 : * oldbuf, to newtup (size newsz) as summary tuple for the page range starting
43 : * at heapBlk. oldbuf must not be locked on entry, and is not locked at exit.
44 : *
45 : * If samepage is true, attempt to put the new tuple in the same page, but if
46 : * there's no room, use some other one.
47 : *
48 : * If the update is successful, return true; the revmap is updated to point to
49 : * the new tuple. If the update is not done for whatever reason, return false.
50 : * Caller may retry the update if this happens.
51 : */
52 : bool
53 27424 : brin_doupdate(Relation idxrel, BlockNumber pagesPerRange,
54 : BrinRevmap *revmap, BlockNumber heapBlk,
55 : Buffer oldbuf, OffsetNumber oldoff,
56 : const BrinTuple *origtup, Size origsz,
57 : const BrinTuple *newtup, Size newsz,
58 : bool samepage)
59 : {
60 : Page oldpage;
61 : ItemId oldlp;
62 : BrinTuple *oldtup;
63 : Size oldsz;
64 : Buffer newbuf;
65 27424 : BlockNumber newblk = InvalidBlockNumber;
66 : bool extended;
67 :
68 : Assert(newsz == MAXALIGN(newsz));
69 :
70 : /* If the item is oversized, don't bother. */
71 27424 : if (newsz > BrinMaxItemSize)
72 : {
73 0 : ereport(ERROR,
74 : (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
75 : errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
76 : newsz, BrinMaxItemSize, RelationGetRelationName(idxrel))));
77 : return false; /* keep compiler quiet */
78 : }
79 :
80 : /* make sure the revmap is long enough to contain the entry we need */
81 27424 : brinRevmapExtend(revmap, heapBlk);
82 :
83 27424 : if (!samepage)
84 : {
85 : /* need a page on which to put the item */
86 614 : newbuf = brin_getinsertbuffer(idxrel, oldbuf, newsz, &extended);
87 614 : if (!BufferIsValid(newbuf))
88 : {
89 : Assert(!extended);
90 0 : return false;
91 : }
92 :
93 : /*
94 : * Note: it's possible (though unlikely) that the returned newbuf is
95 : * the same as oldbuf, if brin_getinsertbuffer determined that the old
96 : * buffer does in fact have enough space.
97 : */
98 614 : if (newbuf == oldbuf)
99 : {
100 : Assert(!extended);
101 0 : newbuf = InvalidBuffer;
102 : }
103 : else
104 614 : newblk = BufferGetBlockNumber(newbuf);
105 : }
106 : else
107 : {
108 26810 : LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);
109 26810 : newbuf = InvalidBuffer;
110 26810 : extended = false;
111 : }
112 27424 : oldpage = BufferGetPage(oldbuf);
113 27424 : oldlp = PageGetItemId(oldpage, oldoff);
114 :
115 : /*
116 : * Check that the old tuple wasn't updated concurrently: it might have
117 : * moved someplace else entirely, and for that matter the whole page
118 : * might've become a revmap page. Note that in the first two cases
119 : * checked here, the "oldlp" we just calculated is garbage; but
120 : * PageGetItemId() is simple enough that it was safe to do that
121 : * calculation anyway.
122 : */
123 54848 : if (!BRIN_IS_REGULAR_PAGE(oldpage) ||
124 27424 : oldoff > PageGetMaxOffsetNumber(oldpage) ||
125 27424 : !ItemIdIsNormal(oldlp))
126 : {
127 0 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
128 :
129 : /*
130 : * If this happens, and the new buffer was obtained by extending the
131 : * relation, then we need to ensure we don't leave it uninitialized or
132 : * forget about it.
133 : */
134 0 : if (BufferIsValid(newbuf))
135 : {
136 0 : if (extended)
137 0 : brin_initialize_empty_new_buffer(idxrel, newbuf);
138 0 : UnlockReleaseBuffer(newbuf);
139 0 : if (extended)
140 0 : FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);
141 : }
142 0 : return false;
143 : }
144 :
145 27424 : oldsz = ItemIdGetLength(oldlp);
146 27424 : oldtup = (BrinTuple *) PageGetItem(oldpage, oldlp);
147 :
148 : /*
149 : * ... or it might have been updated in place to different contents.
150 : */
151 27424 : if (!brin_tuples_equal(oldtup, oldsz, origtup, origsz))
152 : {
153 0 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
154 0 : if (BufferIsValid(newbuf))
155 : {
156 : /* As above, initialize and record new page if we got one */
157 0 : if (extended)
158 0 : brin_initialize_empty_new_buffer(idxrel, newbuf);
159 0 : UnlockReleaseBuffer(newbuf);
160 0 : if (extended)
161 0 : FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);
162 : }
163 0 : return false;
164 : }
165 :
166 : /*
167 : * Great, the old tuple is intact. We can proceed with the update.
168 : *
169 : * If there's enough room in the old page for the new tuple, replace it.
170 : *
171 : * Note that there might now be enough space on the page even though the
172 : * caller told us there isn't, if a concurrent update moved another tuple
173 : * elsewhere or replaced a tuple with a smaller one.
174 : */
175 54266 : if (((BrinPageFlags(oldpage) & BRIN_EVACUATE_PAGE) == 0) &&
176 26842 : brin_can_do_samepage_update(oldbuf, origsz, newsz))
177 : {
178 26810 : START_CRIT_SECTION();
179 26810 : if (!PageIndexTupleOverwrite(oldpage, oldoff, (Item) unconstify(BrinTuple *, newtup), newsz))
180 0 : elog(ERROR, "failed to replace BRIN tuple");
181 26810 : MarkBufferDirty(oldbuf);
182 :
183 : /* XLOG stuff */
184 26810 : if (RelationNeedsWAL(idxrel))
185 : {
186 : xl_brin_samepage_update xlrec;
187 : XLogRecPtr recptr;
188 26804 : uint8 info = XLOG_BRIN_SAMEPAGE_UPDATE;
189 :
190 26804 : xlrec.offnum = oldoff;
191 :
192 26804 : XLogBeginInsert();
193 26804 : XLogRegisterData((char *) &xlrec, SizeOfBrinSamepageUpdate);
194 :
195 26804 : XLogRegisterBuffer(0, oldbuf, REGBUF_STANDARD);
196 26804 : XLogRegisterBufData(0, (char *) unconstify(BrinTuple *, newtup), newsz);
197 :
198 26804 : recptr = XLogInsert(RM_BRIN_ID, info);
199 :
200 26804 : PageSetLSN(oldpage, recptr);
201 : }
202 :
203 26810 : END_CRIT_SECTION();
204 :
205 26810 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
206 :
207 26810 : if (BufferIsValid(newbuf))
208 : {
209 : /* As above, initialize and record new page if we got one */
210 0 : if (extended)
211 0 : brin_initialize_empty_new_buffer(idxrel, newbuf);
212 0 : UnlockReleaseBuffer(newbuf);
213 0 : if (extended)
214 0 : FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);
215 : }
216 :
217 26810 : return true;
218 : }
219 614 : else if (newbuf == InvalidBuffer)
220 : {
221 : /*
222 : * Not enough space, but caller said that there was. Tell them to
223 : * start over.
224 : */
225 0 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
226 0 : return false;
227 : }
228 : else
229 : {
230 : /*
231 : * Not enough free space on the oldpage. Put the new tuple on the new
232 : * page, and update the revmap.
233 : */
234 614 : Page newpage = BufferGetPage(newbuf);
235 : Buffer revmapbuf;
236 : ItemPointerData newtid;
237 : OffsetNumber newoff;
238 614 : Size freespace = 0;
239 :
240 614 : revmapbuf = brinLockRevmapPageForUpdate(revmap, heapBlk);
241 :
242 614 : START_CRIT_SECTION();
243 :
244 : /*
245 : * We need to initialize the page if it's newly obtained. Note we
246 : * will WAL-log the initialization as part of the update, so we don't
247 : * need to do that here.
248 : */
249 614 : if (extended)
250 22 : brin_page_init(newpage, BRIN_PAGETYPE_REGULAR);
251 :
252 614 : PageIndexTupleDeleteNoCompact(oldpage, oldoff);
253 614 : newoff = PageAddItem(newpage, (Item) unconstify(BrinTuple *, newtup), newsz,
254 : InvalidOffsetNumber, false, false);
255 614 : if (newoff == InvalidOffsetNumber)
256 0 : elog(ERROR, "failed to add BRIN tuple to new page");
257 614 : MarkBufferDirty(oldbuf);
258 614 : MarkBufferDirty(newbuf);
259 :
260 : /* needed to update FSM below */
261 614 : if (extended)
262 22 : freespace = br_page_get_freespace(newpage);
263 :
264 614 : ItemPointerSet(&newtid, newblk, newoff);
265 614 : brinSetHeapBlockItemptr(revmapbuf, pagesPerRange, heapBlk, newtid);
266 614 : MarkBufferDirty(revmapbuf);
267 :
268 : /* XLOG stuff */
269 614 : if (RelationNeedsWAL(idxrel))
270 : {
271 : xl_brin_update xlrec;
272 : XLogRecPtr recptr;
273 : uint8 info;
274 :
275 614 : info = XLOG_BRIN_UPDATE | (extended ? XLOG_BRIN_INIT_PAGE : 0);
276 :
277 614 : xlrec.insert.offnum = newoff;
278 614 : xlrec.insert.heapBlk = heapBlk;
279 614 : xlrec.insert.pagesPerRange = pagesPerRange;
280 614 : xlrec.oldOffnum = oldoff;
281 :
282 614 : XLogBeginInsert();
283 :
284 : /* new page */
285 614 : XLogRegisterData((char *) &xlrec, SizeOfBrinUpdate);
286 :
287 614 : XLogRegisterBuffer(0, newbuf, REGBUF_STANDARD | (extended ? REGBUF_WILL_INIT : 0));
288 614 : XLogRegisterBufData(0, (char *) unconstify(BrinTuple *, newtup), newsz);
289 :
290 : /* revmap page */
291 614 : XLogRegisterBuffer(1, revmapbuf, 0);
292 :
293 : /* old page */
294 614 : XLogRegisterBuffer(2, oldbuf, REGBUF_STANDARD);
295 :
296 614 : recptr = XLogInsert(RM_BRIN_ID, info);
297 :
298 614 : PageSetLSN(oldpage, recptr);
299 614 : PageSetLSN(newpage, recptr);
300 614 : PageSetLSN(BufferGetPage(revmapbuf), recptr);
301 : }
302 :
303 614 : END_CRIT_SECTION();
304 :
305 614 : LockBuffer(revmapbuf, BUFFER_LOCK_UNLOCK);
306 614 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
307 614 : UnlockReleaseBuffer(newbuf);
308 :
309 614 : if (extended)
310 : {
311 22 : RecordPageWithFreeSpace(idxrel, newblk, freespace);
312 22 : FreeSpaceMapVacuumRange(idxrel, newblk, newblk + 1);
313 : }
314 :
315 614 : return true;
316 : }
317 : }
318 :
319 : /*
320 : * Return whether brin_doupdate can do a samepage update.
321 : */
322 : bool
323 53684 : brin_can_do_samepage_update(Buffer buffer, Size origsz, Size newsz)
324 : {
325 : return
326 61560 : ((newsz <= origsz) ||
327 7876 : PageGetExactFreeSpace(BufferGetPage(buffer)) >= (newsz - origsz));
328 : }
329 :
330 : /*
331 : * Insert an index tuple into the index relation. The revmap is updated to
332 : * mark the range containing the given page as pointing to the inserted entry.
333 : * A WAL record is written.
334 : *
335 : * The buffer, if valid, is first checked for free space to insert the new
336 : * entry; if there isn't enough, a new buffer is obtained and pinned. No
337 : * buffer lock must be held on entry, no buffer lock is held on exit.
338 : *
339 : * Return value is the offset number where the tuple was inserted.
340 : */
341 : OffsetNumber
342 5578 : brin_doinsert(Relation idxrel, BlockNumber pagesPerRange,
343 : BrinRevmap *revmap, Buffer *buffer, BlockNumber heapBlk,
344 : BrinTuple *tup, Size itemsz)
345 : {
346 : Page page;
347 : BlockNumber blk;
348 : OffsetNumber off;
349 5578 : Size freespace = 0;
350 : Buffer revmapbuf;
351 : ItemPointerData tid;
352 : bool extended;
353 :
354 : Assert(itemsz == MAXALIGN(itemsz));
355 :
356 : /* If the item is oversized, don't even bother. */
357 5578 : if (itemsz > BrinMaxItemSize)
358 : {
359 0 : ereport(ERROR,
360 : (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
361 : errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
362 : itemsz, BrinMaxItemSize, RelationGetRelationName(idxrel))));
363 : return InvalidOffsetNumber; /* keep compiler quiet */
364 : }
365 :
366 : /* Make sure the revmap is long enough to contain the entry we need */
367 5578 : brinRevmapExtend(revmap, heapBlk);
368 :
369 : /*
370 : * Acquire lock on buffer supplied by caller, if any. If it doesn't have
371 : * enough space, unpin it to obtain a new one below.
372 : */
373 5578 : if (BufferIsValid(*buffer))
374 : {
375 : /*
376 : * It's possible that another backend (or ourselves!) extended the
377 : * revmap over the page we held a pin on, so we cannot assume that
378 : * it's still a regular page.
379 : */
380 2306 : LockBuffer(*buffer, BUFFER_LOCK_EXCLUSIVE);
381 2306 : if (br_page_get_freespace(BufferGetPage(*buffer)) < itemsz)
382 : {
383 114 : UnlockReleaseBuffer(*buffer);
384 114 : *buffer = InvalidBuffer;
385 : }
386 : }
387 :
388 : /*
389 : * If we still don't have a usable buffer, have brin_getinsertbuffer
390 : * obtain one for us.
391 : */
392 5578 : if (!BufferIsValid(*buffer))
393 : {
394 : do
395 3386 : *buffer = brin_getinsertbuffer(idxrel, InvalidBuffer, itemsz, &extended);
396 3386 : while (!BufferIsValid(*buffer));
397 : }
398 : else
399 2192 : extended = false;
400 :
401 : /* Now obtain lock on revmap buffer */
402 5578 : revmapbuf = brinLockRevmapPageForUpdate(revmap, heapBlk);
403 :
404 5578 : page = BufferGetPage(*buffer);
405 5578 : blk = BufferGetBlockNumber(*buffer);
406 :
407 : /* Execute the actual insertion */
408 5578 : START_CRIT_SECTION();
409 5578 : if (extended)
410 452 : brin_page_init(page, BRIN_PAGETYPE_REGULAR);
411 5578 : off = PageAddItem(page, (Item) tup, itemsz, InvalidOffsetNumber,
412 : false, false);
413 5578 : if (off == InvalidOffsetNumber)
414 0 : elog(ERROR, "failed to add BRIN tuple to new page");
415 5578 : MarkBufferDirty(*buffer);
416 :
417 : /* needed to update FSM below */
418 5578 : if (extended)
419 452 : freespace = br_page_get_freespace(page);
420 :
421 5578 : ItemPointerSet(&tid, blk, off);
422 5578 : brinSetHeapBlockItemptr(revmapbuf, pagesPerRange, heapBlk, tid);
423 5578 : MarkBufferDirty(revmapbuf);
424 :
425 : /* XLOG stuff */
426 5578 : if (RelationNeedsWAL(idxrel))
427 : {
428 : xl_brin_insert xlrec;
429 : XLogRecPtr recptr;
430 : uint8 info;
431 :
432 4680 : info = XLOG_BRIN_INSERT | (extended ? XLOG_BRIN_INIT_PAGE : 0);
433 4680 : xlrec.heapBlk = heapBlk;
434 4680 : xlrec.pagesPerRange = pagesPerRange;
435 4680 : xlrec.offnum = off;
436 :
437 4680 : XLogBeginInsert();
438 4680 : XLogRegisterData((char *) &xlrec, SizeOfBrinInsert);
439 :
440 4680 : XLogRegisterBuffer(0, *buffer, REGBUF_STANDARD | (extended ? REGBUF_WILL_INIT : 0));
441 4680 : XLogRegisterBufData(0, (char *) tup, itemsz);
442 :
443 4680 : XLogRegisterBuffer(1, revmapbuf, 0);
444 :
445 4680 : recptr = XLogInsert(RM_BRIN_ID, info);
446 :
447 4680 : PageSetLSN(page, recptr);
448 4680 : PageSetLSN(BufferGetPage(revmapbuf), recptr);
449 : }
450 :
451 5578 : END_CRIT_SECTION();
452 :
453 : /* Tuple is firmly on buffer; we can release our locks */
454 5578 : LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
455 5578 : LockBuffer(revmapbuf, BUFFER_LOCK_UNLOCK);
456 :
457 : BRIN_elog((DEBUG2, "inserted tuple (%u,%u) for range starting at %u",
458 : blk, off, heapBlk));
459 :
460 5578 : if (extended)
461 : {
462 452 : RecordPageWithFreeSpace(idxrel, blk, freespace);
463 452 : FreeSpaceMapVacuumRange(idxrel, blk, blk + 1);
464 : }
465 :
466 5578 : return off;
467 : }
468 :
469 : /*
470 : * Initialize a page with the given type.
471 : *
472 : * Caller is responsible for marking it dirty, as appropriate.
473 : */
474 : void
475 1422 : brin_page_init(Page page, uint16 type)
476 : {
477 1422 : PageInit(page, BLCKSZ, sizeof(BrinSpecialSpace));
478 :
479 1422 : BrinPageType(page) = type;
480 1422 : }
481 :
482 : /*
483 : * Initialize a new BRIN index's metapage.
484 : */
485 : void
486 418 : brin_metapage_init(Page page, BlockNumber pagesPerRange, uint16 version)
487 : {
488 : BrinMetaPageData *metadata;
489 :
490 418 : brin_page_init(page, BRIN_PAGETYPE_META);
491 :
492 418 : metadata = (BrinMetaPageData *) PageGetContents(page);
493 :
494 418 : metadata->brinMagic = BRIN_META_MAGIC;
495 418 : metadata->brinVersion = version;
496 418 : metadata->pagesPerRange = pagesPerRange;
497 :
498 : /*
499 : * Note we cheat here a little. 0 is not a valid revmap block number
500 : * (because it's the metapage buffer), but doing this enables the first
501 : * revmap page to be created when the index is.
502 : */
503 418 : metadata->lastRevmapPage = 0;
504 :
505 : /*
506 : * Set pd_lower just past the end of the metadata. This is essential,
507 : * because without doing so, metadata will be lost if xlog.c compresses
508 : * the page.
509 : */
510 418 : ((PageHeader) page)->pd_lower =
511 418 : ((char *) metadata + sizeof(BrinMetaPageData)) - (char *) page;
512 418 : }
513 :
514 : /*
515 : * Initiate page evacuation protocol.
516 : *
517 : * The page must be locked in exclusive mode by the caller.
518 : *
519 : * If the page is not yet initialized or empty, return false without doing
520 : * anything; it can be used for revmap without any further changes. If it
521 : * contains tuples, mark it for evacuation and return true.
522 : */
523 : bool
524 342 : brin_start_evacuating_page(Relation idxRel, Buffer buf)
525 : {
526 : OffsetNumber off;
527 : OffsetNumber maxoff;
528 : Page page;
529 :
530 342 : page = BufferGetPage(buf);
531 :
532 342 : if (PageIsNew(page))
533 338 : return false;
534 :
535 4 : maxoff = PageGetMaxOffsetNumber(page);
536 584 : for (off = FirstOffsetNumber; off <= maxoff; off++)
537 : {
538 : ItemId lp;
539 :
540 582 : lp = PageGetItemId(page, off);
541 582 : if (ItemIdIsUsed(lp))
542 : {
543 : /*
544 : * Prevent other backends from adding more stuff to this page:
545 : * BRIN_EVACUATE_PAGE informs br_page_get_freespace that this page
546 : * can no longer be used to add new tuples. Note that this flag
547 : * is not WAL-logged, except accidentally.
548 : */
549 2 : BrinPageFlags(page) |= BRIN_EVACUATE_PAGE;
550 2 : MarkBufferDirtyHint(buf, true);
551 :
552 2 : return true;
553 : }
554 : }
555 2 : return false;
556 : }
557 :
558 : /*
559 : * Move all tuples out of a page.
560 : *
561 : * The caller must hold lock on the page. The lock and pin are released.
562 : */
563 : void
564 2 : brin_evacuate_page(Relation idxRel, BlockNumber pagesPerRange,
565 : BrinRevmap *revmap, Buffer buf)
566 : {
567 : OffsetNumber off;
568 : OffsetNumber maxoff;
569 : Page page;
570 2 : BrinTuple *btup = NULL;
571 2 : Size btupsz = 0;
572 :
573 2 : page = BufferGetPage(buf);
574 :
575 : Assert(BrinPageFlags(page) & BRIN_EVACUATE_PAGE);
576 :
577 2 : maxoff = PageGetMaxOffsetNumber(page);
578 584 : for (off = FirstOffsetNumber; off <= maxoff; off++)
579 : {
580 : BrinTuple *tup;
581 : Size sz;
582 : ItemId lp;
583 :
584 582 : CHECK_FOR_INTERRUPTS();
585 :
586 582 : lp = PageGetItemId(page, off);
587 582 : if (ItemIdIsUsed(lp))
588 : {
589 582 : sz = ItemIdGetLength(lp);
590 582 : tup = (BrinTuple *) PageGetItem(page, lp);
591 582 : tup = brin_copy_tuple(tup, sz, btup, &btupsz);
592 :
593 582 : LockBuffer(buf, BUFFER_LOCK_UNLOCK);
594 :
595 582 : if (!brin_doupdate(idxRel, pagesPerRange, revmap, tup->bt_blkno,
596 : buf, off, tup, sz, tup, sz, false))
597 0 : off--; /* retry */
598 :
599 582 : LockBuffer(buf, BUFFER_LOCK_SHARE);
600 :
601 : /* It's possible that someone extended the revmap over this page */
602 582 : if (!BRIN_IS_REGULAR_PAGE(page))
603 0 : break;
604 : }
605 : }
606 :
607 2 : UnlockReleaseBuffer(buf);
608 2 : }
609 :
610 : /*
611 : * Given a BRIN index page, initialize it if necessary, and record its
612 : * current free space in the FSM.
613 : *
614 : * The main use for this is when, during vacuuming, an uninitialized page is
615 : * found, which could be the result of relation extension followed by a crash
616 : * before the page can be used.
617 : *
618 : * Here, we don't bother to update upper FSM pages, instead expecting that our
619 : * caller (brin_vacuum_scan) will fix them at the end of the scan. Elsewhere
620 : * in this file, it's generally a good idea to propagate additions of free
621 : * space into the upper FSM pages immediately.
622 : */
623 : void
624 372 : brin_page_cleanup(Relation idxrel, Buffer buf)
625 : {
626 372 : Page page = BufferGetPage(buf);
627 :
628 : /*
629 : * If a page was left uninitialized, initialize it now; also record it in
630 : * FSM.
631 : *
632 : * Somebody else might be extending the relation concurrently. To avoid
633 : * re-initializing the page before they can grab the buffer lock, we
634 : * acquire the extension lock momentarily. Since they hold the extension
635 : * lock from before getting the page and after its been initialized, we're
636 : * sure to see their initialization.
637 : */
638 372 : if (PageIsNew(page))
639 : {
640 0 : LockRelationForExtension(idxrel, ShareLock);
641 0 : UnlockRelationForExtension(idxrel, ShareLock);
642 :
643 0 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
644 0 : if (PageIsNew(page))
645 : {
646 0 : brin_initialize_empty_new_buffer(idxrel, buf);
647 0 : LockBuffer(buf, BUFFER_LOCK_UNLOCK);
648 0 : return;
649 : }
650 0 : LockBuffer(buf, BUFFER_LOCK_UNLOCK);
651 : }
652 :
653 : /* Nothing to be done for non-regular index pages */
654 372 : if (BRIN_IS_META_PAGE(BufferGetPage(buf)) ||
655 286 : BRIN_IS_REVMAP_PAGE(BufferGetPage(buf)))
656 172 : return;
657 :
658 : /* Measure free space and record it */
659 200 : RecordPageWithFreeSpace(idxrel, BufferGetBlockNumber(buf),
660 : br_page_get_freespace(page));
661 : }
662 :
663 : /*
664 : * Return a pinned and exclusively locked buffer which can be used to insert an
665 : * index item of size itemsz (caller must ensure not to request sizes
666 : * impossible to fulfill). If oldbuf is a valid buffer, it is also locked (in
667 : * an order determined to avoid deadlocks).
668 : *
669 : * If we find that the old page is no longer a regular index page (because
670 : * of a revmap extension), the old buffer is unlocked and we return
671 : * InvalidBuffer.
672 : *
673 : * If there's no existing page with enough free space to accommodate the new
674 : * item, the relation is extended. If this happens, *extended is set to true,
675 : * and it is the caller's responsibility to initialize the page (and WAL-log
676 : * that fact) prior to use. The caller should also update the FSM with the
677 : * page's remaining free space after the insertion.
678 : *
679 : * Note that the caller is not expected to update FSM unless *extended is set
680 : * true. This policy means that we'll update FSM when a page is created, and
681 : * when it's found to have too little space for a desired tuple insertion,
682 : * but not every single time we add a tuple to the page.
683 : *
684 : * Note that in some corner cases it is possible for this routine to extend
685 : * the relation and then not return the new page. It is this routine's
686 : * responsibility to WAL-log the page initialization and to record the page in
687 : * FSM if that happens, since the caller certainly can't do it.
688 : */
689 : static Buffer
690 4000 : brin_getinsertbuffer(Relation irel, Buffer oldbuf, Size itemsz,
691 : bool *extended)
692 : {
693 : BlockNumber oldblk;
694 : BlockNumber newblk;
695 : Page page;
696 : Size freespace;
697 :
698 : /* callers must have checked */
699 : Assert(itemsz <= BrinMaxItemSize);
700 :
701 4000 : if (BufferIsValid(oldbuf))
702 614 : oldblk = BufferGetBlockNumber(oldbuf);
703 : else
704 3386 : oldblk = InvalidBlockNumber;
705 :
706 : /* Choose initial target page, re-using existing target if known */
707 4000 : newblk = RelationGetTargetBlock(irel);
708 4000 : if (newblk == InvalidBlockNumber)
709 400 : newblk = GetPageWithFreeSpace(irel, itemsz);
710 :
711 : /*
712 : * Loop until we find a page with sufficient free space. By the time we
713 : * return to caller out of this loop, both buffers are valid and locked;
714 : * if we have to restart here, neither page is locked and newblk isn't
715 : * pinned (if it's even valid).
716 : */
717 : for (;;)
718 148 : {
719 : Buffer buf;
720 4148 : bool extensionLockHeld = false;
721 :
722 4148 : CHECK_FOR_INTERRUPTS();
723 :
724 4148 : *extended = false;
725 :
726 4148 : if (newblk == InvalidBlockNumber)
727 : {
728 : /*
729 : * There's not enough free space in any existing index page,
730 : * according to the FSM: extend the relation to obtain a shiny new
731 : * page.
732 : *
733 : * XXX: It's likely possible to use RBM_ZERO_AND_LOCK here,
734 : * which'd avoid the need to hold the extension lock during buffer
735 : * reclaim.
736 : */
737 474 : if (!RELATION_IS_LOCAL(irel))
738 : {
739 46 : LockRelationForExtension(irel, ExclusiveLock);
740 46 : extensionLockHeld = true;
741 : }
742 474 : buf = ReadBuffer(irel, P_NEW);
743 474 : newblk = BufferGetBlockNumber(buf);
744 474 : *extended = true;
745 :
746 : BRIN_elog((DEBUG2, "brin_getinsertbuffer: extending to page %u",
747 : BufferGetBlockNumber(buf)));
748 : }
749 3674 : else if (newblk == oldblk)
750 : {
751 : /*
752 : * There's an odd corner-case here where the FSM is out-of-date,
753 : * and gave us the old page.
754 : */
755 26 : buf = oldbuf;
756 : }
757 : else
758 : {
759 3648 : buf = ReadBuffer(irel, newblk);
760 : }
761 :
762 : /*
763 : * We lock the old buffer first, if it's earlier than the new one; but
764 : * then we need to check that it hasn't been turned into a revmap page
765 : * concurrently. If we detect that that happened, give up and tell
766 : * caller to start over.
767 : */
768 4148 : if (BufferIsValid(oldbuf) && oldblk < newblk)
769 : {
770 622 : LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);
771 622 : if (!BRIN_IS_REGULAR_PAGE(BufferGetPage(oldbuf)))
772 : {
773 0 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
774 :
775 : /*
776 : * It is possible that the new page was obtained from
777 : * extending the relation. In that case, we must be sure to
778 : * record it in the FSM before leaving, because otherwise the
779 : * space would be lost forever. However, we cannot let an
780 : * uninitialized page get in the FSM, so we need to initialize
781 : * it first.
782 : */
783 0 : if (*extended)
784 0 : brin_initialize_empty_new_buffer(irel, buf);
785 :
786 0 : if (extensionLockHeld)
787 0 : UnlockRelationForExtension(irel, ExclusiveLock);
788 :
789 0 : ReleaseBuffer(buf);
790 :
791 0 : if (*extended)
792 : {
793 0 : FreeSpaceMapVacuumRange(irel, newblk, newblk + 1);
794 : /* shouldn't matter, but don't confuse caller */
795 0 : *extended = false;
796 : }
797 :
798 0 : return InvalidBuffer;
799 : }
800 : }
801 :
802 4148 : LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
803 :
804 4148 : if (extensionLockHeld)
805 46 : UnlockRelationForExtension(irel, ExclusiveLock);
806 :
807 4148 : page = BufferGetPage(buf);
808 :
809 : /*
810 : * We have a new buffer to insert into. Check that the new page has
811 : * enough free space, and return it if it does; otherwise start over.
812 : * (br_page_get_freespace also checks that the FSM didn't hand us a
813 : * page that has since been repurposed for the revmap.)
814 : */
815 8296 : freespace = *extended ?
816 4148 : BrinMaxItemSize : br_page_get_freespace(page);
817 4148 : if (freespace >= itemsz)
818 : {
819 4000 : RelationSetTargetBlock(irel, newblk);
820 :
821 : /*
822 : * Lock the old buffer if not locked already. Note that in this
823 : * case we know for sure it's a regular page: it's later than the
824 : * new page we just got, which is not a revmap page, and revmap
825 : * pages are always consecutive.
826 : */
827 4000 : if (BufferIsValid(oldbuf) && oldblk > newblk)
828 : {
829 0 : LockBuffer(oldbuf, BUFFER_LOCK_EXCLUSIVE);
830 : Assert(BRIN_IS_REGULAR_PAGE(BufferGetPage(oldbuf)));
831 : }
832 :
833 4000 : return buf;
834 : }
835 :
836 : /* This page is no good. */
837 :
838 : /*
839 : * If an entirely new page does not contain enough free space for the
840 : * new item, then surely that item is oversized. Complain loudly; but
841 : * first make sure we initialize the page and record it as free, for
842 : * next time.
843 : */
844 148 : if (*extended)
845 : {
846 0 : brin_initialize_empty_new_buffer(irel, buf);
847 : /* since this should not happen, skip FreeSpaceMapVacuum */
848 :
849 0 : ereport(ERROR,
850 : (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
851 : errmsg("index row size %zu exceeds maximum %zu for index \"%s\"",
852 : itemsz, freespace, RelationGetRelationName(irel))));
853 : return InvalidBuffer; /* keep compiler quiet */
854 : }
855 :
856 148 : if (newblk != oldblk)
857 122 : UnlockReleaseBuffer(buf);
858 148 : if (BufferIsValid(oldbuf) && oldblk <= newblk)
859 34 : LockBuffer(oldbuf, BUFFER_LOCK_UNLOCK);
860 :
861 : /*
862 : * Update the FSM with the new, presumably smaller, freespace value
863 : * for this page, then search for a new target page.
864 : */
865 148 : newblk = RecordAndGetPageWithFreeSpace(irel, newblk, freespace, itemsz);
866 : }
867 : }
868 :
869 : /*
870 : * Initialize a page as an empty regular BRIN page, WAL-log this, and record
871 : * the page in FSM.
872 : *
873 : * There are several corner situations in which we extend the relation to
874 : * obtain a new page and later find that we cannot use it immediately. When
875 : * that happens, we don't want to leave the page go unrecorded in FSM, because
876 : * there is no mechanism to get the space back and the index would bloat.
877 : * Also, because we would not WAL-log the action that would initialize the
878 : * page, the page would go uninitialized in a standby (or after recovery).
879 : *
880 : * While we record the page in FSM here, caller is responsible for doing FSM
881 : * upper-page update if that seems appropriate.
882 : */
883 : static void
884 0 : brin_initialize_empty_new_buffer(Relation idxrel, Buffer buffer)
885 : {
886 : Page page;
887 :
888 : BRIN_elog((DEBUG2,
889 : "brin_initialize_empty_new_buffer: initializing blank page %u",
890 : BufferGetBlockNumber(buffer)));
891 :
892 0 : START_CRIT_SECTION();
893 0 : page = BufferGetPage(buffer);
894 0 : brin_page_init(page, BRIN_PAGETYPE_REGULAR);
895 0 : MarkBufferDirty(buffer);
896 0 : log_newpage_buffer(buffer, true);
897 0 : END_CRIT_SECTION();
898 :
899 : /*
900 : * We update the FSM for this page, but this is not WAL-logged. This is
901 : * acceptable because VACUUM will scan the index and update the FSM with
902 : * pages whose FSM records were forgotten in a crash.
903 : */
904 0 : RecordPageWithFreeSpace(idxrel, BufferGetBlockNumber(buffer),
905 : br_page_get_freespace(page));
906 0 : }
907 :
908 :
909 : /*
910 : * Return the amount of free space on a regular BRIN index page.
911 : *
912 : * If the page is not a regular page, or has been marked with the
913 : * BRIN_EVACUATE_PAGE flag, returns 0.
914 : */
915 : static Size
916 6654 : br_page_get_freespace(Page page)
917 : {
918 6654 : if (!BRIN_IS_REGULAR_PAGE(page) ||
919 6654 : (BrinPageFlags(page) & BRIN_EVACUATE_PAGE) != 0)
920 0 : return 0;
921 : else
922 6654 : return PageGetFreeSpace(page);
923 : }
|
