From: Kyotaro Horiguchi Date: Mon, 25 Mar 2019 13:29:50 +0900 Subject: [PATCH 3/7] Move XLOG stuff from heap_insert and heap_delete Succeeding commit makes heap_update emit insert and delete WAL records. Move out XLOG stuff for insert and delete so that heap_update can use the stuff. --- src/backend/access/heap/heapam.c | 277 ++++++++++++++++++++++----------------- 1 file changed, 157 insertions(+), 120 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 65536c7214..fe5d939c45 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -71,6 +71,11 @@ static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid, CommandId cid, int options); +static XLogRecPtr log_heap_insert(Relation relation, Buffer buffer, + HeapTuple heaptup, int options, bool all_visible_cleared); +static XLogRecPtr log_heap_delete(Relation relation, Buffer buffer, + HeapTuple tp, HeapTuple old_key_tuple, TransactionId new_xmax, + bool changingPart, bool all_visible_cleared); static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, Buffer newbuf, HeapTuple oldtup, HeapTuple newtup, HeapTuple old_key_tup, @@ -1889,6 +1894,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid, TransactionId xid = GetCurrentTransactionId(); HeapTuple heaptup; Buffer buffer; + Page page; Buffer vmbuffer = InvalidBuffer; bool all_visible_cleared = false; @@ -1925,16 +1931,18 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid, */ CheckForSerializableConflictIn(relation, NULL, InvalidBuffer); + page = BufferGetPage(buffer); + /* NO EREPORT(ERROR) from here till changes are logged */ START_CRIT_SECTION(); RelationPutHeapTuple(relation, buffer, heaptup, (options & HEAP_INSERT_SPECULATIVE) != 0); - if (PageIsAllVisible(BufferGetPage(buffer))) + if (PageIsAllVisible(page)) { all_visible_cleared = true; - PageClearAllVisible(BufferGetPage(buffer)); + PageClearAllVisible(page); visibilitymap_clear(relation, ItemPointerGetBlockNumber(&(heaptup->t_self)), vmbuffer, VISIBILITYMAP_VALID_BITS); @@ -1956,76 +1964,11 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid, /* XLOG stuff */ if (!(options & HEAP_INSERT_SKIP_WAL) && RelationNeedsWAL(relation)) { - xl_heap_insert xlrec; - xl_heap_header xlhdr; XLogRecPtr recptr; - Page page = BufferGetPage(buffer); - uint8 info = XLOG_HEAP_INSERT; - int bufflags = 0; - - /* - * If this is a catalog, we need to transmit combocids to properly - * decode, so log that as well. - */ - if (RelationIsAccessibleInLogicalDecoding(relation)) - log_heap_new_cid(relation, heaptup); - - /* - * If this is the single and first tuple on page, we can reinit the - * page instead of restoring the whole thing. Set flag, and hide - * buffer references from XLogInsert. - */ - if (ItemPointerGetOffsetNumber(&(heaptup->t_self)) == FirstOffsetNumber && - PageGetMaxOffsetNumber(page) == FirstOffsetNumber) - { - info |= XLOG_HEAP_INIT_PAGE; - bufflags |= REGBUF_WILL_INIT; - } - - xlrec.offnum = ItemPointerGetOffsetNumber(&heaptup->t_self); - xlrec.flags = 0; - if (all_visible_cleared) - xlrec.flags |= XLH_INSERT_ALL_VISIBLE_CLEARED; - if (options & HEAP_INSERT_SPECULATIVE) - xlrec.flags |= XLH_INSERT_IS_SPECULATIVE; - Assert(ItemPointerGetBlockNumber(&heaptup->t_self) == BufferGetBlockNumber(buffer)); - - /* - * For logical decoding, we need the tuple even if we're doing a full - * page write, so make sure it's included even if we take a full-page - * image. (XXX We could alternatively store a pointer into the FPW). - */ - if (RelationIsLogicallyLogged(relation) && - !(options & HEAP_INSERT_NO_LOGICAL)) - { - xlrec.flags |= XLH_INSERT_CONTAINS_NEW_TUPLE; - bufflags |= REGBUF_KEEP_DATA; - } - - XLogBeginInsert(); - XLogRegisterData((char *) &xlrec, SizeOfHeapInsert); - - xlhdr.t_infomask2 = heaptup->t_data->t_infomask2; - xlhdr.t_infomask = heaptup->t_data->t_infomask; - xlhdr.t_hoff = heaptup->t_data->t_hoff; - - /* - * note we mark xlhdr as belonging to buffer; if XLogInsert decides to - * write the whole page to the xlog, we don't need to store - * xl_heap_header in the xlog. - */ - XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags); - XLogRegisterBufData(0, (char *) &xlhdr, SizeOfHeapHeader); - /* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */ - XLogRegisterBufData(0, - (char *) heaptup->t_data + SizeofHeapTupleHeader, - heaptup->t_len - SizeofHeapTupleHeader); - - /* filtering by origin on a row level is much more efficient */ - XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); - - recptr = XLogInsert(RM_HEAP_ID, info); + recptr = log_heap_insert(relation, buffer, heaptup, + options, all_visible_cleared); + PageSetLSN(page, recptr); } @@ -2744,58 +2687,10 @@ l1: */ if (RelationNeedsWAL(relation)) { - xl_heap_delete xlrec; - xl_heap_header xlhdr; XLogRecPtr recptr; - /* For logical decode we need combocids to properly decode the catalog */ - if (RelationIsAccessibleInLogicalDecoding(relation)) - log_heap_new_cid(relation, &tp); - - xlrec.flags = 0; - if (all_visible_cleared) - xlrec.flags |= XLH_DELETE_ALL_VISIBLE_CLEARED; - if (changingPart) - xlrec.flags |= XLH_DELETE_IS_PARTITION_MOVE; - xlrec.infobits_set = compute_infobits(tp.t_data->t_infomask, - tp.t_data->t_infomask2); - xlrec.offnum = ItemPointerGetOffsetNumber(&tp.t_self); - xlrec.xmax = new_xmax; - - if (old_key_tuple != NULL) - { - if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL) - xlrec.flags |= XLH_DELETE_CONTAINS_OLD_TUPLE; - else - xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY; - } - - XLogBeginInsert(); - XLogRegisterData((char *) &xlrec, SizeOfHeapDelete); - - XLogRegisterBuffer(0, buffer, REGBUF_STANDARD); - - /* - * Log replica identity of the deleted tuple if there is one - */ - if (old_key_tuple != NULL) - { - xlhdr.t_infomask2 = old_key_tuple->t_data->t_infomask2; - xlhdr.t_infomask = old_key_tuple->t_data->t_infomask; - xlhdr.t_hoff = old_key_tuple->t_data->t_hoff; - - XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader); - XLogRegisterData((char *) old_key_tuple->t_data - + SizeofHeapTupleHeader, - old_key_tuple->t_len - - SizeofHeapTupleHeader); - } - - /* filtering by origin on a row level is much more efficient */ - XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); - - recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE); - + recptr = log_heap_delete(relation, buffer, &tp, old_key_tuple, new_xmax, + changingPart, all_visible_cleared); PageSetLSN(page, recptr); } @@ -7045,6 +6940,148 @@ log_heap_visible(RelFileNode rnode, Buffer heap_buffer, Buffer vm_buffer, return recptr; } +/* + * Perform XLogInsert for a heap-insert operation. Caller must already + * have modified the buffer and marked it dirty. + */ +XLogRecPtr +log_heap_insert(Relation relation, Buffer buffer, + HeapTuple heaptup, int options, bool all_visible_cleared) +{ + xl_heap_insert xlrec; + xl_heap_header xlhdr; + uint8 info = XLOG_HEAP_INSERT; + int bufflags = 0; + Page page = BufferGetPage(buffer); + + /* + * If this is a catalog, we need to transmit combocids to properly + * decode, so log that as well. + */ + if (RelationIsAccessibleInLogicalDecoding(relation)) + log_heap_new_cid(relation, heaptup); + + /* + * If this is the single and first tuple on page, we can reinit the + * page instead of restoring the whole thing. Set flag, and hide + * buffer references from XLogInsert. + */ + if (ItemPointerGetOffsetNumber(&(heaptup->t_self)) == FirstOffsetNumber && + PageGetMaxOffsetNumber(page) == FirstOffsetNumber) + { + info |= XLOG_HEAP_INIT_PAGE; + bufflags |= REGBUF_WILL_INIT; + } + + xlrec.offnum = ItemPointerGetOffsetNumber(&heaptup->t_self); + xlrec.flags = 0; + if (all_visible_cleared) + xlrec.flags |= XLH_INSERT_ALL_VISIBLE_CLEARED; + if (options & HEAP_INSERT_SPECULATIVE) + xlrec.flags |= XLH_INSERT_IS_SPECULATIVE; + Assert(ItemPointerGetBlockNumber(&heaptup->t_self) == BufferGetBlockNumber(buffer)); + + /* + * For logical decoding, we need the tuple even if we're doing a full + * page write, so make sure it's included even if we take a full-page + * image. (XXX We could alternatively store a pointer into the FPW). + */ + if (RelationIsLogicallyLogged(relation) && + !(options & HEAP_INSERT_NO_LOGICAL)) + { + xlrec.flags |= XLH_INSERT_CONTAINS_NEW_TUPLE; + bufflags |= REGBUF_KEEP_DATA; + } + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, SizeOfHeapInsert); + + xlhdr.t_infomask2 = heaptup->t_data->t_infomask2; + xlhdr.t_infomask = heaptup->t_data->t_infomask; + xlhdr.t_hoff = heaptup->t_data->t_hoff; + + /* + * note we mark xlhdr as belonging to buffer; if XLogInsert decides to + * write the whole page to the xlog, we don't need to store + * xl_heap_header in the xlog. + */ + XLogRegisterBuffer(0, buffer, REGBUF_STANDARD | bufflags); + XLogRegisterBufData(0, (char *) &xlhdr, SizeOfHeapHeader); + /* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */ + XLogRegisterBufData(0, + (char *) heaptup->t_data + SizeofHeapTupleHeader, + heaptup->t_len - SizeofHeapTupleHeader); + + /* filtering by origin on a row level is much more efficient */ + XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); + + return XLogInsert(RM_HEAP_ID, info); +} + +/* + * Perform XLogInsert for a heap-insert operation. Caller must already + * have modified the buffer and marked it dirty. + * + * NB: heap_abort_speculative() uses the same xlog record and replay + * routines. + */ +static XLogRecPtr +log_heap_delete(Relation relation, Buffer buffer, + HeapTuple tp, HeapTuple old_key_tuple, TransactionId new_xmax, + bool changingPart, bool all_visible_cleared) +{ + xl_heap_delete xlrec; + xl_heap_header xlhdr; + + /* For logical decode we need combocids to properly decode the catalog */ + if (RelationIsAccessibleInLogicalDecoding(relation)) + log_heap_new_cid(relation, tp); + + xlrec.flags = 0; + if (all_visible_cleared) + xlrec.flags |= XLH_DELETE_ALL_VISIBLE_CLEARED; + if (changingPart) + xlrec.flags |= XLH_DELETE_IS_PARTITION_MOVE; + xlrec.infobits_set = compute_infobits(tp->t_data->t_infomask, + tp->t_data->t_infomask2); + xlrec.offnum = ItemPointerGetOffsetNumber(&tp->t_self); + xlrec.xmax = new_xmax; + + if (old_key_tuple != NULL) + { + if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL) + xlrec.flags |= XLH_DELETE_CONTAINS_OLD_TUPLE; + else + xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY; + } + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, SizeOfHeapDelete); + + XLogRegisterBuffer(0, buffer, REGBUF_STANDARD); + + /* + * Log replica identity of the deleted tuple if there is one + */ + if (old_key_tuple != NULL) + { + xlhdr.t_infomask2 = old_key_tuple->t_data->t_infomask2; + xlhdr.t_infomask = old_key_tuple->t_data->t_infomask; + xlhdr.t_hoff = old_key_tuple->t_data->t_hoff; + + XLogRegisterData((char *) &xlhdr, SizeOfHeapHeader); + XLogRegisterData((char *) old_key_tuple->t_data + + SizeofHeapTupleHeader, + old_key_tuple->t_len + - SizeofHeapTupleHeader); + } + + /* filtering by origin on a row level is much more efficient */ + XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN); + + return XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE); +} + /* * Perform XLogInsert for a heap-update operation. Caller must already * have modified the buffer(s) and marked them dirty. -- 2.16.3 ----Next_Part(Mon_Mar_25_21_32_04_2019_838)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v8-0004-Add-infrastructure-to-WAL-logging-skip-feature.patch"