Add reloption support for custom table access method

src/backend/access/aocs/aocsam_handler.c

@@ -19,6 +19,7 @@
 #include "access/appendonlywriter.h"
 #include "access/heapam.h"
 #include "access/multixact.h"
+#include "access/reloptions.h"
 #include "access/tableam.h"
 #include "access/tsmapi.h"
 #include "access/xact.h"
@@ -2254,7 +2255,9 @@ static TableAmRoutine ao_column_methods = {
 	.scan_bitmap_next_block = aoco_scan_bitmap_next_block,
 	.scan_bitmap_next_tuple = aoco_scan_bitmap_next_tuple,
 	.scan_sample_next_block = aoco_scan_sample_next_block,
-	.scan_sample_next_tuple = aoco_scan_sample_next_tuple
+	.scan_sample_next_tuple = aoco_scan_sample_next_tuple,
+
+	.amoptions = ao_amoptions,
 };
 
 Datum

src/backend/access/appendonly/appendonlyam_handler.c

@@ -22,6 +22,7 @@
 #include "access/heapam.h"
 #include "access/heaptoast.h"
 #include "access/multixact.h"
+#include "access/reloptions.h"
 #include "access/tableam.h"
 #include "access/tsmapi.h"
 #include "access/xact.h"
@@ -2381,7 +2382,9 @@ static const TableAmRoutine ao_row_methods = {
 	.scan_bitmap_next_block = appendonly_scan_bitmap_next_block,
 	.scan_bitmap_next_tuple = appendonly_scan_bitmap_next_tuple,
 	.scan_sample_next_block = appendonly_scan_sample_next_block,
-	.scan_sample_next_tuple = appendonly_scan_sample_next_tuple
+	.scan_sample_next_tuple = appendonly_scan_sample_next_tuple,
+
+	.amoptions = ao_amoptions,
 };
 
 Datum

src/backend/access/common/reloptions.c

@@ -1381,7 +1381,7 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				  reloption_function amoptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1403,7 +1403,7 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			options = heap_reloptions(classForm->relkind, datum, false);
+			options = table_reloptions((tamoptions_function)amoptions, datum, classForm->relkind, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
@@ -1413,7 +1413,7 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
-			options = index_reloptions(amoptions, datum, false);
+			options = index_reloptions((amoptions_function)amoptions, datum, false);
 			break;
 		case RELKIND_FOREIGN_TABLE:
 			options = NULL;
@@ -2024,39 +2024,32 @@ view_reloptions(Datum reloptions, bool validate)
 									  tab, lengthof(tab));
 }
 
-/*
- * Parse options for heaps, views and toast tables.
- */
 bytea *
-heap_reloptions(char relkind, Datum reloptions, bool validate)
+table_reloptions(tamoptions_function amoptions, Datum reloptions, char relkind, bool validate)
 {
-	StdRdOptions *rdopts;
-
-	switch (relkind)
+	Assert(relkind == RELKIND_RELATION ||
+			relkind == RELKIND_TOASTVALUE ||
+			relkind == RELKIND_MATVIEW);
+	if (amoptions == NULL)
 	{
-		case RELKIND_TOASTVALUE:
-			rdopts = (StdRdOptions *)
-				default_reloptions(reloptions, validate, RELOPT_KIND_TOAST);
-			if (rdopts != NULL)
-			{
-				/* adjust default-only parameters for TOAST relations */
-				rdopts->fillfactor = 100;
-				rdopts->autovacuum.analyze_threshold = -1;
-				rdopts->autovacuum.analyze_scale_factor = -1;
-			}
-			return (bytea *) rdopts;
-		case RELKIND_RELATION:
-		case RELKIND_MATVIEW:
-			/*
-			 * GPDB_12_AFTER_MERGE_FIXME: should we accept AO-related options for
-			 * partitioned tables? A partitioned table has no data, but the options
-			 * might be inherited by partitions.
-			 */
-			return default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
-		default:
-			/* other relkinds are not supported */
-			return NULL;
+		if (PointerIsValid(DatumGetPointer(reloptions)))
+			elog(ERROR, "table access method doesn't supported reloptions");
+		return NULL;
 	}
+	return amoptions(reloptions, relkind, validate);
+}
+
+bytea *
+table_reloptions_am(Oid accessMethodId, Datum reloptions, char relkind, bool validate)
+{
+	const TableAmRoutine *tam;
+
+	Assert(relkind == RELKIND_RELATION ||
+			relkind == RELKIND_TOASTVALUE ||
+			relkind == RELKIND_MATVIEW);
+
+	tam = GetTableAmRoutineByAmId(accessMethodId);
+	return table_reloptions(tam->amoptions, reloptions, relkind, validate);
 }
 
 /*

src/backend/access/common/reloptions_gp.c

@@ -1697,3 +1697,30 @@ List* transformColumnEncoding(Relation rel, List *colDefs, List *stenc, List *wi
 
 	return result;
 }
+
+bytea *
+ao_amoptions(Datum reloptions, char relkind, bool validate)
+{
+	StdRdOptions *rdopts;
+
+	switch (relkind)
+	{
+		case RELKIND_TOASTVALUE:
+			rdopts = (StdRdOptions *)
+				default_reloptions(reloptions, validate, RELOPT_KIND_TOAST);
+			if (rdopts != NULL)
+			{
+				/* adjust default-only parameters for TOAST relations */
+				rdopts->fillfactor = 100;
+				rdopts->autovacuum.analyze_threshold = -1;
+				rdopts->autovacuum.analyze_scale_factor = -1;
+			}
+			return (bytea *) rdopts;
+		case RELKIND_RELATION:
+		case RELKIND_MATVIEW:
+			return default_reloptions(reloptions, validate, RELOPT_KIND_APPENDOPTIMIZED);
+		default:
+			Assert(false);
+			return NULL;
+	}
+}

