improve-compactify_tuples-6.patch

diff --git a/src/backend/storage/page/bufpage.c b/src/backend/storage/page/bufpage.c
index 41642eb..1af1b85 100644
*** a/src/backend/storage/page/bufpage.c
--- b/src/backend/storage/page/bufpage.c
***************
*** 18,23 ****
--- 18,24 ----
  #include "access/itup.h"
  #include "access/xlog.h"
  #include "storage/checksum.h"
+ #include "utils/inline_sort.h"
  #include "utils/memdebug.h"
  #include "utils/memutils.h"
  
*************** typedef struct itemIdSortData
*** 425,439 ****
  } itemIdSortData;
  typedef itemIdSortData *itemIdSort;
  
! static int
  itemoffcompare(const void *itemidp1, const void *itemidp2)
  {
- 	/* Sort in decreasing itemoff order */
  	return ((itemIdSort) itemidp2)->itemoff -
  		((itemIdSort) itemidp1)->itemoff;
  }
  
  /*
   * After removing or marking some line pointers unused, move the tuples to
   * remove the gaps caused by the removed items.
   */
--- 426,542 ----
  } itemIdSortData;
  typedef itemIdSortData *itemIdSort;
  
! /* Comparator for sorting in decreasing itemoff order */
! static inline int
  itemoffcompare(const void *itemidp1, const void *itemidp2)
  {
  	return ((itemIdSort) itemidp2)->itemoff -
  		((itemIdSort) itemidp1)->itemoff;
  }
  
  /*
+  * Sort an array of itemIdSort's on itemoff, descending.
+  *
+  * This uses Shell sort.  Given that array is small and itemoffcompare
+  * can be inlined, it is much faster than general-purpose qsort.
+  */
+ static void
+ sort_itemIds_small(itemIdSort itemidbase, int nitems)
+ {
+ 	pg_shell_sort(itemIdSortData, itemidbase, nitems, itemoffcompare);
+ }
+ 
+ /*
+  * Sort an array of itemIdSort's on itemoff, descending.
+  *
+  * This uses bucket sort:
+  * - single pass of stable prefix sort on high 8 bits of itemoffs
+  * - then insertion sort on buckets larger than 1 element
+  */
+ static void
+ sort_itemIds(itemIdSort itemidbase, int nitems)
+ {
+ 	/* number of buckets to use: */
+ #define NSPLIT 256
+ 	/* divisor to scale input values into 0..NSPLIT-1: */
+ #define PREFDIV (BLCKSZ / NSPLIT)
+ 	/* per-bucket counts; we need two extra elements, see below */
+ 	uint16		count[NSPLIT + 2];
+ 	itemIdSortData copy[Max(MaxIndexTuplesPerPage, MaxHeapTuplesPerPage)];
+ 	int			i,
+ 				max,
+ 				total,
+ 				pos,
+ 				highbits;
+ 
+ 	Assert(nitems <= lengthof(copy));
+ 
+ 	/*
+ 	 * Count how many items in each bucket.  We assume all itemoff values are
+ 	 * less than BLCKSZ, therefore dividing by PREFDIV gives a value less than
+ 	 * NSPLIT.
+ 	 */
+ 	memset(count, 0, sizeof(count));
+ 	for (i = 0; i < nitems; i++)
+ 	{
+ 		highbits = itemidbase[i].itemoff / PREFDIV;
+ 		count[highbits]++;
+ 	}
+ 
+ 	/*
+ 	 * Now convert counts to bucket position info, placing the buckets in
+ 	 * decreasing order.  After this loop, count[k+1] is start of bucket k
+ 	 * (for 0 <= k < NSPLIT), count[k] is end+1 of bucket k, and therefore
+ 	 * count[k] - count[k+1] is length of bucket k.
+ 	 *
+ 	 * Also detect whether any buckets have more than one element.  For this
+ 	 * purpose, "max" is set to the OR of all the counts (not really the max).
+ 	 */
+ 	max = total = count[NSPLIT - 1];
+ 	for (i = NSPLIT - 2; i >= 0; i--)
+ 	{
+ 		max |= count[i];
+ 		total += count[i];
+ 		count[i] = total;
+ 	}
+ 	Assert(count[0] == nitems);
+ 
+ 	/*
+ 	 * Now copy the data to be sorted into appropriate positions in the copy[]
+ 	 * array.  We increment each bucket-start pointer as we insert data into
+ 	 * its bucket; hence, after this loop count[k+1] is the end+1 of bucket k,
+ 	 * count[k+2] is the start of bucket k, and count[k+1] - count[k+2] is the
+ 	 * length of bucket k.
+ 	 */
+ 	for (i = 0; i < nitems; i++)
+ 	{
+ 		highbits = itemidbase[i].itemoff / PREFDIV;
+ 		pos = count[highbits + 1]++;
+ 		copy[pos] = itemidbase[i];
+ 	}
+ 	Assert(count[1] == nitems);
+ 
+ 	/*
+ 	 * If any buckets are larger than 1 item, we must sort them.  They should
+ 	 * be small enough to make insertion sort effective.
+ 	 */
+ 	if (max > 1)
+ 	{
+ 		/* i is bucket number plus 1 */
+ 		for (i = NSPLIT; i > 0; i--)
+ 		{
+ 			pg_insertion_sort(itemIdSortData,
+ 							  copy + count[i + 1],
+ 							  count[i] - count[i + 1],
+ 							  itemoffcompare);
+ 		}
+ 	}
+ 
+ 	/* And transfer the sorted data back to the caller */
+ 	memcpy(itemidbase, copy, sizeof(itemIdSortData) * nitems);
+ }
+ 
+ /*
   * After removing or marking some line pointers unused, move the tuples to
   * remove the gaps caused by the removed items.
   */
