Feature: add user-defined index access method

The purpose of the commit is in support of user-defined index access
method. Currently, there are many restrictions for index access methods
and new index access methods are incompatible with internal framework.
At the same time, it's impossible for user to change default index
access method. So I build some hooks that are for changing default
behavior and being compatible for those strange restrictions.

For express the meaning of the modification, I show a example as below.

For example ,  I want to add 7 kinds of new index access method that based on diffrent storage engine (maybe unionstore storage) and they are correspond to internal index access methods as below:

- usbtree ---btree
- usgin --- gin
- usgist --- gist
- usspgist --- spgist
- ushash --- hash
- usbitmap --- bitmap

The main diffrence between them is index data can separate from other data and support Compute and Storage Separation.

Suppose we want to use these new functions, we can create a new extension and we can run sql （for example by `create extension unionstore;`) command and you can find new access methods :
```
regression=# select * from pg_am;
  oid   |    amname     |         amhandler         | amtype
--------+---------------+---------------------------+--------
      2 | heap          | heap_tableam_handler      | t
    403 | btree         | bthandler                 | i
    405 | hash          | hashhandler               | i
    783 | gist          | gisthandler               | i
   2742 | gin           | ginhandler                | i
   4000 | spgist        | spghandler                | i
   3580 | brin          | brinhandler               | i
   7024 | ao_row        | ao_row_tableam_handler    | t
   7166 | ao_column     | ao_column_tableam_handler | t
   7013 | bitmap        | bmhandler                 | i
  16394 | union_store   | heap_tableam_handler      | t
  16402 | ushash        | ushashhandler             | i
  16403 | usbtree       | usbthandler               | i
  16404 | usgist        | usgisthandler             | i
  16405 | usgin         | usginhandler              | i
  16406 | usspgist      | usspghandler              | i
  16407 | usbrin        | usbrinhandler             | i
  16408 | usbitmap      | usbmhandler               | i
  22820 | heap2         | heap_tableam_handler      | t
 111831 | ao_row_testam | ao_row_tableam_handler    | t
 111847 | ao_col_testam | ao_column_tableam_handler | t
 111850 | heap_testam   | heap_tableam_handler      | t
```
So you can use these new index access methods as internal index access method. At the same time, I add new guc variable and you can correct default index type by `set default_index_type = usbtree` or show relevant info by `show default_index_type`;

In conclusion, it's more flexible and it decoupe from other part.

src/backend/access/brin/brin.c

@@ -1204,7 +1204,7 @@ brin_summarize_range_internal(PG_FUNCTION_ARGS)
 
 	/* Must be a BRIN index */
 	if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