src/backend/access/heap/heapam_handler.c

@@ -23,6 +23,7 @@
 #include "access/heapam.h"
 #include "access/heaptoast.h"
 #include "access/multixact.h"
+#include "access/reloptions.h"
 #include "access/rewriteheap.h"
 #include "access/syncscan.h"
 #include "access/tableam.h"
@@ -2548,6 +2549,36 @@ SampleHeapTupleVisible(TableScanDesc scan, Buffer buffer,
 	}
 }
 
+/*
+ * Parse options for heaps, views and toast tables.
+ */
+static bytea *
+heapam_amoptions(Datum reloptions, char relkind, bool validate)
+{
+	StdRdOptions *rdopts;
+
+	switch (relkind)
+	{
+		case RELKIND_TOASTVALUE:
+			rdopts = (StdRdOptions *)
+				default_reloptions(reloptions, validate, RELOPT_KIND_TOAST);
+			if (rdopts != NULL)
+			{
+				/* adjust default-only parameters for TOAST relations */
+				rdopts->fillfactor = 100;
+				rdopts->autovacuum.analyze_threshold = -1;
+				rdopts->autovacuum.analyze_scale_factor = -1;
+			}
+			return (bytea *) rdopts;
+		case RELKIND_RELATION:
+		case RELKIND_MATVIEW:
+			return default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
+		default:
+			Assert(false);
+			return NULL;
+	}
+}
+
 
 /* ------------------------------------------------------------------------
  * Definition of the heap table access method.
@@ -2610,7 +2641,10 @@ static const TableAmRoutine heapam_methods = {
 	.scan_bitmap_next_block = heapam_scan_bitmap_next_block,
 	.scan_bitmap_next_tuple = heapam_scan_bitmap_next_tuple,
 	.scan_sample_next_block = heapam_scan_sample_next_block,
-	.scan_sample_next_tuple = heapam_scan_sample_next_tuple
+	.scan_sample_next_tuple = heapam_scan_sample_next_tuple,
+
+	.amoptions	= heapam_amoptions,
+
 };
 
 

src/backend/access/table/tableamapi.c

@@ -106,6 +106,39 @@ GetTableAmRoutine(Oid amhandler)
 	return routine;
 }
 
+/*
+ * GetTableAmRoutineByAmId - look up the handler of the table access method
+ * with the given OID, and get its TableAmRoutine struct.
+ *
+ * If the given OID isn't a valid index access method, throws error.
+ */
+const TableAmRoutine *
+GetTableAmRoutineByAmId(Oid amoid)
+{
+	HeapTuple	tuple;
+	Form_pg_am  amform;
+	regproc     amhandler;
+
+	if (amoid == HEAP_TABLE_AM_OID)
+		return GetHeapamTableAmRoutine();
+
+	tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(amoid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR, (errmsg("cache lookup failed for access method %u", amoid)));
+
+	amform = (Form_pg_am) GETSTRUCT(tuple);
+	if (amform->amtype != AMTYPE_TABLE)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("access method \"%s\" is not of type TABLE",
+						NameStr(amform->amname))));
+
+	amhandler = amform->amhandler;
+	ReleaseSysCache(tuple);
+
+	return GetTableAmRoutine(amhandler);
+}
+
 /* check_hook: validate new default_table_access_method */
 bool
 check_default_table_access_method(char **newval, void **extra, GucSource source)

src/backend/catalog/toasting.c

@@ -15,6 +15,7 @@
 #include "postgres.h"
 
 #include "access/heapam.h"
+#include "access/reloptions.h"
 #include "access/toast_compression.h"
 #include "access/xact.h"
 #include "catalog/binary_upgrade.h"
@@ -152,6 +153,11 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
 	ObjectAddress baseobject,
 				toastobject;
 
