probackup checksum open, support data checksum check

Author: lwx843928 <liyifeng9@huawei.com>
Date:   Fri Jan 21 09:46:16 2022 +0800

    probackup checksum open

    Offering: GaussDB Kernel

    More detail: probackup checksum open, support data checksum check

    merged from liyifeng

src/bin/pg_probackup/backup.cpp

@@ -786,7 +786,7 @@ pgdata_basic_setup(const ConnectionOptions conn_opt, PGNodeInfo *nodeInfo)
         exclusive_backup = true;
     }
 
-    current.checksum_version = 0;
+    current.checksum_version = 1;
 
     nodeInfo->checksum_version = current.checksum_version;
 

src/bin/pg_probackup/data.cpp

@@ -33,9 +33,90 @@ typedef struct DataPage
     char    data[BLCKSZ];
 } DataPage;
 
+uint32 CHECK_STEP = 2;
 static bool get_page_header(FILE *in, const char *fullpath, BackupPageHeader* bph,
                                                 pg_crc32 *crc, bool use_crc32c);
 
+static inline uint32 pg_checksum_init(uint32 seed, uint32 value)
+{
+    CHECKSUM_COMP(seed, value);
+    return seed;
+}
+
+uint32 pg_checksum_block(char* data, uint32 size)
+{
+    uint32 sums[N_SUMS];
+    uint32* dataArr = (uint32*)data;
+    uint32 result = 0;
+    uint32 i, j;
+
+    /* ensure that the size is compatible with the algorithm */
+    Assert((size % (sizeof(uint32) * N_SUMS)) == 0);
+
+    /* initialize partial checksums to their corresponding offsets */
+    for (j = 0; j < N_SUMS; j += CHECK_STEP) {
+        sums[j] = pg_checksum_init(g_checksumBaseOffsets[j], dataArr[j]);
+        sums[j + 1] = pg_checksum_init(g_checksumBaseOffsets[j + 1], dataArr[j + 1]);
+    }
+    dataArr += N_SUMS;
+
+    /* main checksum calculation */
+    for (i = 1; i < size / (sizeof(uint32) * N_SUMS); i++) {
+        for (j = 0; j < N_SUMS; j += CHECK_STEP) {
+            CHECKSUM_COMP(sums[j], dataArr[j]);
+            CHECKSUM_COMP(sums[j + 1], dataArr[j + 1]);
+        }
+        dataArr += N_SUMS;
+    }
+
+    /* finally add in two rounds of zeroes for additional mixing */
+    for (j = 0; j < N_SUMS; j++) {
+        CHECKSUM_COMP(sums[j], 0);
+        CHECKSUM_COMP(sums[j], 0);
+
+        /* xor fold partial checksums together */
+        result ^= sums[j];
+    }
+
+    return result;
+}
+
+/*
+ * Compute the checksum for a openGauss page.  The page must be aligned on a
+ * 4-byte boundary.
+ *
+ * The checksum includes the block number (to detect the case where a page is
+ * somehow moved to a different location), the page header (excluding the
+ * checksum itself), and the page data.
+ */
+uint16 pg_checksum_page(char* page, BlockNumber blkno)
+{
+    PageHeader phdr = (PageHeader)page;
+    uint16 save_checksum;
+    uint32 checksum;
+
+    /*
+     * Save pd_checksum and temporarily set it to zero, so that the checksum
+     * calculation isn't affected by the old checksum stored on the page.
+     * Restore it after, because actually updating the checksum is NOT part of
+     * the API of this function.
+     */
+    save_checksum = phdr->pd_checksum;
+    phdr->pd_checksum = 0;
+    checksum = pg_checksum_block(page, BLCKSZ);
+    phdr->pd_checksum = save_checksum;
+
+    /* Mix in the block number to detect transposed pages */
+    checksum ^= blkno;
+
+    /*
+     * Reduce to a uint16 (to fit in the pd_checksum field) with an offset of
+     * one. That avoids checksums of zero, which seems like a good idea.
+     */
+    return (checksum % UINT16_MAX) + 1;
+}
+
+
 #ifdef HAVE_LIBZ
 /* Implementation of zlib compression method */
 static int32
