From a26fcd6cf1edfc91bd8d798ecc322abb683e883a Mon Sep 17 00:00:00 2001
From: Hari Babu <kommi.haribabu@gmail.com>
Date: Wed, 7 Feb 2018 11:59:55 +1100
Subject: [PATCH 13/14] Using "access method" syntax addition to create table

With the pluggable storage support, user can select the
table access method that he needs for the corresponding table.
The syntax addition is similar like CREATE INDEX.

CREATE TABLE ... [USING ACCESSMETHOD] ...

All the catalog relations are by default uses the HEAP method.
Currently the access method syntax support is added only for
main relations.

Currently Default access method HEAP is used for TOAST, VIEW,
SEQUENCE AND MATERIALIZED VIEWS.

Pending items: support of displaying access method with \d commands
---
 src/backend/bootstrap/bootparse.y         |  2 +
 src/backend/catalog/heap.c                |  4 ++
 src/backend/catalog/index.c               |  2 +-
 src/backend/catalog/toasting.c            |  1 +
 src/backend/commands/cluster.c            |  1 +
 src/backend/commands/tablecmds.c          | 28 ++++++++++++
 src/backend/nodes/copyfuncs.c             |  1 +
 src/backend/parser/gram.y                 | 71 ++++++++++++++++++-------------
 src/backend/utils/cache/relcache.c        | 12 +++---
 src/include/catalog/heap.h                |  2 +
 src/include/catalog/pg_am.h               |  1 +
 src/include/nodes/parsenodes.h            |  1 +
 src/include/utils/relcache.h              |  1 +
 src/test/regress/expected/type_sanity.out | 16 ++++---
 src/test/regress/sql/type_sanity.sql      |  6 +--
 15 files changed, 105 insertions(+), 44 deletions(-)

diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index 9e81f9514d..9a4e895da8 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -225,6 +225,7 @@ Boot_CreateStmt:
 												   shared_relation ? GLOBALTABLESPACE_OID : 0,
 												   $3,
 												   InvalidOid,
+												   HEAP_TABLE_AM_OID,
 												   tupdesc,
 												   RELKIND_RELATION,
 												   RELPERSISTENCE_PERMANENT,
@@ -244,6 +245,7 @@ Boot_CreateStmt:
 													  $7,
 													  InvalidOid,
 													  BOOTSTRAP_SUPERUSERID,