-		indexRel->rd_rel->relam != BRIN_AM_OID)
+		!IsIndexAccessMethod(indexRel->rd_rel->relam, BRIN_AM_OID))
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("\"%s\" is not a BRIN index",
@@ -1290,7 +1290,7 @@ brin_desummarize_range(PG_FUNCTION_ARGS)
 
 	/* Must be a BRIN index */
 	if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
-		indexRel->rd_rel->relam != BRIN_AM_OID)
+		!IsIndexAccessMethod(indexRel->rd_rel->relam, BRIN_AM_OID))
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("\"%s\" is not a BRIN index",

src/backend/access/gin/ginfast.c

@@ -1042,7 +1042,7 @@ gin_clean_pending_list(PG_FUNCTION_ARGS)
 
 	/* Must be a GIN index */
 	if (indexRel->rd_rel->relkind != RELKIND_INDEX ||
-		indexRel->rd_rel->relam != GIN_AM_OID)
+		!IsIndexAccessMethod(indexRel->rd_rel->relam, GIN_AM_OID))
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("\"%s\" is not a GIN index",

src/backend/access/index/amapi.c

@@ -20,7 +20,18 @@
 #include "utils/builtins.h"
 #include "utils/syscache.h"
 
+is_index_access_method_hook_type is_index_access_method_hook = NULL;
 
+bool
+IsIndexAccessMethod(Oid relam, Oid indexAccessMethod)
+{
+	if ((is_index_access_method_hook && (*is_index_access_method_hook)(relam, indexAccessMethod)) ||
+		(!is_index_access_method_hook && relam == indexAccessMethod))
+	{
+		return true;
+	}
+	return false;
+}
 /*
  * GetIndexAmRoutine - call the specified access method handler routine to get
  * its IndexAmRoutine struct, which will be palloc'd in the caller's context.

src/backend/access/index/indexam.c

@@ -988,3 +988,54 @@ index_opclass_options(Relation indrel, AttrNumber attnum, Datum attoptions,
 
 	return build_local_reloptions(&relopts, attoptions, validate);
 }
+
+/* check_hook: validate new default_index_access_method */
+bool
+check_default_index_access_method(char **newval, void **extra, GucSource source)
+{
+	if (**newval == '\0')
+	{
+		GUC_check_errdetail("%s cannot be empty.",
+		                    "check_default_index_access_method");
+		return false;
+	}
+
+	if (strlen(*newval) >= NAMEDATALEN)
+	{
+		GUC_check_errdetail("%s is too long (maximum %d characters).",
+		                    "check_default_index_access_method", NAMEDATALEN - 1);
+		return false;
+	}
+
+	/*
+	 * If we aren't inside a transaction, or not connected to a database, we
+	 * cannot do the catalog access necessary to verify the method.  Must
+	 * accept the value on faith.
+	 */
+	if (IsTransactionState() && MyDatabaseId != InvalidOid)
+	{
+		if (!OidIsValid(get_index_am_oid(*newval, true)))
+		{
+			/*
+			 * When source == PGC_S_TEST, don't throw a hard error for a
+			 * nonexistent index access method, only a NOTICE. See comments in
+			 * guc.h.
+			 */
+			if (source == PGC_S_TEST)
+			{
+				ereport(NOTICE,
+				        (errcode(ERRCODE_UNDEFINED_OBJECT),
+					        errmsg("index access method \"%s\" does not exist",
+					               *newval)));
+			}
+			else
+			{
+				GUC_check_errdetail("index access method \"%s\" does not exist.",
+				                    *newval);
+				return false;
+			}
+		}
+	}
+
+	return true;
+}

src/backend/access/nbtree/nbtree.c

@@ -41,7 +41,6 @@
 #include "catalog/indexing.h"
 #include "catalog/pg_namespace.h"
 
-
 /*
  * BTPARALLEL_NOT_INITIALIZED indicates that the scan has not started.
  *

src/backend/access/spgist/spgutils.c

@@ -35,7 +35,6 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
-
 /*
  * SP-GiST handler function: return IndexAmRoutine with access method parameters
  * and callbacks.

src/backend/catalog/index.c

@@ -99,6 +99,8 @@
 #include "cdb/cdboidsync.h"
 #include "utils/faultinjector.h"
 
+/* GUC variables */
+char       *default_index_access_method = DEFAULT_INDEX_TYPE;
 /* Potentially set by pg_upgrade_support functions */
 Oid			binary_upgrade_next_index_pg_class_oid = InvalidOid;
 
@@ -2808,7 +2810,7 @@ BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii)
 	 */
 	Assert(ii->ii_Unique);
 
-	if (index->rd_rel->relam != BTREE_AM_OID)
+	if (!IsIndexAccessMethod(index->rd_rel->relam, BTREE_AM_OID))
 		elog(ERROR, "unexpected non-btree speculative unique index");
 
 	ii->ii_UniqueOps = (Oid *) palloc(sizeof(Oid) * indnkeyatts);
@@ -3129,7 +3131,7 @@ index_build(Relation heapRelation,
 	 * to introduce parallelism to singlenode mode.
 	 */
 	if (parallel && !IS_SINGLENODE() && IsNormalProcessingMode() &&
-		indexRelation->rd_rel->relam == BTREE_AM_OID)
+		IsIndexAccessMethod(indexRelation->rd_rel->relam, BTREE_AM_OID))
 		indexInfo->ii_ParallelWorkers =
 			plan_create_index_workers(RelationGetRelid(heapRelation),
 									  RelationGetRelid(indexRelation));

src/backend/commands/cluster.c

