Refactor page compactifying code.

The logic to compact away removed tuples from page was duplicated with
small differences in PageRepairFragmentation, PageIndexMultiDelete, and
PageIndexDeleteNoCompact. Put it into a common function.

Reviewed by Peter Geoghegan.

diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index c0095c7ba18111609858bfe06bd0481f57e3c3a9..61e7437f8ffea19964ce3cca0cc27f7b8c093c6e 100644 (file)
--- a/src/backend/storage/page/bufpage.c
+++ b/src/backend/storage/page/bufpage.c
@@ -404,10 +404,9 @@ PageRestoreTempPage(Page tempPage, Page oldPage)
  */
 typedef struct itemIdSortData
 {
-   int         offsetindex;    /* linp array index */
-   int         itemoff;        /* page offset of item data */
-   Size        alignedlen;     /* MAXALIGN(item data len) */
-   ItemIdData  olditemid;      /* used only in PageIndexMultiDelete */
+   uint16      offsetindex;    /* linp array index */
+   int16       itemoff;        /* page offset of item data */
+   uint16      alignedlen;     /* MAXALIGN(item data len) */
 } itemIdSortData;
 typedef itemIdSortData *itemIdSort;
 
@@ -419,6 +418,38 @@ itemoffcompare(const void *itemidp1, const void *itemidp2)
        ((itemIdSort) itemidp1)->itemoff;
 }
 
+/*
+ * After removing or marking some line pointers unused, move the tuples to
+ * remove the gaps caused by the removed items.
+ */
+static void
+compactify_tuples(itemIdSort itemidbase, int nitems, Page page)
+{
+   PageHeader  phdr = (PageHeader) page;
+   Offset      upper;
+   int         i;
+
+   /* sort itemIdSortData array into decreasing itemoff order */
+   qsort((char *) itemidbase, nitems, sizeof(itemIdSortData),
+         itemoffcompare);
+
+   upper = phdr->pd_special;
+   for (i = 0; i < nitems; i++)
+   {
+       itemIdSort  itemidptr = &itemidbase[i];
+       ItemId      lp;
+
+       lp = PageGetItemId(page, itemidptr->offsetindex + 1);
+       upper -= itemidptr->alignedlen;
+       memmove((char *) page + upper,
+               (char *) page + itemidptr->itemoff,
+               itemidptr->alignedlen);
+       lp->lp_off = upper;
+   }
+
+   phdr->pd_upper = upper;
+}
+
 /*
  * PageRepairFragmentation
  *
@@ -441,7 +472,6 @@ PageRepairFragmentation(Page page)
                nunused;
    int         i;
    Size        totallen;
-   Offset      upper;
 
    /*
     * It's worth the trouble to be more paranoid here than in most places,
@@ -515,24 +545,7 @@ PageRepairFragmentation(Page page)
               errmsg("corrupted item lengths: total %u, available space %u",
                      (unsigned int) totallen, pd_special - pd_lower)));
 
-       /* sort itemIdSortData array into decreasing itemoff order */
-       qsort((char *) itemidbase, nstorage, sizeof(itemIdSortData),
-             itemoffcompare);
-
-       /* compactify page */
-       upper = pd_special;
-
-       for (i = 0, itemidptr = itemidbase; i < nstorage; i++, itemidptr++)
-       {
-           lp = PageGetItemId(page, itemidptr->offsetindex + 1);
-           upper -= itemidptr->alignedlen;
-           memmove((char *) page + upper,
-                   (char *) page + itemidptr->itemoff,
-                   itemidptr->alignedlen);
-           lp->lp_off = upper;
-       }
-
-       ((PageHeader) page)->pd_upper = upper;
+       compactify_tuples(itemidbase, nstorage, page);
    }
 
    /* Set hint bit for PageAddItem */
@@ -782,13 +795,12 @@ PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
    Offset      pd_upper = phdr->pd_upper;
    Offset      pd_special = phdr->pd_special;
    itemIdSortData itemidbase[MaxIndexTuplesPerPage];
+   ItemIdData  newitemids[MaxIndexTuplesPerPage];
    itemIdSort  itemidptr;
    ItemId      lp;
    int         nline,
                nused;
-   int         i;
    Size        totallen;
-   Offset      upper;
    Size        size;
    unsigned    offset;
    int         nextitm;
