Use streaming read I/O in heap amcheck

Instead of directly invoking ReadBuffer() for each unskippable block in
the heap relation, verify_heapam() now uses the read stream API to
acquire the next buffer to check for corruption.

Author: Matheus Alcantara <matheusssilv97@gmail.com>
Co-authored-by: Melanie Plageman <melanieplageman@gmail.com>
Reviewed-by: Nazir Bilal Yavuz <byavuz81@gmail.com>
Reviewed-by: Kirill Reshke <reshkekirill@gmail.com>
Reviewed-by: jian he <jian.universality@gmail.com>
Discussion: https://postgr.es/m/flat/CAFY6G8eLyz7%2BsccegZYFj%3D5tAUR-GZ9uEq4Ch5gvwKqUwb_hCA%40mail.gmail.com

diff --git a/contrib/amcheck/verify_heapam.c b/contrib/amcheck/verify_heapam.c
index 827312306f6043ff98716e6ee59c200d71dfae40..9e4d558436bac50d9cf78607a87a05b9c9b6976a 100644 (file)
--- a/contrib/amcheck/verify_heapam.c
+++ b/contrib/amcheck/verify_heapam.c
@@ -25,6 +25,7 @@
 #include "miscadmin.h"
 #include "storage/bufmgr.h"
 #include "storage/procarray.h"
+#include "storage/read_stream.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/rel.h"
@@ -118,7 +119,10 @@ typedef struct HeapCheckContext
    Relation    valid_toast_index;
    int         num_toast_indexes;
 
-   /* Values for iterating over pages in the relation */
+   /*
+    * Values for iterating over pages in the relation. `blkno` is the most
+    * recent block in the buffer yielded by the read stream API.
+    */
    BlockNumber blkno;
    BufferAccessStrategy bstrategy;
    Buffer      buffer;
@@ -153,7 +157,32 @@ typedef struct HeapCheckContext
    Tuplestorestate *tupstore;
 } HeapCheckContext;
 
+/*
+ * The per-relation data provided to the read stream API for heap amcheck to
+ * use in its callback for the SKIP_PAGES_ALL_FROZEN and
+ * SKIP_PAGES_ALL_VISIBLE options.
+ */
+typedef struct HeapCheckReadStreamData
+{
+   /*
+    * `range` is used by all SkipPages options. SKIP_PAGES_NONE uses the
+    * default read stream callback, block_range_read_stream_cb(), which takes
+    * a BlockRangeReadStreamPrivate as its callback_private_data. `range`
+    * keeps track of the current block number across
+    * read_stream_next_buffer() invocations.
+    */
+   BlockRangeReadStreamPrivate range;
+   SkipPages   skip_option;
+   Relation    rel;
+   Buffer     *vmbuffer;
+} HeapCheckReadStreamData;
+
+
 /* Internal implementation */
+static BlockNumber heapcheck_read_stream_next_unskippable(ReadStream *stream,
+                                                         void *callback_private_data,
+                                                         void *per_buffer_data);
+
 static void check_tuple(HeapCheckContext *ctx,
                        bool *xmin_commit_status_ok,
                        XidCommitStatus *xmin_commit_status);
@@ -231,6 +260,11 @@ verify_heapam(PG_FUNCTION_ARGS)
    BlockNumber last_block;
    BlockNumber nblocks;
    const char *skip;
+   ReadStream *stream;
+   int         stream_flags;
+   ReadStreamBlockNumberCB stream_cb;
+   void       *stream_data;
+   HeapCheckReadStreamData stream_skip_data;
 
    /* Check supplied arguments */
    if (PG_ARGISNULL(0))
@@ -404,7 +438,35 @@ verify_heapam(PG_FUNCTION_ARGS)
    if (TransactionIdIsNormal(ctx.relfrozenxid))
        ctx.oldest_xid = ctx.relfrozenxid;
 
-   for (ctx.blkno = first_block; ctx.blkno <= last_block; ctx.blkno++)
+   /* Now that `ctx` is set up, set up the read stream */
+   stream_skip_data.range.current_blocknum = first_block;
+   stream_skip_data.range.last_exclusive = last_block + 1;
+   stream_skip_data.skip_option = skip_option;
+   stream_skip_data.rel = ctx.rel;
+   stream_skip_data.vmbuffer = &vmbuffer;
+
+   if (skip_option == SKIP_PAGES_NONE)
+   {
+       stream_cb = block_range_read_stream_cb;
+       stream_flags = READ_STREAM_SEQUENTIAL | READ_STREAM_FULL;
+       stream_data = &stream_skip_data.range;
+   }
+   else
+   {
+       stream_cb = heapcheck_read_stream_next_unskippable;
+       stream_flags = READ_STREAM_DEFAULT;
+       stream_data = &stream_skip_data;
+   }
+
+   stream = read_stream_begin_relation(stream_flags,
+                                       ctx.bstrategy,
+                                       ctx.rel,
+                                       MAIN_FORKNUM,
+                                       stream_cb,
+                                       stream_data,
+                                       0);
+
+   while ((ctx.buffer = read_stream_next_buffer(stream, NULL)) != InvalidBuffer)
    {
        OffsetNumber maxoff;
        OffsetNumber predecessor[MaxOffsetNumber];
@@ -417,30 +479,11 @@ verify_heapam(PG_FUNCTION_ARGS)
 
        memset(predecessor, 0, sizeof(OffsetNumber) * MaxOffsetNumber);
 
-       /* Optionally skip over all-frozen or all-visible blocks */
-       if (skip_option != SKIP_PAGES_NONE)
-       {
-           int32       mapbits;
-
-           mapbits = (int32) visibilitymap_get_status(ctx.rel, ctx.blkno,
-                                                      &vmbuffer);
-           if (skip_option == SKIP_PAGES_ALL_FROZEN)
-           {
-               if ((mapbits & VISIBILITYMAP_ALL_FROZEN) != 0)
-                   continue;
-           }
-
-           if (skip_option == SKIP_PAGES_ALL_VISIBLE)
-           {
-               if ((mapbits & VISIBILITYMAP_ALL_VISIBLE) != 0)
-                   continue;
-           }
-       }
-
-       /* Read and lock the next page. */
-       ctx.buffer = ReadBufferExtended(ctx.rel, MAIN_FORKNUM, ctx.blkno,
-                                       RBM_NORMAL, ctx.bstrategy);
+       /* Lock the next page. */
+       Assert(BufferIsValid(ctx.buffer));
        LockBuffer(ctx.buffer, BUFFER_LOCK_SHARE);
+
+       ctx.blkno = BufferGetBlockNumber(ctx.buffer);
        ctx.page = BufferGetPage(ctx.buffer);
 
        /* Perform tuple checks */
@@ -799,6 +842,10 @@ verify_heapam(PG_FUNCTION_ARGS)
            break;
    }
 
+   /* Ensure that the stream is completely read */
+   Assert(read_stream_next_buffer(stream, NULL) == InvalidBuffer);
+   read_stream_end(stream);
+
    if (vmbuffer != InvalidBuffer)
        ReleaseBuffer(vmbuffer);
 
@@ -815,6 +862,42 @@ verify_heapam(PG_FUNCTION_ARGS)
    PG_RETURN_NULL();
 }
 
+/*
+ * Heap amcheck's read stream callback for getting the next unskippable block.
+ * This callback is only used when 'all-visible' or 'all-frozen' is provided
+ * as the skip option to verify_heapam(). With the default 'none',
+ * block_range_read_stream_cb() is used instead.
+ */
+static BlockNumber
+heapcheck_read_stream_next_unskippable(ReadStream *stream,
+                                      void *callback_private_data,
+                                      void *per_buffer_data)
+{
+   HeapCheckReadStreamData *p = callback_private_data;
+
+   /* Loops over [current_blocknum, last_exclusive) blocks */
+   for (BlockNumber i; (i = p->range.current_blocknum++) < p->range.last_exclusive;)
+   {
+       uint8       mapbits = visibilitymap_get_status(p->rel, i, p->vmbuffer);
+
+       if (p->skip_option == SKIP_PAGES_ALL_FROZEN)
+       {
+           if ((mapbits & VISIBILITYMAP_ALL_FROZEN) != 0)
+               continue;
+       }
+
+       if (p->skip_option == SKIP_PAGES_ALL_VISIBLE)
+       {
+           if ((mapbits & VISIBILITYMAP_ALL_VISIBLE) != 0)
+               continue;
+       }
+
+       return i;
+   }
+
+   return InvalidBlockNumber;
+}
+
 /*
  * Shared internal implementation for report_corruption and
  * report_toast_corruption.

diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 9442a4841aa3b13e1323576a49960ca8036513a9..1279b69422a59e402b160777928915ed5e1fb181 100644 (file)
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1169,6 +1169,7 @@ HeadlineJsonState
 HeadlineParsedText
 HeadlineWordEntry
 HeapCheckContext
+HeapCheckReadStreamData
 HeapPageFreeze
 HeapScanDesc
 HeapTuple