store_local_map_relcache_v1.patch

diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 69a7a23874..d854a84bf3 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -500,12 +500,6 @@ loop:
 			/* use this page as future insert target, too */
 			RelationSetTargetBlock(relation, targetBlock);
 
-			/*
-			 * In case we used an in-memory map of available blocks, reset it
-			 * for next use.
-			 */
-			FSMClearLocalMap();
-
 			return buffer;
 		}
 
@@ -675,8 +669,5 @@ loop:
 	 */
 	RelationSetTargetBlock(relation, BufferGetBlockNumber(buffer));
 
-	/* This should already be cleared by now, but make sure it is. */
-	FSMClearLocalMap();
-
 	return buffer;
 }
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 8522a2f51c..d14c41ddaa 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2587,12 +2587,6 @@ AbortTransaction(void)
 	pgstat_report_wait_end();
 	pgstat_progress_end_command();
 
-	/*
-	 * In case we aborted during RelationGetBufferForTuple(), clear the local
-	 * map of heap pages.
-	 */
-	FSMClearLocalMap();
-
 	/* Clean up buffer I/O and buffer context locks, too */
 	AbortBufferIO();
 	UnlockBuffers();
@@ -4881,12 +4875,6 @@ AbortSubTransaction(void)
 	pgstat_report_wait_end();
 	pgstat_progress_end_command();
 
-	/*
-	 * In case we aborted during RelationGetBufferForTuple(), clear the local
-	 * map of heap pages.
-	 */
-	FSMClearLocalMap();
-
 	AbortBufferIO();
 	UnlockBuffers();
 
diff --git a/src/backend/storage/freespace/freespace.c b/src/backend/storage/freespace/freespace.c
index c3ed4242e2..107b2da030 100644
--- a/src/backend/storage/freespace/freespace.c
+++ b/src/backend/storage/freespace/freespace.c
@@ -30,6 +30,7 @@
 #include "storage/fsm_internals.h"
 #include "storage/lmgr.h"
 #include "storage/smgr.h"
+#include "utils/memutils.h"
 
 
 /*
@@ -76,14 +77,6 @@
 #define FSM_ROOT_LEVEL	(FSM_TREE_DEPTH - 1)
 #define FSM_BOTTOM_LEVEL 0
 
-/* Status codes for the local map. */
-
-/* Either already tried, or beyond the end of the relation */
-#define FSM_LOCAL_NOT_AVAIL 0x00
-
-/* Available to try */
-#define FSM_LOCAL_AVAIL		0x01
-
 /*
  * The internal FSM routines work on a logical addressing scheme. Each
  * level of the tree can be thought of as a separately addressable file.
@@ -97,31 +90,7 @@ typedef struct
 /* Address of the root page. */
 static const FSMAddress FSM_ROOT_ADDRESS = {FSM_ROOT_LEVEL, 0};
 
-/*
- * For small relations, we don't create FSM to save space, instead we use
- * local in-memory map of pages to try.  To locate free space, we simply try
- * pages directly without knowing ahead of time how much free space they have.
- *
- * Note that this map is used to the find the block with required free space
- * for any given relation.  We clear this map when we have found a block with
- * enough free space, when we extend the relation, or on transaction abort.
- * See src/backend/storage/freespace/README for further details.
- */
-typedef struct
-{
-	BlockNumber nblocks;
-	uint8		map[HEAP_FSM_CREATION_THRESHOLD];
-}			FSMLocalMap;
-
-static FSMLocalMap fsm_local_map =
-{
-	0,
-	{
-		FSM_LOCAL_NOT_AVAIL
-	}
-};
-
-#define FSM_LOCAL_MAP_EXISTS (fsm_local_map.nblocks > 0)
+#define FSM_LOCAL_MAP_EXISTS(rel) (rel->fsm_local_map->nblocks > 0)
 
 /* functions to navigate the tree */
 static FSMAddress fsm_get_child(FSMAddress parent, uint16 slot);