@@ -857,9 +869,9 @@ PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
        {
            itemidptr->offsetindex = nused;     /* where it will go */
            itemidptr->itemoff = offset;
-           itemidptr->olditemid = *lp;
            itemidptr->alignedlen = MAXALIGN(size);
            totallen += itemidptr->alignedlen;
+           newitemids[nused] = *lp;
            itemidptr++;
            nused++;
        }
@@ -875,26 +887,15 @@ PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
               errmsg("corrupted item lengths: total %u, available space %u",
                      (unsigned int) totallen, pd_special - pd_lower)));
 
-   /* sort itemIdSortData array into decreasing itemoff order */
-   qsort((char *) itemidbase, nused, sizeof(itemIdSortData),
-         itemoffcompare);
-
-   /* compactify page and install new itemids */
-   upper = pd_special;
-
-   for (i = 0, itemidptr = itemidbase; i < nused; i++, itemidptr++)
-   {
-       lp = PageGetItemId(page, itemidptr->offsetindex + 1);
-       upper -= itemidptr->alignedlen;
-       memmove((char *) page + upper,
-               (char *) page + itemidptr->itemoff,
-               itemidptr->alignedlen);
-       *lp = itemidptr->olditemid;
-       lp->lp_off = upper;
-   }
-
+   /*
+    * Looks good. Overwrite the line pointers with the copy, from which we've
+    * removed all the unused items.
+    */
+   memcpy(phdr->pd_linp, newitemids, nused * sizeof(ItemIdData));
    phdr->pd_lower = SizeOfPageHeaderData + nused * sizeof(ItemIdData);
-   phdr->pd_upper = upper;
+
+   /* and compactify the tuple data */
+   compactify_tuples(itemidbase, nused, page);
 }
 
 /*
@@ -1000,7 +1001,6 @@ PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
        itemIdSort  itemidptr;
        int         i;
        Size        totallen;
-       Offset      upper;
 
        /*
         * Scan the page taking note of each item that we need to preserve.
@@ -1012,7 +1012,8 @@ PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
         */
        itemidptr = itemidbase;
        totallen = 0;
-       for (i = 0; i < nline; i++, itemidptr++)
+       PageClearHasFreeLinePointers(page);
+       for (i = 0; i < nline; i++)
        {
            ItemId      lp;
 
@@ -1024,13 +1025,15 @@ PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
                itemidptr->itemoff = ItemIdGetOffset(lp);
                itemidptr->alignedlen = MAXALIGN(ItemIdGetLength(lp));
                totallen += itemidptr->alignedlen;
+               itemidptr++;
            }
            else
            {
-               itemidptr->itemoff = 0;
-               itemidptr->alignedlen = 0;
+               PageSetHasFreeLinePointers(page);
+               ItemIdSetUnused(lp);
            }
        }
+       nline = itemidptr - itemidbase;
        /* By here, there are exactly nline elements in itemidbase array */
 
        if (totallen > (Size) (pd_special - pd_lower))
@@ -1039,38 +1042,11 @@ PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
                     errmsg("corrupted item lengths: total %u, available space %u",
                            (unsigned int) totallen, pd_special - pd_lower)));
 
-       /* sort itemIdSortData array into decreasing itemoff order */
-       qsort((char *) itemidbase, nline, sizeof(itemIdSortData),
-             itemoffcompare);
-
        /*
         * Defragment the data areas of each tuple, being careful to preserve
         * each item's position in the linp array.
         */
-       upper = pd_special;
-       PageClearHasFreeLinePointers(page);
-       for (i = 0, itemidptr = itemidbase; i < nline; i++, itemidptr++)
-       {
-           ItemId      lp;
-
-           lp = PageGetItemId(page, itemidptr->offsetindex + 1);
-           if (itemidptr->alignedlen == 0)
-           {
-               PageSetHasFreeLinePointers(page);
-               ItemIdSetUnused(lp);
-               continue;
-           }
-           upper -= itemidptr->alignedlen;
-           memmove((char *) page + upper,
-                   (char *) page + itemidptr->itemoff,
-                   itemidptr->alignedlen);
-           lp->lp_off = upper;
-           /* lp_flags and lp_len remain the same as originally */
-       }
-
-       /* Set the new page limits */
-       phdr->pd_upper = upper;
-       phdr->pd_lower = SizeOfPageHeaderData + i * sizeof(ItemIdData);
+       compactify_tuples(itemidbase, nline, page);
    }
 }