Max compact as an FSM strategy

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index ae2e2ce37a..7f996933ad 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -269,6 +269,17 @@ RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
 	FreeSpaceMapVacuumRange(relation, firstBlock, blockNum + 1);
 }
 
+/*
+ * RelationGetFreeSpaceStrategy
+ *
+ * Allow the FreeSpaceStrategy to be set dynamically using simple heuristics.
+ */
+static FreeSpaceStrategy
+RelationGetFreeSpaceStrategy(Relation rel)
+{
+	return FREESPACE_STRATEGY_MAX_CONCURRENCY;
+}
+
 /*
  * RelationGetBufferForTuple
  *
@@ -402,11 +413,13 @@ RelationGetBufferForTuple(Relation relation, Size len,
 
 	if (targetBlock == InvalidBlockNumber && use_fsm)
 	{
+		FreeSpaceStrategy fss = RelationGetFreeSpaceStrategy(relation);
+
 		/*
 		 * We have no cached target page, so ask the FSM for an initial
 		 * target.
 		 */
-		targetBlock = GetPageWithFreeSpace(relation, targetFreeSpace);
+		targetBlock = GetPageWithFreeSpaceExt(relation, targetFreeSpace, fss);
 	}
 
 	/*
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index 005def56dc..d626858205 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -107,8 +107,8 @@ static Size fsm_space_cat_to_avail(uint8 cat);
 
 /* workhorse functions for various operations */
 static int	fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
-							   uint8 newValue, uint8 minValue);
-static BlockNumber fsm_search(Relation rel, uint8 min_cat);
+							   uint8 newValue, uint8 minValue, FreeSpaceStrategy fss);
+static BlockNumber fsm_search(Relation rel, uint8 min_cat, FreeSpaceStrategy fss);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr,
 							 BlockNumber start, BlockNumber end,
 							 bool *eof);
@@ -134,7 +134,15 @@ GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 {
 	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
 
-	return fsm_search(rel, min_cat);
+	return fsm_search(rel, min_cat, FREESPACE_STRATEGY_MAX_CONCURRENCY);
+}
+
+BlockNumber
+GetPageWithFreeSpaceExt(Relation rel, Size spaceNeeded, FreeSpaceStrategy fss)
+{
+	uint8		min_cat = fsm_space_needed_to_cat(spaceNeeded);
+
+	return fsm_search(rel, min_cat, fss);
 }
 
 /*
@@ -149,6 +157,14 @@ GetPageWithFreeSpace(Relation rel, Size spaceNeeded)
 BlockNumber
 RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
 							  Size oldSpaceAvail, Size spaceNeeded)
+{
+	return RecordAndGetPageWithFreeSpaceExt(rel, oldPage, oldSpaceAvail, spaceNeeded,
+											FREESPACE_STRATEGY_MAX_CONCURRENCY);
+}
+
+BlockNumber
+RecordAndGetPageWithFreeSpaceExt(Relation rel, BlockNumber oldPage,
+							  Size oldSpaceAvail, Size spaceNeeded, FreeSpaceStrategy fss)
 {
 	int			old_cat = fsm_space_avail_to_cat(oldSpaceAvail);
 	int			search_cat = fsm_space_needed_to_cat(spaceNeeded);
@@ -159,7 +175,7 @@ RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
 	/* Get the location of the FSM byte representing the heap block */
 	addr = fsm_get_location(oldPage, &slot);
 
-	search_slot = fsm_set_and_search(rel, addr, slot, old_cat, search_cat);
+	search_slot = fsm_set_and_search(rel, addr, slot, old_cat, search_cat, fss);
 
 	/*
 	 * If fsm_set_and_search found a suitable new block, return that.
@@ -168,7 +184,7 @@ RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
 	if (search_slot != -1)
 		return fsm_get_heap_blk(addr, search_slot);
 	else
-		return fsm_search(rel, search_cat);
+		return fsm_search(rel, search_cat, fss);
 }
 
 /*
@@ -180,6 +196,14 @@ RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
  */
 void
 RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk, Size spaceAvail)
+{
+	return RecordPageWithFreeSpaceExt(rel, heapBlk, spaceAvail,
+									  FREESPACE_STRATEGY_MAX_CONCURRENCY);
+}
+
+void
+RecordPageWithFreeSpaceExt(Relation rel, BlockNumber heapBlk, Size spaceAvail,
+							FreeSpaceStrategy fss)
 {
 	int			new_cat = fsm_space_avail_to_cat(spaceAvail);
 	FSMAddress	addr;
@@ -188,7 +212,7 @@ RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk, Size spaceAvail)
 	/* Get the location of the FSM byte representing the heap block */
 	addr = fsm_get_location(heapBlk, &slot);
 
-	fsm_set_and_search(rel, addr, slot, new_cat, 0);
+	fsm_set_and_search(rel, addr, slot, new_cat, 0, fss);
 }
 
 /*
@@ -668,7 +692,7 @@ fsm_extend(Relation rel, BlockNumber fsm_nblocks)
  */
 static int
 fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
-				   uint8 newValue, uint8 minValue)
+				   uint8 newValue, uint8 minValue, FreeSpaceStrategy fss)
 {
 	Buffer		buf;
 	Page		page;
@@ -686,7 +710,7 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 	{
 		/* Search while we still hold the lock */
 		newslot = fsm_search_avail(buf, minValue,
-								   addr.level == FSM_BOTTOM_LEVEL,
+								   (fss == FREESPACE_STRATEGY_MAX_COMPACT ? false : (addr.level == FSM_BOTTOM_LEVEL)),
 								   true);
 	}
 
@@ -699,7 +723,7 @@ fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
  * Search the tree for a heap page with at least min_cat of free space
  */
 static BlockNumber
-fsm_search(Relation rel, uint8 min_cat)
+fsm_search(Relation rel, uint8 min_cat, FreeSpaceStrategy fss)
 {
 	int			restarts = 0;
 	FSMAddress	addr = FSM_ROOT_ADDRESS;
@@ -718,7 +742,7 @@ fsm_search(Relation rel, uint8 min_cat)
 		{
 			LockBuffer(buf, BUFFER_LOCK_SHARE);
 			slot = fsm_search_avail(buf, min_cat,
-									(addr.level == FSM_BOTTOM_LEVEL),
+									(fss == FREESPACE_STRATEGY_MAX_COMPACT ? false : (addr.level == FSM_BOTTOM_LEVEL)),
 									false);
 			if (slot == -1)
 				max_avail = fsm_get_max_avail(BufferGetPage(buf));
@@ -764,7 +788,7 @@ fsm_search(Relation rel, uint8 min_cat)
 			 * rarely, and will be fixed by the next vacuum.
 			 */
 			parent = fsm_get_parent(addr, &parentslot);
-			fsm_set_and_search(rel, parent, parentslot, max_avail, 0);
+			fsm_set_and_search(rel, parent, parentslot, max_avail, 0, fss);
 
 			/*
 			 * If the upper pages are badly out of date, we might need to loop
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index fcb080210d..3c7d5e6236 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -18,15 +18,48 @@
 #include "storage/relfilelocator.h"
 #include "utils/relcache.h"
 
+typedef enum FreeSpaceStrategy
+{
+	FREESPACE_STRATEGY_MAX_CONCURRENCY = 0,
+		/*
+		 * Each time we ask for a new block with freespace this will set
+		 * the advancenext flag which increments the next block by one.
+		 * The effect of this is to ensure that all backends are given
+		 * a separate block, minimizing block contention and thereby
+		 * maximising concurrency. This is the default strategy used by
+		 * PostgreSQL since at least PostgreSQL 8.4.
+		 */
+
+	FREESPACE_STRATEGY_MAX_COMPACT
+		/*
+		 * All backends are given the earliest block in the table with
+		 * sufficient freespace for the insert. This could cause block
+		 * contention for concurrent inserts, but ensures maximum data
+		 * compaction, which will then allow vacuum truncation to release
+		 * as much space as possible.  This strategy may be appropriate
+		 * for short periods if a table becomes bloated.
+		 */
+
+} FreeSpaceStrategy;
+
 /* prototypes for public functions in freespace.c */
 extern Size GetRecordedFreeSpace(Relation rel, BlockNumber heapBlk);
 extern BlockNumber GetPageWithFreeSpace(Relation rel, Size spaceNeeded);
+extern BlockNumber GetPageWithFreeSpaceExt(Relation rel, Size spaceNeeded,
+											FreeSpaceStrategy fss);
 extern BlockNumber RecordAndGetPageWithFreeSpace(Relation rel,
 												 BlockNumber oldPage,
 												 Size oldSpaceAvail,
 												 Size spaceNeeded);
+extern BlockNumber RecordAndGetPageWithFreeSpaceExt(Relation rel,
+												 BlockNumber oldPage,
+												 Size oldSpaceAvail,
+												 Size spaceNeeded,
+												 FreeSpaceStrategy fss);
 extern void RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk,
 									Size spaceAvail);
+extern void RecordPageWithFreeSpaceExt(Relation rel, BlockNumber heapBlk,
+									Size spaceAvail, FreeSpaceStrategy fss);
 extern void XLogRecordPageWithFreeSpace(RelFileLocator rlocator, BlockNumber heapBlk,
 										Size spaceAvail);