@@ -143,12 +112,13 @@ static int fsm_set_and_search(Relation rel, FSMAddress addr, uint16 slot,
 				   uint8 newValue, uint8 minValue);
 static void fsm_local_set(Relation rel, BlockNumber cur_nblocks);
 static BlockNumber fsm_search(Relation rel, uint8 min_cat);
-static BlockNumber fsm_local_search(void);
+static BlockNumber fsm_local_search(Relation rel);
 static uint8 fsm_vacuum_page(Relation rel, FSMAddress addr,
 				BlockNumber start, BlockNumber end,
 				bool *eof);
 static bool fsm_allow_writes(Relation rel, BlockNumber heapblk,
 				 BlockNumber nblocks, BlockNumber *get_nblocks);
+static void fsm_clear_local_map(Relation rel);
 
 
 /******** Public API ********/
@@ -195,14 +165,17 @@ GetPageWithFreeSpace(Relation rel, Size spaceNeeded, bool check_fsm_only)
 			 * during bootstrapping or in a recently-started system.
 			 */
 			target_block = nblocks - 1;
+			fsm_clear_local_map(rel);
 		}
 		else if (nblocks > 0)
 		{
 			/* Initialize local map and get first candidate block. */
 			fsm_local_set(rel, nblocks);
-			target_block = fsm_local_search();
+			target_block = fsm_local_search(rel);
 		}
 	}
+	else
+		fsm_clear_local_map(rel);
 
 	return target_block;
 }
@@ -231,14 +204,14 @@ RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
 	BlockNumber nblocks = InvalidBlockNumber;
 
 	/* First try the local map, if it exists. */
-	if (FSM_LOCAL_MAP_EXISTS)
+	if (FSM_LOCAL_MAP_EXISTS(rel))
 	{
 		Assert((rel->rd_rel->relkind == RELKIND_RELATION ||
 				rel->rd_rel->relkind == RELKIND_TOASTVALUE) &&
-			   fsm_local_map.map[oldPage] == FSM_LOCAL_AVAIL);
+			   rel->fsm_local_map->map[oldPage] == FSM_LOCAL_AVAIL);
 
-		fsm_local_map.map[oldPage] = FSM_LOCAL_NOT_AVAIL;
-		return fsm_local_search();
+		rel->fsm_local_map->map[oldPage] = FSM_LOCAL_NOT_AVAIL;
+		return fsm_local_search(rel);
 	}
 
 	if (!fsm_allow_writes(rel, oldPage, InvalidBlockNumber, &nblocks))
@@ -249,8 +222,10 @@ RecordAndGetPageWithFreeSpace(Relation rel, BlockNumber oldPage,
 		 * need to create the local map.
 		 */
 		fsm_local_set(rel, nblocks);
-		return fsm_local_search();
+		return fsm_local_search(rel);
 	}
+	else
+		fsm_clear_local_map(rel);
 
 	/* Normal FSM logic follows */
 
@@ -302,18 +277,21 @@ RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk,
 }
 
 /*
- * Clear the local map.  We must call this when we have found a block with
- * enough free space, when we extend the relation, or on transaction abort.
+ * Initialize FSM Relation Map.
  */
 void
