undo page consistency checker

Patch provide a mechanism for masking the cid bit in undo pages so that
consistecy checker function can compared the undo pages.  Actual consistency
check should be called under the RM's consistency checker function who is
writing the undo because undo pages will be registered under that RM's WAL

Dilip Kumar with help from Amit Khandekar and Rafia Sabih

diff --git a/src/backend/access/undo/undorecord.c b/src/backend/access/undo/undorecord.c
index a44bfbd3745c4f3f88fa30b6af85fbfe328ab08a..5699153accbcb7ccd9986e4b30b21aff50628084 100755 (executable)
--- a/src/backend/access/undo/undorecord.c
+++ b/src/backend/access/undo/undorecord.c
@@ -1049,3 +1049,113 @@ UndoRecordSetInfo(UnpackedUndoRecord *uur)
        if (uur->uur_payload.len || uur->uur_tuple.len)
                uur->uur_info |= UREC_INFO_PAYLOAD;
 }
+
+/*
+ * Get the offset of cid information in undo record.
+ */
+static Size
+get_undo_rec_cid_offset(uint16 urec_info)
+{
+       Size            offset_size = SizeOfUndoRecordHeader;
+
+       if ((urec_info & UREC_INFO_TRANSACTION) != 0)
+               offset_size += SizeOfUndoRecordTransaction;
+
+       if ((urec_info & UREC_INFO_RMID) != 0)
+               offset_size += sizeof(RmgrId);
+
+       if ((urec_info & UREC_INFO_RELOID) != 0)
+               offset_size += sizeof(Oid);
+
+       if ((urec_info & UREC_INFO_XID) != 0)
+               offset_size += sizeof(FullTransactionId);
+
+       return offset_size;
+}
+
+/*
+ * Mask a undo page before performing consistency checks on it.
+ */
+void
+mask_undo_page(char *pagedata)
+{
+       Page            page = (Page) pagedata;
+       char       *page_end = pagedata + PageGetPageSize(page);
+       char       *next_record;
+       int                     cid_offset;
+       UndoPageHeader phdr = (UndoPageHeader) page;
+
+       next_record = (char *) page + SizeOfUndoPageHeaderData;
+
+       /*
+        * If record_offset is non-zero value in the page header that means page
+        * has a partial record.
+        */
+       if (phdr->record_offset != 0)
+       {
+               Size            partial_rec_size;
+
+               /* Calculate the size of the partial record. */
+               partial_rec_size = UndoRecordHeaderSize(phdr->uur_info) +
+                       phdr->tuple_len + phdr->payload_len -
+                       phdr->record_offset;
+               if ((phdr->uur_info & UREC_INFO_CID) != 0)
+               {
+                       cid_offset = get_undo_rec_cid_offset(phdr->uur_info);
+
+                       /*
+                        * We just want to mask the cid in the undo record header.  So
+                        * only if the partial record in the current page include the undo
+                        * record header then we need to mask the cid bytes in this page.
+                        * Otherwise, directly jump to the next record.
+                        */
+                       if (phdr->record_offset < (cid_offset + sizeof(CommandId)))
+                       {
+                               char       *cid_data;
+                               Size            mask_size;
+
+                               mask_size = Min(cid_offset - phdr->record_offset,
+                                                               sizeof(CommandId));
+
+                               cid_data = next_record + cid_offset - phdr->record_offset;
+                               memset(&cid_data, MASK_MARKER, mask_size);
+                       }
+               }
+
+               next_record += partial_rec_size;
+       }
+
+       /*
+        * Process the undo record of the page and mask their cid filed.
+        */
+       while (next_record < page_end)
+       {
+               UndoRecordHeader *header = (UndoRecordHeader *) next_record;
+
+               /* If this undo record has cid present, then mask it */
+               if ((header->urec_info & UREC_INFO_CID) != 0)
+               {
+                       cid_offset = get_undo_rec_cid_offset(header->urec_info);
+
+                       /*
+                        * If this is not complete record then check whether cid is on
+                        * this page or not.  If not then we are done with this page.
+                        */
+                       if ((next_record + cid_offset + sizeof(CommandId)) > page_end)
+                       {
+                               int                     mask_size = page_end - next_record - cid_offset;
+
+                               if (mask_size > 0)
+                                       memset(next_record + cid_offset, MASK_MARKER, mask_size);
+                               break;
+                       }
+                       else
+                       {
+                               /* Mask cid */
+                               memset(next_record + cid_offset, MASK_MARKER, sizeof(CommandId));
+                       }
+               }
+               /* Go to next record. */
+               next_record += UndoRecordSizeOnPage(next_record);
+       }
+}

diff --git a/src/include/access/undorecord.h b/src/include/access/undorecord.h
index 065326705d4f0cd33467d085836a8e08645e23db..6a2c5cc05910003a7ce8b14d13b24b91853204a9 100644 (file)
--- a/src/include/access/undorecord.h
+++ b/src/include/access/undorecord.h
@@ -276,4 +276,5 @@ extern void FinishUnpackUndo(UndoPackContext *ucontext,
                                                         UnpackedUndoRecord *uur);
 extern void UndoRecordSetInfo(UnpackedUndoRecord *uur);
 
+extern void mask_undo_page(char *pagedata);
 #endif                                                 /* UNDORECORD_H */