Fix PageAddItem BRIN bug

BRIN was relying on the ability to remove a tuple from an index page,
then putting another tuple in the same line pointer.  But PageAddItem
refuses to add a tuple beyond the first free item past the last used
item, and in particular, it rejects an attempt to add an item to an
empty page anywhere other than the first line pointer.  PageAddItem
issues a WARNING and indicates to the caller that it failed, which in
turn causes the BRIN calling code to issue a PANIC, so the whole
sequence looks like this:
	WARNING:  specified item offset is too large
	PANIC:  failed to add BRIN tuple

To fix, create a new function PageAddItemExtended which is like
PageAddItem except that the two boolean arguments become a flags bitmap;
the "overwrite" and "is_heap" boolean flags in PageAddItem become
PAI_OVERWITE and PAI_IS_HEAP flags in the new function, and a new flag
PAI_ALLOW_FAR_OFFSET enables the behavior required by BRIN.
PageAddItem() retains its original signature, for compatibility with
third-party modules (other callers in core code are not modified,
either).

Also, in the belt-and-suspenders spirit, I added a new sanity check in
brinGetTupleForHeapBlock to raise an error if an TID found in the revmap
is not marked as live by the page header.  This causes it to react with
"ERROR: corrupted BRIN index" to the bug at hand, rather than a hard
crash.

Backpatch to 9.5.

Bug reported by Andreas Seltenreich as detected by his handy sqlsmith
fuzzer.
Discussion: https://www.postgresql.org/message-id/[email protected]