@@ -962,7 +962,7 @@ copy_table_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose,
 	 * tells us it's cheaper.  Otherwise, always indexscan if an index is
 	 * provided, else plain seqscan.
 	 */
-	if (OldIndex != NULL && OldIndex->rd_rel->relam == BTREE_AM_OID)
+	if (OldIndex != NULL && IsIndexAccessMethod(OldIndex->rd_rel->relam, BTREE_AM_OID))
 		use_sort = plan_cluster_use_sort(OIDOldHeap, OIDOldIndex);
 	else
 		use_sort = false;

src/backend/commands/indexcmds.c

@@ -1021,6 +1021,10 @@ DefineIndex(Oid relationId,
 	 * look up the access method, verify it can handle the requested features
 	 */
 	accessMethodName = stmt->accessMethod;
+	if (accessMethodName == NULL)
+	{
+		accessMethodName = default_index_access_method;
+	}
 	tuple = SearchSysCache1(AMNAME, PointerGetDatum(accessMethodName));
 	if (!HeapTupleIsValid(tuple))
 	{
@@ -1268,7 +1272,7 @@ DefineIndex(Oid relationId,
 			 * btree opclasses; if there are ever any other index types that
 			 * support unique indexes, this logic will need extension.
 			 */
-			if (accessMethodId == BTREE_AM_OID)
+			if (IsIndexAccessMethod(accessMethodId, BTREE_AM_OID))
 				eq_strategy = BTEqualStrategyNumber;
 			else
 				ereport(ERROR,

src/backend/commands/matview.c

@@ -15,6 +15,7 @@
  */
 #include "postgres.h"
 
+#include "access/amapi.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/htup_details.h"
@@ -908,7 +909,7 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
 				if (!HeapTupleIsValid(cla_ht))
 					elog(ERROR, "cache lookup failed for opclass %u", opclass);
 				cla_tup = (Form_pg_opclass) GETSTRUCT(cla_ht);
-				Assert(cla_tup->opcmethod == BTREE_AM_OID);
+				Assert(IsIndexAccessMethod(cla_tup->opcmethod, BTREE_AM_OID));
 				opfamily = cla_tup->opcfamily;
 				opcintype = cla_tup->opcintype;
 				ReleaseSysCache(cla_ht);
@@ -1068,7 +1069,7 @@ is_usable_unique_index(Relation indexRel)
 	 */
 	if (indexStruct->indisunique &&
 		indexStruct->indimmediate &&
-		indexRel->rd_rel->relam == BTREE_AM_OID &&
+        IsIndexAccessMethod(indexRel->rd_rel->relam, BTREE_AM_OID) &&
 		indexStruct->indisvalid &&
 		RelationGetIndexPredicate(indexRel) == NIL &&
 		indexStruct->indnatts > 0)

src/backend/commands/opclasscmds.c

@@ -595,7 +595,7 @@ DefineOpClass(CreateOpClassStmt *stmt)
 	if (OidIsValid(storageoid))
 	{
 		/* Just drop the spec if same as column datatype */
-		if (storageoid == typeoid)
+		if (storageoid == typeoid && !amstorage)
 			storageoid = InvalidOid;
 		else if (!amstorage)
 			ereport(ERROR,
@@ -1289,7 +1289,7 @@ assignProcTypes(OpFamilyMember *member, Oid amoid, Oid typeoid,
 	 * returning int4, while proc 2 must be a 2-arg proc returning int8.
 	 * Otherwise we don't know.
 	 */
-	else if (amoid == BTREE_AM_OID)
+	else if (IsIndexAccessMethod(amoid, BTREE_AM_OID))
 	{
 		if (member->number == BTORDER_PROC)
 		{
@@ -1372,7 +1372,7 @@ assignProcTypes(OpFamilyMember *member, Oid amoid, Oid typeoid,
 						 errmsg("btree equal image functions must not be cross-type")));
 		}
 	}
-	else if (amoid == HASH_AM_OID)
+	else if (IsIndexAccessMethod(amoid, HASH_AM_OID))
 	{
 		if (member->number == HASHSTANDARD_PROC)
 		{

src/backend/commands/tablecmds.c

@@ -10751,7 +10751,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
 		 * strategy number is equality.  (Is it reasonable to insist that
 		 * every such index AM use btree's number for equality?)
 		 */
-		if (amid != BTREE_AM_OID)
+		if (!IsIndexAccessMethod(amid, BTREE_AM_OID))
 			elog(ERROR, "only b-tree indexes are supported for foreign keys");
 		eqstrategy = BTEqualStrategyNumber;
 

src/backend/executor/nodeIndexscan.c

@@ -1378,7 +1378,7 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
 				 * We have to look up the operator's associated btree support
 				 * function
 				 */
-				if (index->rd_rel->relam != BTREE_AM_OID ||
+				if (!IsIndexAccessMethod(index->rd_rel->relam, BTREE_AM_OID) ||
 					varattno < 1 || varattno > indnkeyatts)
 					elog(ERROR, "bogus RowCompare index qualification");
 				opfamily = index->rd_opfamily[varattno - 1];

src/backend/mock.mk

@@ -37,9 +37,6 @@ EXCL_OBJS=\
 # of the test programs. Feel free to link them back (i.e. remove them from
 # this exclusion list) as needed.
 EXCL_OBJS+=\
-	src/backend/access/hash/hash.o \
-	src/backend/access/hash/hashsearch.o \
-	\
 	src/backend/utils/adt/cash.o \
 	src/backend/utils/adt/char.o \
 	src/backend/utils/adt/complex_type.o \

src/backend/optimizer/path/indxpath.c

@@ -2830,7 +2830,7 @@ match_rowcompare_to_indexcol(PlannerInfo *root,
 	Oid			expr_coll;
 
 	/* Forget it if we're not dealing with a btree index */
-	if (index->relam != BTREE_AM_OID)
+	if (!IsIndexAccessMethod(index->relam, BTREE_AM_OID))
 		return NULL;
 
 	index_relid = index->rel->relid;
@@ -3529,7 +3529,7 @@ ec_member_matches_indexcol(PlannerInfo *root, RelOptInfo *rel,
 	 * generate_implied_equalities_for_column; see
 	 * match_eclass_clauses_to_index.
 	 */
-	if (index->relam == BTREE_AM_OID &&
+	if (IsIndexAccessMethod(index->relam, BTREE_AM_OID) &&
 		!list_member_oid(ec->ec_opfamilies, curFamily))
 		return false;
 

src/backend/optimizer/util/pathnode.c

@@ -19,6 +19,7 @@
 
 #include <math.h>
 
+#include "access/amapi.h"
 #include "foreign/fdwapi.h"
 #include "miscadmin.h"
 #include "nodes/extensible.h"
@@ -1124,7 +1125,7 @@ create_index_path(PlannerInfo *root,
 														  required_outer);
 	pathnode->path.parallel_aware = false;
 	/* GPDB_12_MERGE_FEATURE_NOT_SUPPORTED: the parallel StreamBitmap scan is not implemented */
-	pathnode->path.parallel_safe = rel->consider_parallel && (index->relam != BITMAP_AM_OID);
+	pathnode->path.parallel_safe = rel->consider_parallel && !IsIndexAccessMethod(index->relam, BITMAP_AM_OID);
 	pathnode->path.parallel_workers = 0;
 	pathnode->path.pathkeys = pathkeys;
 

src/backend/optimizer/util/plancat.c

@@ -311,7 +311,7 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
 			/*
 			 * Fetch the ordering information for the index, if any.
 			 */
-			if (info->relam == BTREE_AM_OID)
+			if (IsIndexAccessMethod(info->relam, BTREE_AM_OID))
 			{
 				/*
 				 * If it's a btree index, we can use its opfamily OIDs
@@ -436,7 +436,7 @@ get_relation_info(PlannerInfo *root, Oid relationObjectId, bool inhparent,
 				info->tuples > rel->tuples)
 				info->tuples = rel->tuples;
 
-			if (info->relam == BTREE_AM_OID)
+			if (IsIndexAccessMethod(info->relam, BTREE_AM_OID))
 			{
 				/* For btrees, get tree height while we have the index open */
 				info->tree_height = _bt_getrootheight(indexRelation);

src/backend/parser/gram.y

@@ -10165,7 +10165,7 @@ opt_index_name:
 
 access_method_clause:
 			USING name								{ $$ = $2; }
-			| /*EMPTY*/								{ $$ = DEFAULT_INDEX_TYPE; }
+			| /*EMPTY*/								{ $$ = NULL; }
 		;
 
 index_params:	index_elem							{ $$ = list_make1($1); }

src/backend/parser/parse_utilcmd.c

@@ -3289,7 +3289,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 		index->idxname = NULL;	/* DefineIndex will choose name */
 
 	index->relation = cxt->relation;
-	index->accessMethod = constraint->access_method ? constraint->access_method : DEFAULT_INDEX_TYPE;
+	index->accessMethod = constraint->access_method ? constraint->access_method : default_index_access_method;
 	index->options = constraint->options;
 	index->tableSpace = constraint->indexspace;
 	index->whereClause = constraint->where_clause;
@@ -3413,7 +3413,7 @@ transformIndexConstraint(Constraint *constraint, CreateStmtContext *cxt)
 		 * else dump and reload will produce a different index (breaking
 		 * pg_upgrade in particular).
 		 */
-		if (index_rel->rd_rel->relam != get_index_am_oid(DEFAULT_INDEX_TYPE, false))
+		if (!IsIndexAccessMethod(index_rel->rd_rel->relam, BTREE_AM_OID))
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 					 errmsg("index \"%s\" is not a btree", index_name),

src/backend/utils/adt/selfuncs.c

@@ -6229,7 +6229,7 @@ get_actual_variable_range(PlannerInfo *root, VariableStatData *vardata,
 		ScanDirection indexscandir;
 
 		/* Ignore non-btree indexes */
-		if (index->relam != BTREE_AM_OID)
+		if (!IsIndexAccessMethod(index->relam, BTREE_AM_OID))
 			continue;
 
 		/*

src/backend/utils/cache/lsyscache.c

@@ -866,8 +866,8 @@ equality_ops_are_compatible(Oid opno1, Oid opno2)
 		Form_pg_amop op_form = (Form_pg_amop) GETSTRUCT(op_tuple);
 
 		/* must be btree or hash */
-		if (op_form->amopmethod == BTREE_AM_OID ||
-			op_form->amopmethod == HASH_AM_OID)
+		if (IsIndexAccessMethod(op_form->amopmethod, BTREE_AM_OID) ||
+			IsIndexAccessMethod(op_form->amopmethod, HASH_AM_OID))
 		{
 			if (op_in_opfamily(opno2, op_form->amopfamily))
 			{

src/backend/utils/misc/guc.c

@@ -48,6 +48,7 @@
 #include "catalog/pg_authid.h"
 #include "catalog/pg_profile.h"
 #include "catalog/storage.h"
+#include "catalog/index.h"
 #include "commands/async.h"
 #include "commands/prepare.h"
 #include "commands/trigger.h"

src/backend/utils/misc/guc_gp.c

@@ -58,6 +58,7 @@
 #include "utils/resource_manager.h"
 #include "utils/varlena.h"
 #include "utils/vmem_tracker.h"
+#include "catalog/index.h"
 
 /*
  * These constants are copied from guc.c. They should not bitrot when we
@@ -4572,6 +4573,17 @@ struct config_string ConfigureNamesString_gp[] =
 		GP_VERSION,
 		NULL, NULL, NULL
 	},
+
+	{
+		{"default_index_access_method", PGC_USERSET, CLIENT_CONN_STATEMENT,
+		 gettext_noop("Sets the default index access method."),
+		 NULL,
+		 GUC_IS_NAME
+		},
+		&default_index_access_method,
+		DEFAULT_INDEX_TYPE,
+		check_default_index_access_method, NULL, NULL
+	},
 #ifndef USE_INTERNAL_FTS
 	{
 		{"gp_etcd_account_id", PGC_BACKEND, CUSTOM_OPTIONS,