From 365f40e1d94341d4427a47005ceb84c927346ea5 Mon Sep 17 00:00:00 2001
From: Hari Babu <kommi.haribabu@gmail.com>
Date: Thu, 29 Mar 2018 16:17:54 +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 4ea3aa97cf..8996a2fecd 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 a1def77944..0ad60e8b3b 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -256,6 +256,7 @@ heap_create(const char *relname,
 			Oid reltablespace,
 			Oid relid,
 			Oid relfilenode,
+			Oid accessmtd,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
@@ -350,6 +351,7 @@ heap_create(const char *relname,
 									 relnamespace,
 									 tupDesc,
 									 relid,
+									 accessmtd,
 									 relfilenode,
 									 reltablespace,
 									 shared_relation,
@@ -1032,6 +1034,7 @@ heap_create_with_catalog(const char *relname,
 						 Oid reltypeid,
 						 Oid reloftypeid,
 						 Oid ownerid,
+						 Oid accessmtd,
 						 TupleDesc tupdesc,
 						 List *cooked_constraints,
 						 char relkind,
@@ -1175,6 +1178,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 5d7336d77f..e62786b5bf 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -901,6 +901,7 @@ index_create(Relation heapRelation,
 								tableSpaceId,
 								indexRelationId,
 								relFileNode,
+								accessMethodObjectId,
 								indexTupDesc,
 								relkind,
 								relpersistence,
@@ -924,7 +925,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 9007dc6ebe..377d534dc0 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 556f745012..045b790b93 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 fd197d9042..cc98ba488b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -537,6 +537,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
@@ -741,6 +742,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
@@ -753,6 +780,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 c7293a60d7..261582abf7 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -3284,6 +3284,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 71b5854447..9b4553730c 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -339,7 +339,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
@@ -3191,7 +3191,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;
@@ -3201,15 +3202,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;
@@ -3219,15 +3221,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;
@@ -3238,15 +3241,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;
@@ -3257,15 +3261,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;
@@ -3276,15 +3281,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;
@@ -3295,9 +3301,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;
 				}
@@ -3936,6 +3943,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 94bca41432..c99602c74f 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -1959,11 +1959,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;
 	}
@@ -2145,6 +2142,7 @@ formrdesc(const char *relationName, Oid relationReltype,
 	/*
 	 * initialize the table am handler
 	 */
+	relation->rd_rel->relam = HEAP_TABLE_AM_OID;
 	relation->rd_tableamroutine = GetHeapamTableAmRoutine();
 
 	/*
@@ -3317,6 +3315,7 @@ RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
 						   Oid relid,
+						   Oid accessmtd,
 						   Oid relfilenode,
 						   Oid reltablespace,
 						   bool shared_relation,
@@ -3497,6 +3496,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
@@ -4029,6 +4030,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 59fc052494..00688d4718 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -45,6 +45,7 @@ extern Relation heap_create(const char *relname,
 			Oid reltablespace,
 			Oid relid,
 			Oid relfilenode,
+			Oid accessmtd,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
@@ -59,6 +60,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 92082b3a7a..277f6b4485 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2020,6 +2020,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 df16b4ca98..858a7b30d2 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -101,6 +101,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

