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: