1 /*-------------------------------------------------------------------------
4 * WAL replay logic for btrees.
7 * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
8 * Portions Copyright (c) 1994, Regents of the University of California
11 * $PostgreSQL: pgsql/src/backend/access/nbtree/nbtxlog.c,v 1.39 2006/11/01 19:43:17 tgl Exp $
13 *-------------------------------------------------------------------------
17 #include "access/nbtree.h"
18 #include "access/transam.h"
21 * We must keep track of expected insertions due to page splits, and apply
22 * them manually if they are not seen in the WAL log during replay. This
23 * makes it safe for page insertion to be a multiple-WAL-action process.
25 * Similarly, deletion of an only child page and deletion of its parent page
26 * form multiple WAL log entries, and we have to be prepared to follow through
27 * with the deletion if the log ends between.
29 * The data structure is a simple linked list --- this should be good enough,
30 * since we don't expect a page split or multi deletion to remain incomplete
31 * for long. In any case we need to respect the order of operations.
33 typedef struct bt_incomplete_action
35 RelFileNode node; /* the index */
36 bool is_split; /* T = pending split, F = pending delete */
37 /* these fields are for a split: */
38 bool is_root; /* we split the root */
39 BlockNumber leftblk; /* left half of split */
40 BlockNumber rightblk; /* right half of split */
41 /* these fields are for a delete: */
42 BlockNumber delblk; /* parent block to be deleted */
43 } bt_incomplete_action;
45 static List *incomplete_actions;
49 log_incomplete_split(RelFileNode node, BlockNumber leftblk,
50 BlockNumber rightblk, bool is_root)
52 bt_incomplete_action *action = palloc(sizeof(bt_incomplete_action));
55 action->is_split = true;
56 action->is_root = is_root;
57 action->leftblk = leftblk;
58 action->rightblk = rightblk;
59 incomplete_actions = lappend(incomplete_actions, action);
63 forget_matching_split(RelFileNode node, BlockNumber downlink, bool is_root)
67 foreach(l, incomplete_actions)
69 bt_incomplete_action *action = (bt_incomplete_action *) lfirst(l);
71 if (RelFileNodeEquals(node, action->node) &&
73 downlink == action->rightblk)
75 if (is_root != action->is_root)
76 elog(LOG, "forget_matching_split: fishy is_root data (expected %d, got %d)",
77 action->is_root, is_root);
78 incomplete_actions = list_delete_ptr(incomplete_actions, action);
80 break; /* need not look further */
86 log_incomplete_deletion(RelFileNode node, BlockNumber delblk)
88 bt_incomplete_action *action = palloc(sizeof(bt_incomplete_action));
91 action->is_split = false;
92 action->delblk = delblk;
93 incomplete_actions = lappend(incomplete_actions, action);
97 forget_matching_deletion(RelFileNode node, BlockNumber delblk)
101 foreach(l, incomplete_actions)
103 bt_incomplete_action *action = (bt_incomplete_action *) lfirst(l);
105 if (RelFileNodeEquals(node, action->node) &&
107 delblk == action->delblk)
109 incomplete_actions = list_delete_ptr(incomplete_actions, action);
111 break; /* need not look further */
117 * _bt_restore_page -- re-enter all the index tuples on a page
119 * The page is freshly init'd, and *from (length len) is a copy of what
120 * had been its upper part (pd_upper to pd_special). We assume that the
121 * tuples had been added to the page in item-number order, and therefore
122 * the one with highest item number appears first (lowest on the page).
124 * NOTE: the way this routine is coded, the rebuilt page will have the items
125 * in correct itemno sequence, but physically the opposite order from the
126 * original, because we insert them in the opposite of itemno order. This
127 * does not matter in any current btree code, but it's something to keep an
128 * eye on. Is it worth changing just on general principles?
131 _bt_restore_page(Page page, char *from, int len)
133 IndexTupleData itupdata;
135 char *end = from + len;
139 /* Need to copy tuple header due to alignment considerations */
140 memcpy(&itupdata, from, sizeof(IndexTupleData));
141 itemsz = IndexTupleDSize(itupdata);
142 itemsz = MAXALIGN(itemsz);
143 if (PageAddItem(page, (Item) from, itemsz,
144 FirstOffsetNumber, LP_USED) == InvalidOffsetNumber)
145 elog(PANIC, "_bt_restore_page: can't add item to page");
151 _bt_restore_meta(Relation reln, XLogRecPtr lsn,
152 BlockNumber root, uint32 level,
153 BlockNumber fastroot, uint32 fastlevel)
160 metabuf = XLogReadBuffer(reln, BTREE_METAPAGE, true);
161 Assert(BufferIsValid(metabuf));
162 metapg = BufferGetPage(metabuf);
164 _bt_pageinit(metapg, BufferGetPageSize(metabuf));
166 md = BTPageGetMeta(metapg);
167 md->btm_magic = BTREE_MAGIC;
168 md->btm_version = BTREE_VERSION;
170 md->btm_level = level;
171 md->btm_fastroot = fastroot;
172 md->btm_fastlevel = fastlevel;
174 pageop = (BTPageOpaque) PageGetSpecialPointer(metapg);
175 pageop->btpo_flags = BTP_META;
178 * Set pd_lower just past the end of the metadata. This is not essential
179 * but it makes the page look compressible to xlog.c.
181 ((PageHeader) metapg)->pd_lower =
182 ((char *) md + sizeof(BTMetaPageData)) - (char *) metapg;
184 PageSetLSN(metapg, lsn);
185 PageSetTLI(metapg, ThisTimeLineID);
186 MarkBufferDirty(metabuf);
187 UnlockReleaseBuffer(metabuf);
191 btree_xlog_insert(bool isleaf, bool ismeta,
192 XLogRecPtr lsn, XLogRecord *record)
194 xl_btree_insert *xlrec = (xl_btree_insert *) XLogRecGetData(record);
200 xl_btree_metadata md;
201 BlockNumber downlink = 0;
203 datapos = (char *) xlrec + SizeOfBtreeInsert;
204 datalen = record->xl_len - SizeOfBtreeInsert;
207 memcpy(&downlink, datapos, sizeof(BlockNumber));
208 datapos += sizeof(BlockNumber);
209 datalen -= sizeof(BlockNumber);
213 memcpy(&md, datapos, sizeof(xl_btree_metadata));
214 datapos += sizeof(xl_btree_metadata);
215 datalen -= sizeof(xl_btree_metadata);
218 if ((record->xl_info & XLR_BKP_BLOCK_1) && !ismeta && isleaf)
219 return; /* nothing to do */
221 reln = XLogOpenRelation(xlrec->target.node);
223 if (!(record->xl_info & XLR_BKP_BLOCK_1))
225 buffer = XLogReadBuffer(reln,
226 ItemPointerGetBlockNumber(&(xlrec->target.tid)),
228 if (BufferIsValid(buffer))
230 page = (Page) BufferGetPage(buffer);
232 if (XLByteLE(lsn, PageGetLSN(page)))
234 UnlockReleaseBuffer(buffer);
238 if (PageAddItem(page, (Item) datapos, datalen,
239 ItemPointerGetOffsetNumber(&(xlrec->target.tid)),
240 LP_USED) == InvalidOffsetNumber)
241 elog(PANIC, "btree_insert_redo: failed to add item");
243 PageSetLSN(page, lsn);
244 PageSetTLI(page, ThisTimeLineID);
245 MarkBufferDirty(buffer);
246 UnlockReleaseBuffer(buffer);
252 _bt_restore_meta(reln, lsn,
254 md.fastroot, md.fastlevel);
256 /* Forget any split this insertion completes */
258 forget_matching_split(xlrec->target.node, downlink, false);
262 btree_xlog_split(bool onleft, bool isroot,
263 XLogRecPtr lsn, XLogRecord *record)
265 xl_btree_split *xlrec = (xl_btree_split *) XLogRecGetData(record);
267 BlockNumber targetblk;
268 OffsetNumber targetoff;
270 BlockNumber rightsib;
271 BlockNumber downlink = 0;
276 reln = XLogOpenRelation(xlrec->target.node);
277 targetblk = ItemPointerGetBlockNumber(&(xlrec->target.tid));
278 targetoff = ItemPointerGetOffsetNumber(&(xlrec->target.tid));
279 leftsib = (onleft) ? targetblk : xlrec->otherblk;
280 rightsib = (onleft) ? xlrec->otherblk : targetblk;
282 /* Left (original) sibling */
283 buffer = XLogReadBuffer(reln, leftsib, true);
284 Assert(BufferIsValid(buffer));
285 page = (Page) BufferGetPage(buffer);
287 _bt_pageinit(page, BufferGetPageSize(buffer));
288 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
290 pageop->btpo_prev = xlrec->leftblk;
291 pageop->btpo_next = rightsib;
292 pageop->btpo.level = xlrec->level;
293 pageop->btpo_flags = (xlrec->level == 0) ? BTP_LEAF : 0;
294 pageop->btpo_cycleid = 0;
296 _bt_restore_page(page,
297 (char *) xlrec + SizeOfBtreeSplit,
300 if (onleft && xlrec->level > 0)
304 /* extract downlink in the target tuple */
305 itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, targetoff));
306 downlink = ItemPointerGetBlockNumber(&(itup->t_tid));
307 Assert(ItemPointerGetOffsetNumber(&(itup->t_tid)) == P_HIKEY);
310 PageSetLSN(page, lsn);
311 PageSetTLI(page, ThisTimeLineID);
312 MarkBufferDirty(buffer);
313 UnlockReleaseBuffer(buffer);
315 /* Right (new) sibling */
316 buffer = XLogReadBuffer(reln, rightsib, true);
317 Assert(BufferIsValid(buffer));
318 page = (Page) BufferGetPage(buffer);
320 _bt_pageinit(page, BufferGetPageSize(buffer));
321 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
323 pageop->btpo_prev = leftsib;
324 pageop->btpo_next = xlrec->rightblk;
325 pageop->btpo.level = xlrec->level;
326 pageop->btpo_flags = (xlrec->level == 0) ? BTP_LEAF : 0;
327 pageop->btpo_cycleid = 0;
329 _bt_restore_page(page,
330 (char *) xlrec + SizeOfBtreeSplit + xlrec->leftlen,
331 record->xl_len - SizeOfBtreeSplit - xlrec->leftlen);
333 if (!onleft && xlrec->level > 0)
337 /* extract downlink in the target tuple */
338 itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, targetoff));
339 downlink = ItemPointerGetBlockNumber(&(itup->t_tid));
340 Assert(ItemPointerGetOffsetNumber(&(itup->t_tid)) == P_HIKEY);
343 PageSetLSN(page, lsn);
344 PageSetTLI(page, ThisTimeLineID);
345 MarkBufferDirty(buffer);
346 UnlockReleaseBuffer(buffer);
348 /* Fix left-link of right (next) page */
349 if (!(record->xl_info & XLR_BKP_BLOCK_1))
351 if (xlrec->rightblk != P_NONE)
353 buffer = XLogReadBuffer(reln, xlrec->rightblk, false);
354 if (BufferIsValid(buffer))
356 page = (Page) BufferGetPage(buffer);
358 if (XLByteLE(lsn, PageGetLSN(page)))
360 UnlockReleaseBuffer(buffer);
364 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
365 pageop->btpo_prev = rightsib;
367 PageSetLSN(page, lsn);
368 PageSetTLI(page, ThisTimeLineID);
369 MarkBufferDirty(buffer);
370 UnlockReleaseBuffer(buffer);
376 /* Forget any split this insertion completes */
377 if (xlrec->level > 0)
378 forget_matching_split(xlrec->target.node, downlink, false);
380 /* The job ain't done till the parent link is inserted... */
381 log_incomplete_split(xlrec->target.node,
382 leftsib, rightsib, isroot);
386 btree_xlog_delete(XLogRecPtr lsn, XLogRecord *record)
388 xl_btree_delete *xlrec;
394 if (record->xl_info & XLR_BKP_BLOCK_1)
397 xlrec = (xl_btree_delete *) XLogRecGetData(record);
398 reln = XLogOpenRelation(xlrec->node);
399 buffer = XLogReadBuffer(reln, xlrec->block, false);
400 if (!BufferIsValid(buffer))
402 page = (Page) BufferGetPage(buffer);
404 if (XLByteLE(lsn, PageGetLSN(page)))
406 UnlockReleaseBuffer(buffer);
410 if (record->xl_len > SizeOfBtreeDelete)
412 OffsetNumber *unused;
415 unused = (OffsetNumber *) ((char *) xlrec + SizeOfBtreeDelete);
416 unend = (OffsetNumber *) ((char *) xlrec + record->xl_len);
418 PageIndexMultiDelete(page, unused, unend - unused);
422 * Mark the page as not containing any LP_DELETE items --- see comments in
425 opaque = (BTPageOpaque) PageGetSpecialPointer(page);
426 opaque->btpo_flags &= ~BTP_HAS_GARBAGE;
428 PageSetLSN(page, lsn);
429 PageSetTLI(page, ThisTimeLineID);
430 MarkBufferDirty(buffer);
431 UnlockReleaseBuffer(buffer);
435 btree_xlog_delete_page(uint8 info, XLogRecPtr lsn, XLogRecord *record)
437 xl_btree_delete_page *xlrec = (xl_btree_delete_page *) XLogRecGetData(record);
442 BlockNumber rightsib;
447 reln = XLogOpenRelation(xlrec->target.node);
448 parent = ItemPointerGetBlockNumber(&(xlrec->target.tid));
449 target = xlrec->deadblk;
450 leftsib = xlrec->leftblk;
451 rightsib = xlrec->rightblk;
454 if (!(record->xl_info & XLR_BKP_BLOCK_1))
456 buffer = XLogReadBuffer(reln, parent, false);
457 if (BufferIsValid(buffer))
459 page = (Page) BufferGetPage(buffer);
460 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
461 if (XLByteLE(lsn, PageGetLSN(page)))
463 UnlockReleaseBuffer(buffer);
467 OffsetNumber poffset;
469 poffset = ItemPointerGetOffsetNumber(&(xlrec->target.tid));
470 if (poffset >= PageGetMaxOffsetNumber(page))
472 Assert(info == XLOG_BTREE_DELETE_PAGE_HALF);
473 Assert(poffset == P_FIRSTDATAKEY(pageop));
474 PageIndexTupleDelete(page, poffset);
475 pageop->btpo_flags |= BTP_HALF_DEAD;
481 OffsetNumber nextoffset;
483 Assert(info != XLOG_BTREE_DELETE_PAGE_HALF);
484 itemid = PageGetItemId(page, poffset);
485 itup = (IndexTuple) PageGetItem(page, itemid);
486 ItemPointerSet(&(itup->t_tid), rightsib, P_HIKEY);
487 nextoffset = OffsetNumberNext(poffset);
488 PageIndexTupleDelete(page, nextoffset);
491 PageSetLSN(page, lsn);
492 PageSetTLI(page, ThisTimeLineID);
493 MarkBufferDirty(buffer);
494 UnlockReleaseBuffer(buffer);
499 /* Fix left-link of right sibling */
500 if (!(record->xl_info & XLR_BKP_BLOCK_2))
502 buffer = XLogReadBuffer(reln, rightsib, false);
503 if (BufferIsValid(buffer))
505 page = (Page) BufferGetPage(buffer);
506 if (XLByteLE(lsn, PageGetLSN(page)))
508 UnlockReleaseBuffer(buffer);
512 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
513 pageop->btpo_prev = leftsib;
515 PageSetLSN(page, lsn);
516 PageSetTLI(page, ThisTimeLineID);
517 MarkBufferDirty(buffer);
518 UnlockReleaseBuffer(buffer);
523 /* Fix right-link of left sibling, if any */
524 if (!(record->xl_info & XLR_BKP_BLOCK_3))
526 if (leftsib != P_NONE)
528 buffer = XLogReadBuffer(reln, leftsib, false);
529 if (BufferIsValid(buffer))
531 page = (Page) BufferGetPage(buffer);
532 if (XLByteLE(lsn, PageGetLSN(page)))
534 UnlockReleaseBuffer(buffer);
538 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
539 pageop->btpo_next = rightsib;
541 PageSetLSN(page, lsn);
542 PageSetTLI(page, ThisTimeLineID);
543 MarkBufferDirty(buffer);
544 UnlockReleaseBuffer(buffer);
550 /* Rewrite target page as empty deleted page */
551 buffer = XLogReadBuffer(reln, target, true);
552 Assert(BufferIsValid(buffer));
553 page = (Page) BufferGetPage(buffer);
555 _bt_pageinit(page, BufferGetPageSize(buffer));
556 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
558 pageop->btpo_prev = leftsib;
559 pageop->btpo_next = rightsib;
560 pageop->btpo.xact = FrozenTransactionId;
561 pageop->btpo_flags = BTP_DELETED;
562 pageop->btpo_cycleid = 0;
564 PageSetLSN(page, lsn);
565 PageSetTLI(page, ThisTimeLineID);
566 MarkBufferDirty(buffer);
567 UnlockReleaseBuffer(buffer);
569 /* Update metapage if needed */
570 if (info == XLOG_BTREE_DELETE_PAGE_META)
572 xl_btree_metadata md;
574 memcpy(&md, (char *) xlrec + SizeOfBtreeDeletePage,
575 sizeof(xl_btree_metadata));
576 _bt_restore_meta(reln, lsn,
578 md.fastroot, md.fastlevel);
581 /* Forget any completed deletion */
582 forget_matching_deletion(xlrec->target.node, target);
584 /* If parent became half-dead, remember it for deletion */
585 if (info == XLOG_BTREE_DELETE_PAGE_HALF)
586 log_incomplete_deletion(xlrec->target.node, parent);
590 btree_xlog_newroot(XLogRecPtr lsn, XLogRecord *record)
592 xl_btree_newroot *xlrec = (xl_btree_newroot *) XLogRecGetData(record);
597 BlockNumber downlink = 0;
599 reln = XLogOpenRelation(xlrec->node);
600 buffer = XLogReadBuffer(reln, xlrec->rootblk, true);
601 Assert(BufferIsValid(buffer));
602 page = (Page) BufferGetPage(buffer);
604 _bt_pageinit(page, BufferGetPageSize(buffer));
605 pageop = (BTPageOpaque) PageGetSpecialPointer(page);
607 pageop->btpo_flags = BTP_ROOT;
608 pageop->btpo_prev = pageop->btpo_next = P_NONE;
609 pageop->btpo.level = xlrec->level;
610 if (xlrec->level == 0)
611 pageop->btpo_flags |= BTP_LEAF;
612 pageop->btpo_cycleid = 0;
614 if (record->xl_len > SizeOfBtreeNewroot)
618 _bt_restore_page(page,
619 (char *) xlrec + SizeOfBtreeNewroot,
620 record->xl_len - SizeOfBtreeNewroot);
621 /* extract downlink to the right-hand split page */
622 itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, P_FIRSTKEY));
623 downlink = ItemPointerGetBlockNumber(&(itup->t_tid));
624 Assert(ItemPointerGetOffsetNumber(&(itup->t_tid)) == P_HIKEY);
627 PageSetLSN(page, lsn);
628 PageSetTLI(page, ThisTimeLineID);
629 MarkBufferDirty(buffer);
630 UnlockReleaseBuffer(buffer);
632 _bt_restore_meta(reln, lsn,
633 xlrec->rootblk, xlrec->level,
634 xlrec->rootblk, xlrec->level);
636 /* Check to see if this satisfies any incomplete insertions */
637 if (record->xl_len > SizeOfBtreeNewroot)
638 forget_matching_split(xlrec->node, downlink, true);
643 btree_redo(XLogRecPtr lsn, XLogRecord *record)
645 uint8 info = record->xl_info & ~XLR_INFO_MASK;
649 case XLOG_BTREE_INSERT_LEAF:
650 btree_xlog_insert(true, false, lsn, record);
652 case XLOG_BTREE_INSERT_UPPER:
653 btree_xlog_insert(false, false, lsn, record);
655 case XLOG_BTREE_INSERT_META:
656 btree_xlog_insert(false, true, lsn, record);
658 case XLOG_BTREE_SPLIT_L:
659 btree_xlog_split(true, false, lsn, record);
661 case XLOG_BTREE_SPLIT_R:
662 btree_xlog_split(false, false, lsn, record);
664 case XLOG_BTREE_SPLIT_L_ROOT:
665 btree_xlog_split(true, true, lsn, record);
667 case XLOG_BTREE_SPLIT_R_ROOT:
668 btree_xlog_split(false, true, lsn, record);
670 case XLOG_BTREE_DELETE:
671 btree_xlog_delete(lsn, record);
673 case XLOG_BTREE_DELETE_PAGE:
674 case XLOG_BTREE_DELETE_PAGE_META:
675 case XLOG_BTREE_DELETE_PAGE_HALF:
676 btree_xlog_delete_page(info, lsn, record);
678 case XLOG_BTREE_NEWROOT:
679 btree_xlog_newroot(lsn, record);
682 elog(PANIC, "btree_redo: unknown op code %u", info);
687 out_target(StringInfo buf, xl_btreetid *target)
689 appendStringInfo(buf, "rel %u/%u/%u; tid %u/%u",
690 target->node.spcNode, target->node.dbNode, target->node.relNode,
691 ItemPointerGetBlockNumber(&(target->tid)),
692 ItemPointerGetOffsetNumber(&(target->tid)));
696 btree_desc(StringInfo buf, uint8 xl_info, char *rec)
698 uint8 info = xl_info & ~XLR_INFO_MASK;
702 case XLOG_BTREE_INSERT_LEAF:
704 xl_btree_insert *xlrec = (xl_btree_insert *) rec;
706 appendStringInfo(buf, "insert: ");
707 out_target(buf, &(xlrec->target));
710 case XLOG_BTREE_INSERT_UPPER:
712 xl_btree_insert *xlrec = (xl_btree_insert *) rec;
714 appendStringInfo(buf, "insert_upper: ");
715 out_target(buf, &(xlrec->target));
718 case XLOG_BTREE_INSERT_META:
720 xl_btree_insert *xlrec = (xl_btree_insert *) rec;
722 appendStringInfo(buf, "insert_meta: ");
723 out_target(buf, &(xlrec->target));
726 case XLOG_BTREE_SPLIT_L:
728 xl_btree_split *xlrec = (xl_btree_split *) rec;
730 appendStringInfo(buf, "split_l: ");
731 out_target(buf, &(xlrec->target));
732 appendStringInfo(buf, "; oth %u; rgh %u",
733 xlrec->otherblk, xlrec->rightblk);
736 case XLOG_BTREE_SPLIT_R:
738 xl_btree_split *xlrec = (xl_btree_split *) rec;
740 appendStringInfo(buf, "split_r: ");
741 out_target(buf, &(xlrec->target));
742 appendStringInfo(buf, "; oth %u; rgh %u",
743 xlrec->otherblk, xlrec->rightblk);
746 case XLOG_BTREE_SPLIT_L_ROOT:
748 xl_btree_split *xlrec = (xl_btree_split *) rec;
750 appendStringInfo(buf, "split_l_root: ");
751 out_target(buf, &(xlrec->target));
752 appendStringInfo(buf, "; oth %u; rgh %u",
753 xlrec->otherblk, xlrec->rightblk);
756 case XLOG_BTREE_SPLIT_R_ROOT:
758 xl_btree_split *xlrec = (xl_btree_split *) rec;
760 appendStringInfo(buf, "split_r_root: ");
761 out_target(buf, &(xlrec->target));
762 appendStringInfo(buf, "; oth %u; rgh %u",
763 xlrec->otherblk, xlrec->rightblk);
766 case XLOG_BTREE_DELETE:
768 xl_btree_delete *xlrec = (xl_btree_delete *) rec;
770 appendStringInfo(buf, "delete: rel %u/%u/%u; blk %u",
771 xlrec->node.spcNode, xlrec->node.dbNode,
772 xlrec->node.relNode, xlrec->block);
775 case XLOG_BTREE_DELETE_PAGE:
776 case XLOG_BTREE_DELETE_PAGE_META:
777 case XLOG_BTREE_DELETE_PAGE_HALF:
779 xl_btree_delete_page *xlrec = (xl_btree_delete_page *) rec;
781 appendStringInfo(buf, "delete_page: ");
782 out_target(buf, &(xlrec->target));
783 appendStringInfo(buf, "; dead %u; left %u; right %u",
784 xlrec->deadblk, xlrec->leftblk, xlrec->rightblk);
787 case XLOG_BTREE_NEWROOT:
789 xl_btree_newroot *xlrec = (xl_btree_newroot *) rec;
791 appendStringInfo(buf, "newroot: rel %u/%u/%u; root %u lev %u",
792 xlrec->node.spcNode, xlrec->node.dbNode,
794 xlrec->rootblk, xlrec->level);
798 appendStringInfo(buf, "UNKNOWN");
804 btree_xlog_startup(void)
806 incomplete_actions = NIL;
810 btree_xlog_cleanup(void)
814 foreach(l, incomplete_actions)
816 bt_incomplete_action *action = (bt_incomplete_action *) lfirst(l);
819 reln = XLogOpenRelation(action->node);
820 if (action->is_split)
822 /* finish an incomplete split */
827 BTPageOpaque lpageop,
831 lbuf = XLogReadBuffer(reln, action->leftblk, false);
832 /* failure is impossible because we wrote this page earlier */
833 if (!BufferIsValid(lbuf))
834 elog(PANIC, "btree_xlog_cleanup: left block unfound");
835 lpage = (Page) BufferGetPage(lbuf);
836 lpageop = (BTPageOpaque) PageGetSpecialPointer(lpage);
837 rbuf = XLogReadBuffer(reln, action->rightblk, false);
838 /* failure is impossible because we wrote this page earlier */
839 if (!BufferIsValid(rbuf))
840 elog(PANIC, "btree_xlog_cleanup: right block unfound");
841 rpage = (Page) BufferGetPage(rbuf);
842 rpageop = (BTPageOpaque) PageGetSpecialPointer(rpage);
844 /* if the pages are all of their level, it's a only-page split */
845 is_only = P_LEFTMOST(lpageop) && P_RIGHTMOST(rpageop);
847 _bt_insert_parent(reln, lbuf, rbuf, NULL,
848 action->is_root, is_only);
852 /* finish an incomplete deletion (of a half-dead page) */
855 buf = XLogReadBuffer(reln, action->delblk, false);
856 if (BufferIsValid(buf))
857 if (_bt_pagedel(reln, buf, NULL, true) == 0)
858 elog(PANIC, "btree_xlog_cleanup: _bt_pagdel failed");
861 incomplete_actions = NIL;
865 btree_safe_restartpoint(void)
867 if (incomplete_actions)