Fix COPY FREEZE with CLOBBER_CACHE_ALWAYS

This adds code omitted from commit 7db0cd2145 by accident, which had
two consequences. Firstly, only rows inserted by heap_multi_insert were
frozen as expected when running COPY FREEZE, while heap_insert left
rows unfrozen. That however includes rows in TOAST tables, so a lot of
data might have been left unfrozen. Secondly, page might have been left
partially empty after relcache invalidation.

This addresses both of those issues.

Discussion: https://postgr.es/m/CABOikdN-ptGv0mZntrK2Q8OtfUuAjqaYMGmkdU1dCKFtUxVLrg@mail.gmail.com

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index faffbb18658d7a7b2fe1929686352271a3d52137..d107e14690c7e50fcd52fdc07c1b6bfd6170c63b 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1880,8 +1880,12 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
    TransactionId xid = GetCurrentTransactionId();
    HeapTuple   heaptup;
    Buffer      buffer;
+   Page        page = NULL;
    Buffer      vmbuffer = InvalidBuffer;
+   bool        starting_with_empty_page;
    bool        all_visible_cleared = false;
+   bool        all_frozen_set = false;
+   uint8       vmstatus = 0;
 
    /*
     * Fill in tuple header fields and toast the tuple if necessary.
@@ -1894,11 +1898,36 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
    /*
     * Find buffer to insert this tuple into.  If the page is all visible,
     * this will also pin the requisite visibility map page.
+    *
+    * Also pin visibility map page if COPY FREEZE inserts tuples into an
+    * empty page. See all_frozen_set below.
     */
    buffer = RelationGetBufferForTuple(relation, heaptup->t_len,
                                       InvalidBuffer, options, bistate,
                                       &vmbuffer, NULL);
 
+
+   /*
+    * If we're inserting frozen entry into an empty page,
+    * set visibility map bits and PageAllVisible() hint.
+    *
+    * If we're inserting frozen entry into already all_frozen page,
+    * preserve this state.
+    */
+   if (options & HEAP_INSERT_FROZEN)
+   {
+       page = BufferGetPage(buffer);
+
+       starting_with_empty_page = PageGetMaxOffsetNumber(page) == 0;
+
+       if (visibilitymap_pin_ok(BufferGetBlockNumber(buffer), vmbuffer))
+           vmstatus = visibilitymap_get_status(relation,
+                                BufferGetBlockNumber(buffer), &vmbuffer);
+
+       if ((starting_with_empty_page || vmstatus & VISIBILITYMAP_ALL_FROZEN))
+           all_frozen_set = true;
+   }
+
    /*
     * We're about to do the actual insert -- but check for conflict first, to
     * avoid possibly having to roll back work we've just done.
@@ -1922,7 +1951,14 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
    RelationPutHeapTuple(relation, buffer, heaptup,
                         (options & HEAP_INSERT_SPECULATIVE) != 0);
 
-   if (PageIsAllVisible(BufferGetPage(buffer)))
+   /*
+    * If the page is all visible, need to clear that, unless we're only
+    * going to add further frozen rows to it.
+    *
+    * If we're only adding already frozen rows to a page that was empty or
+    * marked as all visible, mark it as all-visible.
+    */
+   if (PageIsAllVisible(BufferGetPage(buffer)) && !(options & HEAP_INSERT_FROZEN))
    {
        all_visible_cleared = true;
        PageClearAllVisible(BufferGetPage(buffer));
@@ -1930,6 +1966,13 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
                            ItemPointerGetBlockNumber(&(heaptup->t_self)),
                            vmbuffer, VISIBILITYMAP_VALID_BITS);
    }