diff --git a/src/backend/access/brin/brin_pageops.c b/src/backend/access/brin/brin_pageops.c
index f876f62cbbdf36239ec67e42fc698cf1b983f7a7..b627e3b34e8c040dbbb557b366e21b3ac4785fcd 100644 (file)
--- a/src/backend/access/brin/brin_pageops.c
+++ b/src/backend/access/brin/brin_pageops.c
@@ -179,8 +179,8 @@ brin_doupdate(Relation idxrel, BlockNumber pagesPerRange,
 
        START_CRIT_SECTION();
        PageIndexDeleteNoCompact(oldpage, &oldoff, 1);
-       if (PageAddItem(oldpage, (Item) newtup, newsz, oldoff, true,
-                       false) == InvalidOffsetNumber)
+       if (PageAddItemExtended(oldpage, (Item) newtup, newsz, oldoff,
+               PAI_OVERWRITE | PAI_ALLOW_FAR_OFFSET) == InvalidOffsetNumber)
            elog(ERROR, "failed to add BRIN tuple");
        MarkBufferDirty(oldbuf);
 

diff --git a/src/backend/access/brin/brin_revmap.c b/src/backend/access/brin/brin_revmap.c
index 6ddcfda0a6ca0c24359b90fa364e135019efd7bc..d16d87b9af75354c2fbaad31c206d375f083a6b2 100644 (file)
--- a/src/backend/access/brin/brin_revmap.c
+++ b/src/backend/access/brin/brin_revmap.c
@@ -266,6 +266,10 @@ brinGetTupleForHeapBlock(BrinRevmap *revmap, BlockNumber heapBlk,
        /* If we land on a revmap page, start over */
        if (BRIN_IS_REGULAR_PAGE(page))
        {
+           if (*off > PageGetMaxOffsetNumber(page))
+               ereport(ERROR,
+                       (errcode(ERRCODE_INDEX_CORRUPTED),
+                        errmsg_internal("corrupted BRIN index: inconsistent range map")));
            lp = PageGetItemId(page, *off);
            if (ItemIdIsUsed(lp))
            {

diff --git a/src/backend/access/brin/brin_xlog.c b/src/backend/access/brin/brin_xlog.c
index 760f0daf024df4fd14af409d4ea08d1193f8da12..1ebeec78acec3524c0fb6f7cdbf6b43d60519cf7 100644 (file)
--- a/src/backend/access/brin/brin_xlog.c
+++ b/src/backend/access/brin/brin_xlog.c
@@ -193,7 +193,8 @@ brin_xlog_samepage_update(XLogReaderState *record)
            elog(PANIC, "brin_xlog_samepage_update: invalid max offset number");
 
        PageIndexDeleteNoCompact(page, &offnum, 1);
-       offnum = PageAddItem(page, (Item) brintuple, tuplen, offnum, true, false);
+       offnum = PageAddItemExtended(page, (Item) brintuple, tuplen, offnum,
+                                    PAI_OVERWRITE | PAI_ALLOW_FAR_OFFSET);
        if (offnum == InvalidOffsetNumber)
            elog(PANIC, "brin_xlog_samepage_update: failed to add tuple");
 

diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index df77bb2f5c0825069a966fd82b1f0f0a416ba784..779bf7c48d6b3ea76044e9971da2789b308320e4 100644 (file)
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -153,12 +153,13 @@ PageIsVerified(Page page, BlockNumber blkno)
 
 
 /*
- * PageAddItem
+ * PageAddItemExtended
  *
- * Add an item to a page.  Return value is offset at which it was
- * inserted, or InvalidOffsetNumber if there's not room to insert.
+ * Add an item to a page.  Return value is the offset at which it was
+ * inserted, or InvalidOffsetNumber if the item is not inserted for any
+ * reason.  A WARNING is issued indicating the reason for the refusal.
  *
- * If overwrite is true, we just store the item at the specified
+ * If flag PAI_OVERWRITE is set, we just store the item at the specified
  * offsetNumber (which must be either a currently-unused item pointer,
  * or one past the last existing item).  Otherwise,
  * if offsetNumber is valid and <= current max offset in the page,
@@ -167,18 +168,20 @@ PageIsVerified(Page page, BlockNumber blkno)
  * If offsetNumber is not valid, then assign one by finding the first
  * one that is both unused and deallocated.
  *
- * If is_heap is true, we enforce that there can't be more than
+ * If flag PAI_IS_HEAP is set, we enforce that there can't be more than
  * MaxHeapTuplesPerPage line pointers on the page.
  *
+ * If flag PAI_ALLOW_FAR_OFFSET is not set, we disallow placing items
+ * beyond one past the last existing item.
+ *
  * !!! EREPORT(ERROR) IS DISALLOWED HERE !!!
  */
 OffsetNumber
-PageAddItem(Page page,
-           Item item,
-           Size size,
-           OffsetNumber offsetNumber,
-           bool overwrite,
-           bool is_heap)
+PageAddItemExtended(Page page,
+                   Item item,
+                   Size size,
+                   OffsetNumber offsetNumber,
+                   int flags)
 {
    PageHeader  phdr = (PageHeader) page;
    Size        alignedSize;
@@ -209,7 +212,7 @@ PageAddItem(Page page,
    if (OffsetNumberIsValid(offsetNumber))
    {
        /* yes, check it */
-       if (overwrite)
+       if ((flags & PAI_OVERWRITE) != 0)
        {
            if (offsetNumber < limit)
            {
@@ -257,13 +260,18 @@ PageAddItem(Page page,
        }
    }
 
-   if (offsetNumber > limit)
+   /*
+    * Reject placing items beyond the first unused line pointer, unless
+    * caller asked for that behavior specifically.
+    */
+   if ((flags & PAI_ALLOW_FAR_OFFSET) == 0 && offsetNumber > limit)
    {
        elog(WARNING, "specified item offset is too large");
        return InvalidOffsetNumber;
    }
 
-   if (is_heap && offsetNumber > MaxHeapTuplesPerPage)
+   /* Reject placing items beyond heap boundary, if heap */
+   if ((flags & PAI_IS_HEAP) != 0 && offsetNumber > MaxHeapTuplesPerPage)
    {
        elog(WARNING, "can't put more than MaxHeapTuplesPerPage items in a heap page");
        return InvalidOffsetNumber;
@@ -275,7 +283,10 @@ PageAddItem(Page page,
     * Note: do arithmetic as signed ints, to avoid mistakes if, say,
     * alignedSize > pd_upper.
     */
-   if (offsetNumber == limit || needshuffle)
+   if ((flags & PAI_ALLOW_FAR_OFFSET) != 0)
+       lower = Max(phdr->pd_lower,
+                   SizeOfPageHeaderData + sizeof(ItemIdData) * offsetNumber);
+   else if (offsetNumber == limit || needshuffle)
        lower = phdr->pd_lower + sizeof(ItemIdData);
    else
        lower = phdr->pd_lower;
@@ -323,6 +334,27 @@ PageAddItem(Page page,
    return offsetNumber;
 }
 
+/*
+ * PageAddItem
+ *
+ * Add an item to a page.  Return value is offset at which it was
+ * inserted, or InvalidOffsetNumber if the item is not inserted for
+ * any reason.
+ *
+ * Passing the 'overwrite' and 'is_heap' parameters as true causes the
+ * PAI_OVERWRITE and PAI_IS_HEAP flags to be set, respectively.
+ *
+ * !!! EREPORT(ERROR) IS DISALLOWED HERE !!!
+ */
+OffsetNumber
+PageAddItem(Page page, Item item, Size size, OffsetNumber offsetNumber,
+           bool overwrite, bool is_heap)
+{
+   return PageAddItemExtended(page, item, size, offsetNumber,
+                              overwrite ? PAI_OVERWRITE : 0 |
+                              is_heap ? PAI_IS_HEAP : 0);
+}
+
 /*
  * PageGetTempPage
  *     Get a temporary page in local memory for special processing.

diff --git a/src/include/storage/bufpage.h b/src/include/storage/bufpage.h
index a2f78ee56ce9e8c045b0a6b3bfd639a92fa8959d..78408e79e0446625820ca36cd81ea6d58669b859 100644 (file)
--- a/src/include/storage/bufpage.h
+++ b/src/include/storage/bufpage.h
@@ -390,11 +390,16 @@ do { \
  *     extern declarations
  * ----------------------------------------------------------------
  */
+#define PAI_OVERWRITE          (1 << 0)
+#define PAI_IS_HEAP                (1 << 1)
+#define PAI_ALLOW_FAR_OFFSET   (1 << 2)
 
 extern void PageInit(Page page, Size pageSize, Size specialSize);
 extern bool PageIsVerified(Page page, BlockNumber blkno);
 extern OffsetNumber PageAddItem(Page page, Item item, Size size,
            OffsetNumber offsetNumber, bool overwrite, bool is_heap);
+extern OffsetNumber PageAddItemExtended(Page page, Item item, Size size,
+                   OffsetNumber offsetNumber, int flags);
 extern Page PageGetTempPage(Page page);
 extern Page PageGetTempPageCopy(Page page);
 extern Page PageGetTempPageCopySpecial(Page page);