+	Assert(rel->rd_tableam || rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
+	if (rel->rd_tableam && table_relation_needs_toast_table(rel))
+		(void) table_reloptions_am(table_relation_toast_am(rel), reloptions,
+									RELKIND_TOASTVALUE, true);
+
 	/*
 	 * Is it already toasted?
 	 */

src/backend/commands/createas.c

@@ -216,8 +216,6 @@ create_ctas_internal(List *attrList, IntoClause *into, QueryDesc *queryDesc, boo
 										validnsps,
 										true, false);
 
-	(void) heap_reloptions(RELKIND_TOASTVALUE, toast_options, true);
-
 	NewRelationCreateToastTable(intoRelationAddr.objectId, toast_options);
 
 	/* Create the "view" part of a materialized view. */

src/backend/commands/tablecmds.c

@@ -825,7 +825,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		amHandlerOid = get_table_am_handler_oid(accessMethod, false);
 	}
 
-
 	/*
 	 * Parse and validate reloptions, if any.
 	 */
@@ -871,8 +870,13 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		case RELKIND_PARTITIONED_TABLE:
 			(void) partitioned_table_reloptions(reloptions, true);
 			break;
+		case RELKIND_RELATION:
+		case RELKIND_MATVIEW:
+		case RELKIND_TOASTVALUE:
+			(void) table_reloptions_am(accessMethodId, reloptions, relkind, true);
+			break;
 		default:
-			(void) heap_reloptions(relkind, reloptions, true);
+			break;
 	}
 
 	if (stmt->ofTypename)
@@ -15683,10 +15687,7 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 		case RELKIND_AOSEGMENTS:
 		case RELKIND_AOBLOCKDIR:
 		case RELKIND_AOVISIMAP:
-			if (RelationIsAppendOptimized(rel))
-				(void) default_reloptions(newOptions, true, RELOPT_KIND_APPENDOPTIMIZED);
-			else
-				(void) heap_reloptions(rel->rd_rel->relkind, newOptions, true);
+			(void) table_reloptions(rel->rd_tableam->amoptions, newOptions, rel->rd_rel->relkind, true);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			(void) partitioned_table_reloptions(newOptions, true);
@@ -15798,7 +15799,7 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 										 defList, "toast", validnsps, false,
 										 operation == AT_ResetRelOptions);
 
-		(void) heap_reloptions(RELKIND_TOASTVALUE, newOptions, true);
+		(void) table_reloptions(toastrel->rd_tableam->amoptions, newOptions, RELKIND_TOASTVALUE, true);
 
 		memset(repl_val, 0, sizeof(repl_val));
 		memset(repl_null, false, sizeof(repl_null));

src/backend/postmaster/autovacuum.c

@@ -122,6 +122,7 @@
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/namespace.h"
+#include "catalog/pg_am.h"
 #include "catalog/pg_database.h"
 #include "commands/dbcommands.h"
 #include "commands/vacuum.h"
@@ -2853,16 +2854,26 @@ extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc)
 {
 	bytea	   *relopts;
 	AutoVacOpts *av;
+	Oid relam;
+	const TableAmRoutine *tam;
 
 	Assert(((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_RELATION ||
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_MATVIEW ||
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_TOASTVALUE ||
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_AOSEGMENTS ||
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_AOBLOCKDIR ||
-		   ((Form_pg_class) GETSTRUCT(tup))->relkind ==  RELKIND_AOVISIMAP);
+		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_AOVISIMAP);
 
+	relam = ((Form_pg_class) GETSTRUCT(tup))->relam;
+	tam = GetTableAmRoutineByAmId(relam);
 
-	relopts = extractRelOptions(tup, pg_class_desc, NULL);
+	/* FIXME: external TAM may have reloption other than StdRdOptions. */
+	if (relam != HEAP_TABLE_AM_OID &&
+		relam != AO_ROW_TABLE_AM_OID &&
+		relam != AO_COLUMN_TABLE_AM_OID)
+		return NULL;
+
+	relopts = extractRelOptions(tup, pg_class_desc, tam->amoptions);
 	if (relopts == NULL)
 		return NULL;
 

src/backend/tcop/utility.c

@@ -1489,9 +1489,6 @@ ProcessUtilitySlow(ParseState *pstate,
 																	validnsps,
 																	true,
 																	false);
-								(void) heap_reloptions(RELKIND_TOASTVALUE,
-													   toast_options,
-													   true);
 
 								NewRelationCreateToastTable(address.objectId,
 															toast_options);

src/backend/utils/cache/relcache.c

@@ -480,7 +480,7 @@ static void
 RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
-	amoptions_function amoptsfn;
+	reloption_function amoptsfn;
 
 	relation->rd_options = NULL;
 
@@ -492,11 +492,13 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
+		case RELKIND_MATVIEW:
 		case RELKIND_AOSEGMENTS:
 		case RELKIND_AOBLOCKDIR:
 		case RELKIND_AOVISIMAP:
+			amoptsfn = relation->rd_tableam->amoptions;
+			break;
 		case RELKIND_VIEW:
-		case RELKIND_MATVIEW:
 		case RELKIND_PARTITIONED_TABLE:
 			amoptsfn = NULL;
 			break;