+   else if (all_frozen_set)
+   {
+       /* We only ever set all_frozen_set after reading the page. */
+       Assert(page);
+
+       PageSetAllVisible(page);
+   }
 
    /*
     * XXX Should we set PageSetPrunable on this page ?
@@ -1977,6 +2020,8 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
        xlrec.flags = 0;
        if (all_visible_cleared)
            xlrec.flags |= XLH_INSERT_ALL_VISIBLE_CLEARED;
+       if (all_frozen_set)
+           xlrec.flags = XLH_INSERT_ALL_FROZEN_SET;
        if (options & HEAP_INSERT_SPECULATIVE)
            xlrec.flags |= XLH_INSERT_IS_SPECULATIVE;
        Assert(ItemPointerGetBlockNumber(&heaptup->t_self) == BufferGetBlockNumber(buffer));
@@ -2025,6 +2070,29 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 
    END_CRIT_SECTION();
 
+   /*
+    * If we've frozen everything on the page, update the visibilitymap.
+    * We're already holding pin on the vmbuffer.
+    *
+    * No need to update the visibilitymap if it had all_frozen bit set
+    * before this insertion.
+    */
+   if (all_frozen_set && ((vmstatus & VISIBILITYMAP_ALL_FROZEN) == 0))
+   {
+       Assert(PageIsAllVisible(page));
+       Assert(visibilitymap_pin_ok(BufferGetBlockNumber(buffer), vmbuffer));
+
+       /*
+        * It's fine to use InvalidTransactionId here - this is only used
+        * when HEAP_INSERT_FROZEN is specified, which intentionally
+        * violates visibility rules.
+        */
+       visibilitymap_set(relation, BufferGetBlockNumber(buffer), buffer,
+                           InvalidXLogRecPtr, vmbuffer,
+                           InvalidTransactionId,
+                           VISIBILITYMAP_ALL_VISIBLE | VISIBILITYMAP_ALL_FROZEN);
+   }
+
    UnlockReleaseBuffer(buffer);
    if (vmbuffer != InvalidBuffer)
        ReleaseBuffer(vmbuffer);
@@ -8708,6 +8776,10 @@ heap_xlog_insert(XLogReaderState *record)
    ItemPointerSetBlockNumber(&target_tid, blkno);
    ItemPointerSetOffsetNumber(&target_tid, xlrec->offnum);
 
+   /* check that the mutually exclusive flags are not both set */
+   Assert (!((xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED) &&
+             (xlrec->flags & XLH_INSERT_ALL_FROZEN_SET)));
+
    /*
     * The visibility map may need to be fixed even if the heap page is
     * already up-to-date.
@@ -8925,6 +8997,10 @@ heap_xlog_multi_insert(XLogReaderState *record)
        if (xlrec->flags & XLH_INSERT_ALL_VISIBLE_CLEARED)
            PageClearAllVisible(page);
 
+       /* XLH_INSERT_ALL_FROZEN_SET implies that all tuples are visible */
+       if (xlrec->flags & XLH_INSERT_ALL_FROZEN_SET)
+           PageSetAllVisible(page);
+
        MarkBufferDirty(buffer);
    }
    if (BufferIsValid(buffer))

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 2d23b3ef7194516d5494e62e0aae595f67da6b45..fb7ad0bab47ac4049586bf92a787410f5dea7e39 100644 (file)
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -396,19 +396,19 @@ RelationGetBufferForTuple(Relation relation, Size len,
         * target.
         */
        targetBlock = GetPageWithFreeSpace(relation, len + saveFreeSpace);
+   }
 
-       /*
-        * If the FSM knows nothing of the rel, try the last page before we
-        * give up and extend.  This avoids one-tuple-per-page syndrome during
-        * bootstrapping or in a recently-started system.
-        */
-       if (targetBlock == InvalidBlockNumber)
-       {
-           BlockNumber nblocks = RelationGetNumberOfBlocks(relation);
+   /*
+    * If the FSM knows nothing of the rel, try the last page before we
+    * give up and extend.  This avoids one-tuple-per-page syndrome during
+    * bootstrapping or in a recently-started system.
+    */
+   if (targetBlock == InvalidBlockNumber)
+   {
+       BlockNumber nblocks = RelationGetNumberOfBlocks(relation);
 
-           if (nblocks > 0)
-               targetBlock = nblocks - 1;
-       }
+       if (nblocks > 0)
+           targetBlock = nblocks - 1;
    }
 
 loop: