From bb4ef35668ba13430fe669d902f554290f50fa77 Mon Sep 17 00:00:00 2001
From: Hari Babu <haribabuk@fast.au.fujitsu.com>
Date: Wed, 30 Aug 2017 12:49:46 +1000
Subject: [PATCH 3/8] Adding storageam hanlder to relation structure

And also the necessary functions to initialize
the storageam handler
---
 src/backend/utils/cache/relcache.c | 117 ++++++++++++++++++++++++++++++++++++-
 src/include/utils/rel.h            |  12 ++++
 src/include/utils/relcache.h       |   2 +
 3 files changed, 128 insertions(+), 3 deletions(-)

diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index b8e3780..40cbc67 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -34,6 +34,7 @@
 #include "access/multixact.h"
 #include "access/nbtree.h"
 #include "access/reloptions.h"
+#include "access/storageamapi.h"
 #include "access/sysattr.h"
 #include "access/xact.h"
 #include "access/xlog.h"
@@ -1368,10 +1369,26 @@ RelationBuildDesc(Oid targetRelId, bool insertIt)
 	}
 
 	/*
-	 * if it's an index, initialize index-related information
+	 * initialize access method information
 	 */
-	if (OidIsValid(relation->rd_rel->relam))
-		RelationInitIndexAccessInfo(relation);
+	switch (relation->rd_rel->relkind)
+	{
+		case RELKIND_INDEX:
+			Assert(relation->rd_rel->relkind != InvalidOid);
+			RelationInitIndexAccessInfo(relation);
+			break;
+		case RELKIND_RELATION:
+		case RELKIND_SEQUENCE:
+		case RELKIND_TOASTVALUE:
+		case RELKIND_VIEW: /* Not exactly the storage, but underlying tuple access, it is required */
+		case RELKIND_MATVIEW:
+		case RELKIND_PARTITIONED_TABLE:
+			RelationInitStorageAccessInfo(relation);
+			break;
+		default:
+			/* nothing to do in other cases */
+			break;
+	}
 
 	/* extract reloptions if any */
 	RelationParseRelOptions(relation, pg_class_tuple);
@@ -1868,6 +1885,71 @@ LookupOpclassInfo(Oid operatorClassOid,
 	return opcentry;
 }
 
+/*
+ * Fill in the StorageAmRoutine for a relation
+ *
+ * relation's rd_amhandler and rd_indexcxt (XXX?) must be valid already.
+ */
+static void
+InitStorageAmRoutine(Relation relation)
+{
+	StorageAmRoutine *cached,
+					 *tmp;
+
+	/*
+	 * Call the amhandler in current, short-lived memory context, just in case
+	 * it leaks anything (it probably won't, but let's be paranoid).
+	 */
+	tmp = GetStorageAmRoutine(relation->rd_amhandler);
+
+	/* XXX do we need a separate memory context for this? */
+	/* OK, now transfer the data into cache context */
+	cached = (StorageAmRoutine *) MemoryContextAlloc(CacheMemoryContext,
+													 sizeof(StorageAmRoutine));
+	memcpy(cached, tmp, sizeof(StorageAmRoutine));
+	relation->rd_stamroutine = cached;
+
+	pfree(tmp);
+}
+
+/*
+ * Initialize storage-access-method support data for a heap relation
+ */
+void
+RelationInitStorageAccessInfo(Relation relation)
+{
+	HeapTuple	tuple;
+	Form_pg_am	aform;
+
+	/*
+	 * Relations that don't have a catalogued storage access method use the
+	 * standard heapam module; otherwise a catalog lookup is in order.
+	 */
+	if (!OidIsValid(relation->rd_rel->relam))
+	{
+		relation->rd_amhandler = HEAPAM_STORAGE_AM_HANDLER_OID;
+	}
+	else
+	{
+		/*
+		 * Look up the storage access method, save the OID of its handler
+		 * function.
+		 */
+		tuple = SearchSysCache1(AMOID,
+								ObjectIdGetDatum(relation->rd_rel->relam));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for access method %u",
+				 relation->rd_rel->relam);
+		aform = (Form_pg_am) GETSTRUCT(tuple);
+		relation->rd_amhandler = aform->amhandler;
+		ReleaseSysCache(tuple);
+	}
+
+	/*
+	 * Now we can fetch the storage AM's API struct
+	 */
+	InitStorageAmRoutine(relation);
+}
 
 /*
  *		formrdesc
@@ -2027,6 +2109,11 @@ formrdesc(const char *relationName, Oid relationReltype,
 	RelationInitPhysicalAddr(relation);
 
 	/*
+	 * initialize the storage am handler
+	 */
+	relation->rd_stamroutine = GetHeapamStorageAmRoutine();
+
+	/*
 	 * initialize the rel-has-index flag, using hardwired knowledge
 	 */
 	if (IsBootstrapProcessingMode())
@@ -2354,6 +2441,8 @@ RelationDestroyRelation(Relation relation, bool remember_tupdesc)
 		pfree(relation->rd_pubactions);
 	if (relation->rd_options)
 		pfree(relation->rd_options);
+	if (relation->rd_stamroutine)
+		pfree(relation->rd_stamroutine);
 	if (relation->rd_indextuple)
 		pfree(relation->rd_indextuple);
 	if (relation->rd_indexcxt)
@@ -3368,6 +3457,13 @@ RelationBuildLocalRelation(const char *relname,
 
 	RelationInitPhysicalAddr(rel);
 
+	if (relkind == RELKIND_RELATION ||
+		relkind == RELKIND_MATVIEW ||
+		relkind == RELKIND_VIEW || /* Not exactly the storage, but underlying tuple access, it is required */
+		relkind == RELKIND_PARTITIONED_TABLE ||
+		relkind == RELKIND_TOASTVALUE)
+		RelationInitStorageAccessInfo(rel);
+
 	/*
 	 * Okay to insert into the relcache hash table.
 	 *
@@ -3889,6 +3985,18 @@ RelationCacheInitializePhase3(void)
 			restart = true;
 		}
 
+		if (relation->rd_stamroutine == NULL &&
+			(relation->rd_rel->relkind == RELKIND_RELATION ||
+			relation->rd_rel->relkind == RELKIND_MATVIEW ||
+			relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ||
+			relation->rd_rel->relkind == RELKIND_TOASTVALUE))
+		{
+			RelationInitStorageAccessInfo(relation);
+			Assert (relation->rd_stamroutine != NULL);
+
+			restart = true;
+		}
+
 		/* Release hold on the relation */
 		RelationDecrementReferenceCount(relation);
 
@@ -5607,6 +5715,9 @@ load_relcache_init_file(bool shared)
 			if (rel->rd_isnailed)
 				nailed_rels++;
 
+			/* Load storage AM stuff */
+			RelationInitStorageAccessInfo(rel);
+
 			Assert(rel->rd_index == NULL);
 			Assert(rel->rd_indextuple == NULL);
 			Assert(rel->rd_indexcxt == NULL);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 4bc61e5..82d8a53 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -161,6 +161,12 @@ typedef struct RelationData
 	struct HeapTupleData *rd_indextuple;	/* all of pg_index tuple */
 
 	/*
+	 * Underlying storage support
+	 */
+	Oid		rd_storageam;		/* OID of storage AM handler function */
+	struct StorageAmRoutine *rd_stamroutine; /* storage AM's API struct */
+
+	/*
 	 * index access support info (used only for an index relation)
 	 *
 	 * Note: only default support procs for each opclass are cached, namely
@@ -428,6 +434,12 @@ typedef struct ViewOptions
 #define RelationGetDescr(relation) ((relation)->rd_att)
 
 /*
+ * RelationGetStorageRoutine
+ *		Returns the storage AM routine for a relation.
+ */
+#define RelationGetStorageRoutine(relation) ((relation)->rd_stamroutine)
+
+/*
  * RelationGetRelationName
  *		Returns the rel's name.
  *
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 3c53cef..03d996f 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -71,6 +71,8 @@ extern void RelationInitIndexAccessInfo(Relation relation);
 struct PublicationActions;
 extern struct PublicationActions *GetRelationPublicationActions(Relation relation);
 
+extern void RelationInitStorageAccessInfo(Relation relation);
+
 /*
  * Routines to support ereport() reports of relation-related errors
  */
-- 
2.7.4.windows.1