@@ -270,7 +351,7 @@ get_checksum_errormsg(Page page, char **errormsg, BlockNumber absolute_blkno)
          "page verification failed, "
          "calculated checksum %u but expected %u",
          phdr->pd_checksum,
-         /*pg_checksum_page(page, absolute_blkno)*/0);
+         pg_checksum_page(page, absolute_blkno));
     securec_check_ss_c(nRet, "\0", "\0");
 }
 
@@ -1428,7 +1509,7 @@ validate_one_page(Page page, BlockNumber absolute_blkno,
     }
 
     /* Verify checksum */
-    page_st->checksum = 0;//pg_checksum_page(page, absolute_blkno);
+    page_st->checksum = pg_checksum_page(page, absolute_blkno);
 
     if (checksum_version)
     {
@@ -1734,7 +1815,7 @@ validate_file_pages(pgFile *file, const char *fullpath, XLogRecPtr stop_lsn,
     {
         elog(WARNING, "Invalid CRC of backup file \"%s\": %X. Expected %X",
             fullpath, crc, file->crc);
-            //is_valid = false;
+        is_valid = false;
     }
 
     pg_free(headers);
@@ -1789,7 +1870,9 @@ get_checksum_map(const char *fullpath, uint32 checksum_version,
 
             if (rc == PAGE_IS_VALID)
             {
-                
+                if (checksum_version)
+                    checksum_map[blknum].checksum = ((PageHeader) read_buffer)->pd_checksum;
+                else
                     checksum_map[blknum].checksum = page_st.checksum;
                 checksum_map[blknum].lsn = page_st.lsn;
             }

src/bin/pg_probackup/util.cpp

@@ -36,6 +36,9 @@ static const char *statusName[] =
     "CORRUPT"
 };
 
+uint32 NUM_65536 = 65536;
+uint32 NUM_10000 = 10000;
+
 const char *
 base36enc(long unsigned int value)
 {
@@ -53,7 +56,7 @@ base36enc(long unsigned int value)
 }
 
 /*
- * Same as base36enc(), but the result must be released by the user.
+ * Same as base36enc(), but the results must be released by the user.
  */
 char *
 base36enc_dup(long unsigned int value)
@@ -77,6 +80,30 @@ base36dec(const char *text)
     return strtoul(text, NULL, 36);
 }
 
+static void
+checkControlFile(ControlFileData *ControlFile)
+{
+    pg_crc32c   crc;
+
+    /* Calculate CRC */
+    INIT_CRC32C(crc);
+    COMP_CRC32C(crc, (char *) ControlFile, offsetof(ControlFileData, crc));
+    FIN_CRC32C(crc);
+
+    /* Then compare it */
+    if (!EQ_CRC32C(crc, ControlFile->crc))
+        elog(ERROR, "Calculated CRC checksum does not match value stored in file.\n"
+             "Either the file is corrupt, or it has a different layout than this program\n"
+             "is expecting. The results below are untrustworthy.");
+
+    if ((ControlFile->pg_control_version % NUM_65536 == 0 || ControlFile->pg_control_version % NUM_65536 > NUM_10000) &&
+        ControlFile->pg_control_version / NUM_65536 != 0)
+        elog(ERROR, "possible byte ordering mismatch\n"
+                 "The byte ordering used to store the pg_control file might not match the one\n"
+                 "used by this program. In that case the results below would be incorrect, and\n"
+                 "the PostgreSQL installation would be incompatible with this data directory.");
+}
+
 /*
  * Verify control file contents in the buffer src, and copy it to *ControlFile.
  */
@@ -89,6 +116,9 @@ digestControlFile(ControlFileData *ControlFile, char *src, size_t size)
 
     errno_t rc = memcpy_s(ControlFile, sizeof(ControlFileData), src, sizeof(ControlFileData));
     securec_check_c(rc, "\0", "\0");
+
+    /* Additional checks on control file */
+    checkControlFile(ControlFile);
 }
 
 /*