-FSMClearLocalMap(void)
+RelationInitFSMMap(Relation rel)
 {
-	if (FSM_LOCAL_MAP_EXISTS)
-	{
-		fsm_local_map.nblocks = 0;
-		memset(&fsm_local_map.map, FSM_LOCAL_NOT_AVAIL,
-			   sizeof(fsm_local_map.map));
-	}
+	FSMLocalMap *fsm_local_map;
+	MemoryContext	oldcontext;
+
+	oldcontext = MemoryContextSwitchTo(CacheMemoryContext);
+	fsm_local_map = palloc(sizeof(FSMLocalMap));
+	fsm_local_map->nblocks = 0;
+	memset(&fsm_local_map->map, FSM_LOCAL_NOT_AVAIL,
+		   sizeof(fsm_local_map->map));
+	rel->fsm_local_map = fsm_local_map;
+	MemoryContextSwitchTo(oldcontext);
 }
 
 /*
@@ -1132,9 +1110,6 @@ fsm_local_set(Relation rel, BlockNumber cur_nblocks)
 	BlockNumber blkno,
 				cached_target_block;
 
-	/* The local map must not be set already. */
-	Assert(!FSM_LOCAL_MAP_EXISTS);
-
 	/*
 	 * Starting at the current last block in the relation and working
 	 * backwards, mark alternating blocks as available.
@@ -1142,7 +1117,7 @@ fsm_local_set(Relation rel, BlockNumber cur_nblocks)
 	blkno = cur_nblocks - 1;
 	while (true)
 	{
-		fsm_local_map.map[blkno] = FSM_LOCAL_AVAIL;
+		rel->fsm_local_map->map[blkno] = FSM_LOCAL_AVAIL;
 		if (blkno >= 2)
 			blkno -= 2;
 		else
@@ -1150,13 +1125,13 @@ fsm_local_set(Relation rel, BlockNumber cur_nblocks)
 	}
 
 	/* Cache the number of blocks. */
-	fsm_local_map.nblocks = cur_nblocks;
+	rel->fsm_local_map->nblocks = cur_nblocks;
 
 	/* Set the status of the cached target block to 'unavailable'. */
 	cached_target_block = RelationGetTargetBlock(rel);
 	if (cached_target_block != InvalidBlockNumber &&
 		cached_target_block < cur_nblocks)
-		fsm_local_map.map[cached_target_block] = FSM_LOCAL_NOT_AVAIL;
+		rel->fsm_local_map->map[cached_target_block] = FSM_LOCAL_NOT_AVAIL;
 }
 
 /*
@@ -1168,18 +1143,18 @@ fsm_local_set(Relation rel, BlockNumber cur_nblocks)
  * This function is used when there is no FSM.
  */
 static BlockNumber
-fsm_local_search(void)
+fsm_local_search(Relation rel)
 {
 	BlockNumber target_block;
 
 	/* Local map must be set by now. */
-	Assert(FSM_LOCAL_MAP_EXISTS);
+	Assert(FSM_LOCAL_MAP_EXISTS(rel));
 
-	target_block = fsm_local_map.nblocks;
+	target_block = rel->fsm_local_map->nblocks;
 	do
 	{
 		target_block--;
-		if (fsm_local_map.map[target_block] == FSM_LOCAL_AVAIL)
+		if (rel->fsm_local_map->map[target_block] == FSM_LOCAL_AVAIL)
 			return target_block;
 	} while (target_block > 0);
 
@@ -1189,7 +1164,22 @@ fsm_local_search(void)
 	 * first, which would otherwise lead to the same conclusion again and
 	 * again.
 	 */
-	FSMClearLocalMap();
+	fsm_clear_local_map(rel);
 
 	return InvalidBlockNumber;
 }
+
+/*
+ * Clear the local map.  We must call this when we have found a block with
+ * enough free space, when we extend the relation, or on transaction abort.
+ */
+static void
+fsm_clear_local_map(Relation rel)
+{
+	if (FSM_LOCAL_MAP_EXISTS(rel))
+	{
+		rel->fsm_local_map->nblocks = 0;
+		memset(&rel->fsm_local_map->map, FSM_LOCAL_NOT_AVAIL,
+			sizeof(rel->fsm_local_map->map));
+	}
+}
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index bab59f16e6..57fac56c92 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -75,6 +75,7 @@
 #include "partitioning/partdesc.h"
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rowsecurity.h"
+#include "storage/freespace.h"
 #include "storage/lmgr.h"
 #include "storage/smgr.h"
 #include "utils/array.h"
@@ -1226,6 +1227,9 @@ RelationBuildDesc(Oid targetRelId, bool insertIt)
 	 */
 	RelationInitPhysicalAddr(relation);
 
+	/* Initialize FSM map for the relation. */
+	RelationInitFSMMap(relation);
+
 	/* make sure relation is marked as having no open file yet */
 	relation->rd_smgr = NULL;
 