*************** compactify_tuples(itemIdSort itemidbase,
*** 445,452 ****
  	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++)
--- 548,558 ----
  	int			i;
  
  	/* sort itemIdSortData array into decreasing itemoff order */
! 	/* empirically, bucket sort is worth the trouble above 48 items */
! 	if (nitems > 48)
! 		sort_itemIds(itemidbase, nitems);
! 	else
! 		sort_itemIds_small(itemidbase, nitems);
  
  	upper = phdr->pd_special;
  	for (i = 0; i < nitems; i++)
diff --git a/src/include/utils/inline_sort.h b/src/include/utils/inline_sort.h
index ...c97a248 .
*** a/src/include/utils/inline_sort.h
--- b/src/include/utils/inline_sort.h
***************
*** 0 ****
--- 1,88 ----
+ /*-------------------------------------------------------------------------
+  *
+  * inline_sort.h
+  *	  Macros to perform specialized types of sorts.
+  *
+  *
+  * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
+  * Portions Copyright (c) 1994, Regents of the University of California
+  *
+  * src/include/utils/inline_sort.h
+  *
+  *-------------------------------------------------------------------------
+  */
+ #ifndef INLINE_SORT_H
+ #define INLINE_SORT_H
+ 
+ /*
+  * pg_shell_sort - sort for small arrays with inlinable comparator.
+  *
+  * This is best used with arrays smaller than 200 elements, and could be
+  * safely used with up to 1000 elements.  But it degrades fast after that.
+  *
+  * Since this is implemented as a macro it can be optimized together with
+  * comparison function; using a macro or inlinable function is recommended.
+  *
+  * Arguments:
+  *	 elem_t - type of array elements (for declaring temporary variables)
+  *	 array	- pointer to elements to be sorted
+  *	 nitems - number of elements to be sorted
+  *	 cmp	- comparison function that accepts addresses of 2 elements
+  *			  (same API as qsort comparison function).
+  * cmp argument should be a function or macro name.
+  * array and nitems arguments are evaluated only once.
+  *
+  * This uses Shellsort (see e.g. wikipedia's entry), with gaps selected as
+  * "gap(i) = smallest prime number below e^i".  These are close to the gaps
+  * recommended by Incerpi & Sedwick, but look to be better on average.
+  */
+ #define pg_shell_sort(elem_t, array, nitems, cmp) \
+ 	do { \
+ 		elem_t *_arr = (array); \
+ 		int		_n = (nitems); \
+ 		static const int _offsets[] = {401, 139, 53, 19, 7, 3}; \
+ 		int		_noff; \
+ 		for (_noff = 0; _noff < lengthof(_offsets); _noff++) \
+ 		{ \
+ 			int		_off = _offsets[_noff]; \
+ 			pg_shell_sort_pass(elem_t, cmp, _off, _arr, _n); \
+ 		} \
+ 		pg_shell_sort_pass(elem_t, cmp, 1, _arr, _n); \
+ 	} while (0)
+ 
+ /*
+  * pg_insertion_sort - plain insertion sort.
+  * Useful for very small array, or if array was almost sorted already.
+  * Same API as pg_shell_sort.
+  */
+ #define pg_insertion_sort(elem_t, array, nitems, cmp) \
+ 	do { \
+ 		elem_t *_arr = (array); \
+ 		int		_n = (nitems); \
+ 		pg_shell_sort_pass(elem_t, cmp, 1, _arr, _n); \
+ 	} while (0)
+ 
+ /*
+  * One pass of Shellsort: simple insertion sort of the subset of entries
+  * at stride "off".  Not intended to be used outside of above macros.
+  */
+ #define pg_shell_sort_pass(elem_t, cmp, off, _arr, _n) \
+ 	do { \
+ 		int		_i; \
+ 		for (_i = off; _i < _n; _i += off) \
+ 		{ \
+ 			if (cmp(_arr + _i - off, _arr + _i) > 0) \
+ 			{ \
+ 				elem_t	_temp = _arr[_i]; \
+ 				int		_j = _i; \
+ 				do \
+ 				{ \
+ 					_arr[_j] = _arr[_j - off]; \
+ 					_j -= off; \
+ 				} while (_j >= off && cmp(_arr + _j - off, &_temp) > 0); \
+ 				_arr[_j] = _temp; \
+ 			} \
+ 		} \
+ 	} while (0)
+ 
+ #endif							/* INLINE_SORT_H */