+													  HEAP_TABLE_AM_OID,
 													  tupdesc,
 													  NIL,
 													  RELKIND_RELATION,
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index cf36ce4add..d9227effc8 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -252,6 +252,7 @@ heap_create(const char *relname,
 			Oid reltablespace,
 			Oid relid,
 			Oid relfilenode,
+			Oid accessmtd,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
@@ -346,6 +347,7 @@ heap_create(const char *relname,
 									 relnamespace,
 									 tupDesc,
 									 relid,
+									 accessmtd,
 									 relfilenode,
 									 reltablespace,
 									 shared_relation,
@@ -1026,6 +1028,7 @@ heap_create_with_catalog(const char *relname,
 						 Oid reltypeid,
 						 Oid reloftypeid,
 						 Oid ownerid,
+						 Oid accessmtd,
 						 TupleDesc tupdesc,
 						 List *cooked_constraints,
 						 char relkind,
@@ -1168,6 +1171,7 @@ heap_create_with_catalog(const char *relname,
 							   reltablespace,
 							   relid,
 							   InvalidOid,
+							   accessmtd,
 							   tupdesc,
 							   relkind,
 							   relpersistence,
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index c8b889c54e..74ff6541c8 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -899,6 +899,7 @@ index_create(Relation heapRelation,
 								tableSpaceId,
 								indexRelationId,
 								relFileNode,
+								accessMethodObjectId,
 								indexTupDesc,
 								relkind,
 								relpersistence,
@@ -922,7 +923,6 @@ index_create(Relation heapRelation,
 	 * XXX should have a cleaner way to create cataloged indexes
 	 */
 	indexRelation->rd_rel->relowner = heapRelation->rd_rel->relowner;
-	indexRelation->rd_rel->relam = accessMethodObjectId;
 	indexRelation->rd_rel->relhasoids = false;
 
 	/*
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 8bf2698545..6e1affd75a 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -266,6 +266,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
 										   toast_typid,
 										   InvalidOid,
 										   rel->rd_rel->relowner,
+										   rel->rd_rel->relam,
 										   tupdesc,
 										   NIL,
 										   RELKIND_TOASTVALUE,
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c
index ae953d90ff..65c7fef6ef 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -679,6 +679,7 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, char relpersistence,
 										  InvalidOid,
 										  InvalidOid,
 										  OldHeap->rd_rel->relowner,
+										  OldHeap->rd_rel->relam,
 										  OldHeapDesc,
 										  NIL,
 										  RELKIND_RELATION,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index d3f4ba3c1c..b237a26b88 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -536,6 +536,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
 	Oid			ofTypeId;
 	ObjectAddress address;
+	Oid			accessMethodId = InvalidOid;
 
 	/*
 	 * Truncate relname to appropriate length (probably a waste of time, as
@@ -739,6 +740,32 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			attr->attidentity = colDef->identity;
 	}
 
+	/*
+	 * look up the access method, verify it can handle the requested features
+	 */
+	if (stmt->accessMethod != NULL)
+	{
+		HeapTuple	tuple;
+
+		tuple = SearchSysCache1(AMNAME, PointerGetDatum(stmt->accessMethod));
+		if (!HeapTupleIsValid(tuple))
+				ereport(ERROR,
+						(errcode(ERRCODE_UNDEFINED_OBJECT),
+						 errmsg("table access method \"%s\" does not exist",
+								 stmt->accessMethod)));
+		accessMethodId = HeapTupleGetOid(tuple);
+		ReleaseSysCache(tuple);
+	}
+	else if (relkind == RELKIND_RELATION ||
+			relkind == RELKIND_SEQUENCE ||
+			relkind == RELKIND_TOASTVALUE ||
+			relkind == RELKIND_VIEW ||
+			relkind == RELKIND_MATVIEW ||
+			relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		accessMethodId = HEAP_TABLE_AM_OID;
+	}
+
 	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
@@ -751,6 +778,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 										  InvalidOid,
 										  ofTypeId,
 										  ownerId,
+										  accessMethodId,
 										  descriptor,
 										  list_concat(cookedDefaults,
 													  old_constraints),
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 82255b0d1d..96130de7e2 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -3282,6 +3282,7 @@ CopyCreateStmtFields(const CreateStmt *from, CreateStmt *newnode)
 	COPY_NODE_FIELD(options);
 	COPY_SCALAR_FIELD(oncommit);
 	COPY_STRING_FIELD(tablespacename);
+	COPY_STRING_FIELD(accessMethod);
 	COPY_SCALAR_FIELD(if_not_exists);
 }
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 7da347e6bb..ee3907fccf 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -338,7 +338,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 %type <str>		copy_file_name
 				database_name access_method_clause access_method attr_name
-				name cursor_name file_name
+				table_access_method_clause name cursor_name file_name
 				index_name opt_index_name cluster_index_specification
 
 %type <list>	func_name handler_name qual_Op qual_all_Op subquery_Op
@@ -3190,7 +3190,8 @@ copy_generic_opt_arg_list_item:
  *****************************************************************************/
 
 CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
-			OptInherit OptPartitionSpec OptWith OnCommitOption OptTableSpace
+			OptInherit OptPartitionSpec table_access_method_clause OptWith
+			OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$4->relpersistence = $2;
@@ -3200,15 +3201,16 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->partspec = $9;
 					n->ofTypename = NULL;
 					n->constraints = NIL;
-					n->options = $10;
-					n->oncommit = $11;
-					n->tablespacename = $12;
+					n->accessMethod = $10;
+					n->options = $11;
+					n->oncommit = $12;
+					n->tablespacename = $13;
 					n->if_not_exists = false;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE IF_P NOT EXISTS qualified_name '('
-			OptTableElementList ')' OptInherit OptPartitionSpec OptWith
-			OnCommitOption OptTableSpace
+			OptTableElementList ')' OptInherit OptPartitionSpec table_access_method_clause
+			OptWith OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$7->relpersistence = $2;
@@ -3218,15 +3220,16 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->partspec = $12;
 					n->ofTypename = NULL;
 					n->constraints = NIL;
-					n->options = $13;
-					n->oncommit = $14;
-					n->tablespacename = $15;
+					n->accessMethod = $13;
+					n->options = $14;
+					n->oncommit = $15;
+					n->tablespacename = $16;
 					n->if_not_exists = true;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE qualified_name OF any_name
-			OptTypedTableElementList OptPartitionSpec OptWith OnCommitOption
-			OptTableSpace
+			OptTypedTableElementList OptPartitionSpec table_access_method_clause
+			OptWith OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$4->relpersistence = $2;
@@ -3237,15 +3240,16 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->ofTypename = makeTypeNameFromNameList($6);
 					n->ofTypename->location = @6;
 					n->constraints = NIL;
-					n->options = $9;
-					n->oncommit = $10;
-					n->tablespacename = $11;
+					n->accessMethod = $9;
+					n->options = $10;
+					n->oncommit = $11;
+					n->tablespacename = $12;
 					n->if_not_exists = false;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE IF_P NOT EXISTS qualified_name OF any_name
-			OptTypedTableElementList OptPartitionSpec OptWith OnCommitOption
-			OptTableSpace
+			OptTypedTableElementList OptPartitionSpec table_access_method_clause
+			OptWith OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$7->relpersistence = $2;
@@ -3256,15 +3260,16 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->ofTypename = makeTypeNameFromNameList($9);
 					n->ofTypename->location = @9;
 					n->constraints = NIL;
-					n->options = $12;
-					n->oncommit = $13;
-					n->tablespacename = $14;
+					n->accessMethod = $12;
+					n->options = $13;
+					n->oncommit = $14;
+					n->tablespacename = $15;
 					n->if_not_exists = true;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE qualified_name PARTITION OF qualified_name
-			OptTypedTableElementList PartitionBoundSpec OptPartitionSpec OptWith
-			OnCommitOption OptTableSpace
+			OptTypedTableElementList PartitionBoundSpec OptPartitionSpec
+			table_access_method_clause OptWith OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$4->relpersistence = $2;
@@ -3275,15 +3280,16 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->partspec = $10;
 					n->ofTypename = NULL;
 					n->constraints = NIL;
-					n->options = $11;
-					n->oncommit = $12;
-					n->tablespacename = $13;
+					n->accessMethod = $11;
+					n->options = $12;
+					n->oncommit = $13;
+					n->tablespacename = $14;
 					n->if_not_exists = false;
 					$$ = (Node *)n;
 				}
 		| CREATE OptTemp TABLE IF_P NOT EXISTS qualified_name PARTITION OF
 			qualified_name OptTypedTableElementList PartitionBoundSpec OptPartitionSpec
-			OptWith OnCommitOption OptTableSpace
+			table_access_method_clause OptWith OnCommitOption OptTableSpace
 				{
 					CreateStmt *n = makeNode(CreateStmt);
 					$7->relpersistence = $2;
@@ -3294,9 +3300,10 @@ CreateStmt:	CREATE OptTemp TABLE qualified_name '(' OptTableElementList ')'
 					n->partspec = $13;
 					n->ofTypename = NULL;
 					n->constraints = NIL;
-					n->options = $14;
-					n->oncommit = $15;
-					n->tablespacename = $16;
+					n->accessMethod = $14;
+					n->options = $15;
+					n->oncommit = $16;
+					n->tablespacename = $17;
 					n->if_not_exists = true;
 					$$ = (Node *)n;
 				}
@@ -3934,6 +3941,12 @@ part_elem: ColId opt_collate opt_class
 					$$ = n;
 				}
 		;
+
+table_access_method_clause:
+			USING access_method					{ $$ = $2; }
+			| /*EMPTY*/							{ $$ = DEFAULT_TABLE_ACCESS_METHOD; }
+		;
+
 /* WITH (options) is preferred, WITH OIDS and WITHOUT OIDS are legacy forms */
 OptWith:
 			WITH reloptions				{ $$ = $2; }
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 9281860ff1..98579cc22f 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -1891,11 +1891,8 @@ RelationInitTableAccessMethod(Relation relation)
 	HeapTuple	tuple;
 	Form_pg_am	aform;
 
-	/*
-	 * Relations that don't have a catalogued table access method use the
-	 * standard heap tableam module; otherwise a catalog lookup is in order.
-	 */
-	if (!OidIsValid(relation->rd_rel->relam))
+	if (IsCatalogRelation(relation) ||
+			!OidIsValid(relation->rd_rel->relam))
 	{
 		relation->rd_tableamhandler = HEAP_TABLE_AM_HANDLER_OID;
 	}
@@ -2081,6 +2078,7 @@ formrdesc(const char *relationName, Oid relationReltype,
 	/*
 	 * initialize the table am handler
 	 */
+	relation->rd_rel->relam = HEAP_TABLE_AM_OID;
 	relation->rd_tableamroutine = GetHeapamTableAmRoutine();
 
 	/*
@@ -3251,6 +3249,7 @@ RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
 						   Oid relid,
+						   Oid accessmtd,
 						   Oid relfilenode,
 						   Oid reltablespace,
 						   bool shared_relation,
@@ -3431,6 +3430,8 @@ RelationBuildLocalRelation(const char *relname,
 
 	RelationInitPhysicalAddr(rel);
 
+	rel->rd_rel->relam = accessmtd;
+
 	if (relkind == RELKIND_RELATION ||
 		relkind == RELKIND_MATVIEW ||
 		relkind == RELKIND_VIEW ||	/* Not exactly the storage, but underlying
@@ -3963,6 +3964,7 @@ RelationCacheInitializePhase3(void)
 		if (relation->rd_tableamroutine == NULL &&
 			(relation->rd_rel->relkind == RELKIND_RELATION ||
 			 relation->rd_rel->relkind == RELKIND_MATVIEW ||
+			 relation->rd_rel->relkind == RELKIND_VIEW ||
 			 relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ||
 			 relation->rd_rel->relkind == RELKIND_TOASTVALUE))
 		{
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.h
index 9bdc63ceb5..e1376ee4d4 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -44,6 +44,7 @@ extern Relation heap_create(const char *relname,
 			Oid reltablespace,
 			Oid relid,
 			Oid relfilenode,
+			Oid accessmtd,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
@@ -58,6 +59,7 @@ extern Oid heap_create_with_catalog(const char *relname,
 						 Oid reltypeid,
 						 Oid reloftypeid,
 						 Oid ownerid,
+						 Oid accessmtd,
 						 TupleDesc tupdesc,
 						 List *cooked_constraints,
 						 char relkind,
diff --git a/src/include/catalog/pg_am.h b/src/include/catalog/pg_am.h
index 127973dc84..675a929dd3 100644
--- a/src/include/catalog/pg_am.h
+++ b/src/include/catalog/pg_am.h
@@ -87,5 +87,6 @@ DESCR("block range index (BRIN) access method");
 DATA(insert OID = 4001 (  heap_tableam         heap_tableam_handler s ));
 DESCR("heap table access method");
 #define HEAP_TABLE_AM_OID 4001
+#define DEFAULT_TABLE_ACCESS_METHOD "heap_tableam"
 
 #endif							/* PG_AM_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index c7a43b8933..6df183a9ff 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2019,6 +2019,7 @@ typedef struct CreateStmt
 	List	   *options;		/* options from WITH clause */
 	OnCommitAction oncommit;	/* what do we do at COMMIT? */
 	char	   *tablespacename; /* table space to use, or NULL */
+	char	   *accessMethod;	/* table access method */
 	bool		if_not_exists;	/* just do nothing if it already exists? */
 } CreateStmt;
 
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 0c500729c0..3668b35c58 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -100,6 +100,7 @@ extern Relation RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
 						   Oid relid,
+						   Oid accessmtd,
 						   Oid relfilenode,
 						   Oid reltablespace,
 						   bool shared_relation,
diff --git a/src/test/regress/expected/type_sanity.out b/src/test/regress/expected/type_sanity.out
index b1419d4bc2..ccd88c0260 100644
--- a/src/test/regress/expected/type_sanity.out
+++ b/src/test/regress/expected/type_sanity.out
@@ -502,14 +502,18 @@ WHERE relkind NOT IN ('r', 'i', 'S', 't', 'v', 'm', 'c', 'f', 'p') OR
 -----+---------
 (0 rows)
 
--- Indexes should have an access method, others not.
+-- Except default tables, others should have an access method.
 SELECT p1.oid, p1.relname
 FROM pg_class as p1
-WHERE (p1.relkind = 'i' AND p1.relam = 0) OR
-    (p1.relkind != 'i' AND p1.relam != 0);
- oid | relname 
------+---------
-(0 rows)
+WHERE p1.relkind IN ('r', 'i', 'S', 't', 'v', 'm', 'p', 'i', 'I') and
+    p1.relam = 0;
+ oid  |   relname    
+------+--------------
+ 1247 | pg_type
+ 1249 | pg_attribute
+ 1255 | pg_proc
+ 1259 | pg_class
+(4 rows)
 
 -- **************** pg_attribute ****************
 -- Look for illegal values in pg_attribute fields
diff --git a/src/test/regress/sql/type_sanity.sql b/src/test/regress/sql/type_sanity.sql
index f9aeea3214..3ed0b9efcb 100644
--- a/src/test/regress/sql/type_sanity.sql
+++ b/src/test/regress/sql/type_sanity.sql
@@ -367,12 +367,12 @@ WHERE relkind NOT IN ('r', 'i', 'S', 't', 'v', 'm', 'c', 'f', 'p') OR
     relpersistence NOT IN ('p', 'u', 't') OR
     relreplident NOT IN ('d', 'n', 'f', 'i');
 
--- Indexes should have an access method, others not.
+-- Except default tables, others should have an access method.
 
 SELECT p1.oid, p1.relname
 FROM pg_class as p1
-WHERE (p1.relkind = 'i' AND p1.relam = 0) OR
-    (p1.relkind != 'i' AND p1.relam != 0);
+WHERE p1.relkind IN ('r', 'i', 'S', 't', 'v', 'm', 'p', 'i', 'I') and
+    p1.relam = 0;
 
 -- **************** pg_attribute ****************
 
-- 
2.16.1.windows.4