@@ -1936,6 +1940,9 @@ formrdesc(const char *relationName, Oid relationReltype,
 	 */
 	RelationInitPhysicalAddr(relation);
 
+	/* Initialize FSM map for the relation. */
+	RelationInitFSMMap(relation);
+
 	/*
 	 * initialize the table am handler
 	 */
@@ -2612,6 +2619,7 @@ RelationClearRelation(Relation relation, bool rebuild)
 		SWAPFIELD(Oid, rd_toastoid);
 		/* pgstat_info must be preserved */
 		SWAPFIELD(struct PgStat_TableStatus *, pgstat_info);
+		SWAPFIELD(struct FSMLocalMap *, fsm_local_map);
 		/* preserve old partitioning info if no logical change */
 		if (keep_partkey)
 		{
@@ -3376,6 +3384,9 @@ RelationBuildLocalRelation(const char *relname,
 
 	RelationInitPhysicalAddr(rel);
 
+	/* Initialize FSM map for the relation. */
+	RelationInitFSMMap(rel);
+
 	rel->rd_rel->relam = accessmtd;
 
 	if (relkind == RELKIND_RELATION ||
@@ -5665,6 +5676,9 @@ load_relcache_init_file(bool shared)
 		 */
 		RelationInitLockInfo(rel);
 		RelationInitPhysicalAddr(rel);
+
+		/* Initialize FSM map for the relation. */
+		RelationInitFSMMap(rel);
 	}
 
 	/*
diff --git a/src/include/storage/freespace.h b/src/include/storage/freespace.h
index dbaae651c5..2cabb725e2 100644
--- a/src/include/storage/freespace.h
+++ b/src/include/storage/freespace.h
@@ -21,6 +21,31 @@
 /* Only create the FSM if the heap has greater than this many blocks */
 #define HEAP_FSM_CREATION_THRESHOLD 4
 
+/*
+ * For small relations, we don't create FSM to save space, instead we use
+ * local in-memory map of pages to try.  To locate free space, we simply try
+ * pages directly without knowing ahead of time how much free space they have.
+ *
+ * Note that this map is used to the find the block with required free space
+ * for any given relation.  We clear this map when we have found a block with
+ * enough free space, when we extend the relation, or on transaction abort.
+ * See src/backend/storage/freespace/README for further details.
+ */
+typedef struct FSMLocalMap
+{
+	BlockNumber nblocks;
+	uint8		map[HEAP_FSM_CREATION_THRESHOLD];
+}			FSMLocalMap;
+
+/* Status codes for the local map. */
+
+/* Either already tried, or beyond the end of the relation */
+#define FSM_LOCAL_NOT_AVAIL 0x00
+
+/* Available to try */
+#define FSM_LOCAL_AVAIL		0x01
+
+
 /* prototypes for public functions in freespace.c */
 extern Size GetRecordedFreeSpace(Relation rel, BlockNumber heapBlk);
 extern BlockNumber GetPageWithFreeSpace(Relation rel, Size spaceNeeded,
@@ -31,7 +56,7 @@ extern BlockNumber RecordAndGetPageWithFreeSpace(Relation rel,
 							  Size spaceNeeded);
 extern void RecordPageWithFreeSpace(Relation rel, BlockNumber heapBlk,
 						Size spaceAvail, BlockNumber nblocks);
-extern void FSMClearLocalMap(void);
+extern void RelationInitFSMMap(Relation rel);
 extern void XLogRecordPageWithFreeSpace(RelFileNode rnode, BlockNumber heapBlk,
 							Size spaceAvail);
 
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index bddfcafe26..5e12c40890 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -200,6 +200,7 @@ typedef struct RelationData
 
 	/* use "struct" here to avoid needing to include pgstat.h: */
 	struct PgStat_TableStatus *pgstat_info; /* statistics collection area */
+	struct FSMLocalMap *fsm_local_map;
 } RelationData;