Custom reloptions for table AM

Let table AM define custom reloptions for its tables. This allows specifying
AM-specific parameters by the WITH clause when creating a table.

The reloptions, which could be used outside of table AM, are now extracted
into the CommonRdOptions data structure.  These options could be by decision
of table AM directly specified by a user or calculated in some way.

The new test module test_tam_options evaluates the ability to set up custom
reloptions and calculate fields of CommonRdOptions on their base.

The code may use some parts from prior work by Hao Wu.

Discussion: https://postgr.es/m/CAPpHfdurb9ycV8udYqM%3Do0sPS66PJ4RCBM1g-bBpvzUfogY0EA%40mail.gmail.com
Discussion: https://postgr.es/m/AMUA1wBBBxfc3tKRLLdU64rb.1.1683276279979.Hmail.wuhao%40hashdata.cn
Reviewed-by: Reviewed-by: Pavel Borisov, Matthias van de Meent, Jess Davis

9 files changed, 141 insertions, 45 deletions

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d6eb5d85599..c1de092a42d 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -24,6 +24,7 @@
 #include "access/nbtree.h"
 #include "access/reloptions.h"
 #include "access/spgist_private.h"
+#include "access/tableam.h"
 #include "catalog/pg_type.h"
 #include "commands/defrem.h"
 #include "commands/tablespace.h"
@@ -44,7 +45,7 @@
  * value, upper and lower bounds (if applicable); for strings, consider a
  * validation routine.
  * (ii) add a record below (or use add_<type>_reloption).
- * (iii) add it to the appropriate options struct (perhaps StdRdOptions)
+ * (iii) add it to the appropriate options struct (perhaps HeapRdOptions)
  * (iv) add it to the appropriate handling routine (perhaps
  * default_reloptions)
  * (v) make sure the lock level is set correctly for that operation
@@ -1374,10 +1375,16 @@ untransformRelOptions(Datum options)
  * tupdesc is pg_class' tuple descriptor.  amoptions is a pointer to the index
  * AM's options parser function in the case of a tuple corresponding to an
  * index, or NULL otherwise.
+ *
+ * If common pointer is provided, then the corresponding struct will be
+ * filled with options that table AM exposes for external usage.  That must
+ * be filled with defaults before passing here.
  */
+
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				  const TableAmRoutine *tableam, amoptions_function amoptions,
+				  CommonRdOptions *common)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1399,7 +1406,8 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			options = heap_reloptions(classForm->relkind, datum, false);
+			options = tableam_reloptions(tableam, classForm->relkind,
+										 datum, common, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
@@ -1695,7 +1703,7 @@ parse_one_reloption(relopt_value *option, char *text_str, int text_len,
  * Given the result from parseRelOptions, allocate a struct that's of the
  * specified base size plus any extra space that's needed for string variables.
  *
- * "base" should be sizeof(struct) of the reloptions struct (StdRdOptions or
+ * "base" should be sizeof(struct) of the reloptions struct (HeapRdOptions or
  * equivalent).
  */
 static void *
@@ -1832,59 +1840,95 @@ fillRelOptions(void *rdopts, Size basesize,
 
 
 /*
- * Option parser for anything that uses StdRdOptions.
+ * Option parser for anything that uses HeapRdOptions.
  */
-bytea *
+static bytea *
 default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 {
 	static const relopt_parse_elt tab[] = {
-		{"fillfactor", RELOPT_TYPE_INT, offsetof(StdRdOptions, fillfactor)},
+		{"fillfactor", RELOPT_TYPE_INT, offsetof(HeapRdOptions, fillfactor)},
 		{"autovacuum_enabled", RELOPT_TYPE_BOOL,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, enabled)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, enabled)},
 		{"autovacuum_vacuum_threshold", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_threshold)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_threshold)},
 		{"autovacuum_vacuum_insert_threshold", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_ins_threshold)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_ins_threshold)},
 		{"autovacuum_analyze_threshold", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, analyze_threshold)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, analyze_threshold)},
 		{"autovacuum_vacuum_cost_limit", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_cost_limit)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_cost_limit)},
 		{"autovacuum_freeze_min_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, freeze_min_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, freeze_min_age)},
 		{"autovacuum_freeze_max_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, freeze_max_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, freeze_max_age)},
 		{"autovacuum_freeze_table_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, freeze_table_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, freeze_table_age)},
 		{"autovacuum_multixact_freeze_min_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, multixact_freeze_min_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, multixact_freeze_min_age)},
 		{"autovacuum_multixact_freeze_max_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, multixact_freeze_max_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, multixact_freeze_max_age)},
 		{"autovacuum_multixact_freeze_table_age", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, multixact_freeze_table_age)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, multixact_freeze_table_age)},
 		{"log_autovacuum_min_duration", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, log_min_duration)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, log_min_duration)},
 		{"toast_tuple_target", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, toast_tuple_target)},
+		offsetof(HeapRdOptions, toast_tuple_target)},
 		{"autovacuum_vacuum_cost_delay", RELOPT_TYPE_REAL,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_cost_delay)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_cost_delay)},
 		{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_scale_factor)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_scale_factor)},
 		{"autovacuum_vacuum_insert_scale_factor", RELOPT_TYPE_REAL,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, vacuum_ins_scale_factor)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, vacuum_ins_scale_factor)},
 		{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
-		offsetof(StdRdOptions, autovacuum) + offsetof(AutoVacOpts, analyze_scale_factor)},
+			offsetof(HeapRdOptions, common) +
+			offsetof(CommonRdOptions, autovacuum) +
+		offsetof(AutoVacOpts, analyze_scale_factor)},
 		{"user_catalog_table", RELOPT_TYPE_BOOL,
-		offsetof(StdRdOptions, user_catalog_table)},
+			offsetof(HeapRdOptions, common) +
+		offsetof(CommonRdOptions, user_catalog_table)},
 		{"parallel_workers", RELOPT_TYPE_INT,
-		offsetof(StdRdOptions, parallel_workers)},
+			offsetof(HeapRdOptions, common) +
+		offsetof(CommonRdOptions, parallel_workers)},
 		{"vacuum_index_cleanup", RELOPT_TYPE_ENUM,
-		offsetof(StdRdOptions, vacuum_index_cleanup)},
+			offsetof(HeapRdOptions, common) +
+		offsetof(CommonRdOptions, vacuum_index_cleanup)},
 		{"vacuum_truncate", RELOPT_TYPE_BOOL,
-		offsetof(StdRdOptions, vacuum_truncate)}
+			offsetof(HeapRdOptions, common) +
+		offsetof(CommonRdOptions, vacuum_truncate)}
 	};
 
 	return (bytea *) build_reloptions(reloptions, validate, kind,
-									  sizeof(StdRdOptions),
+									  sizeof(HeapRdOptions),
 									  tab, lengthof(tab));
 }
 
@@ -2016,26 +2060,33 @@ view_reloptions(Datum reloptions, bool validate)
  * Parse options for heaps, views and toast tables.
  */
 bytea *
-heap_reloptions(char relkind, Datum reloptions, bool validate)
+heap_reloptions(char relkind, Datum reloptions,
+				CommonRdOptions *common, bool validate)
 {
-	StdRdOptions *rdopts;
+	HeapRdOptions *rdopts;
 
 	switch (relkind)
 	{
 		case RELKIND_TOASTVALUE:
-			rdopts = (StdRdOptions *)
+			rdopts = (HeapRdOptions *)
 				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;
+				rdopts->common.autovacuum.analyze_threshold = -1;
+				rdopts->common.autovacuum.analyze_scale_factor = -1;
 			}
+			if (rdopts != NULL && common != NULL)
+				*common = rdopts->common;
 			return (bytea *) rdopts;
 		case RELKIND_RELATION:
 		case RELKIND_MATVIEW:
-			return default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
+			rdopts = (HeapRdOptions *)
+				default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
+			if (rdopts != NULL && common != NULL)
+				*common = rdopts->common;
+			return (bytea *) rdopts;
 		default:
 			/* other relkinds are not supported */
 			return NULL;
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 08f6a10ac3d..4607505106d 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2279,8 +2279,8 @@ heap_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
 	Assert(!(options & HEAP_INSERT_NO_LOGICAL));
 
 	needwal = RelationNeedsWAL(relation);
-	saveFreeSpace = RelationGetTargetPageFreeSpace(relation,
-												   HEAP_DEFAULT_FILLFACTOR);
+	saveFreeSpace = HeapGetTargetPageFreeSpace(relation,
+											   HEAP_DEFAULT_FILLFACTOR);
 
 	/* Toast and set header data in all the slots */
 	heaptuples = palloc(ntuples * sizeof(HeapTuple));
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index cc4d51d5514..6b1d2dd7a48 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/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"
@@ -2161,6 +2162,17 @@ heapam_relation_toast_am(Relation rel)
 	return rel->rd_rel->relam;
 }
 
+static bytea *
+heapam_reloptions(char relkind, Datum reloptions,
+				  CommonRdOptions *common, bool validate)
+{
+	Assert(relkind == RELKIND_RELATION ||
+		   relkind == RELKIND_TOASTVALUE ||
+		   relkind == RELKIND_MATVIEW);
+
+	return heap_reloptions(relkind, reloptions, common, validate);
+}
+
 
 /* ------------------------------------------------------------------------
  * Planner related callbacks for the heap AM
@@ -2710,6 +2722,7 @@ static const TableAmRoutine heapam_methods = {
 	.relation_needs_toast_table = heapam_relation_needs_toast_table,
 	.relation_toast_am = heapam_relation_toast_am,
 	.relation_fetch_toast_slice = heap_fetch_toast_slice,
+	.reloptions = heapam_reloptions,
 
 	.relation_estimate_size = heapam_estimate_rel_size,
 
diff --git a/src/backend/access/heap/heaptoast.c b/src/backend/access/heap/heaptoast.c
index a420e165304..cffee7e1b7a 100644
--- a/src/backend/access/heap/heaptoast.c
+++ b/src/backend/access/heap/heaptoast.c
@@ -32,6 +32,13 @@
 #include "access/toast_internals.h"
 #include "utils/fmgroids.h"
 
+/*
+ * HeapGetToastTupleTarget
+ *      Returns the heap relation's toast_tuple_target.  Note multiple eval of argument!
+ */
+#define HeapGetToastTupleTarget(relation, defaulttarg) \
+		((HeapRdOptions *) (relation)->rd_options ? \
+		((HeapRdOptions *) (relation)->rd_options)->toast_tuple_target : (defaulttarg))
 
 /* ----------
  * heap_toast_delete -
@@ -174,7 +181,7 @@ heap_toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		hoff += BITMAPLEN(numAttrs);
 	hoff = MAXALIGN(hoff);
 	/* now convert to a limit on the tuple data size */
-	maxDataLen = RelationGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
+	maxDataLen = HeapGetToastTupleTarget(rel, TOAST_TUPLE_TARGET) - hoff;
 
 	/*
 	 * Look for attributes with attstorage EXTENDED to compress.  Also find
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index 7c662cdf46e..ed731179b45 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -536,8 +536,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 						len, MaxHeapTupleSize)));
 
 	/* Compute desired extra freespace due to fillfactor option */
-	saveFreeSpace = RelationGetTargetPageFreeSpace(relation,
-												   HEAP_DEFAULT_FILLFACTOR);
+	saveFreeSpace = HeapGetTargetPageFreeSpace(relation,
+											   HEAP_DEFAULT_FILLFACTOR);
 
 	/*
 	 * Since pages without tuples can still have line pointers, we consider
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
index d2eecaf7ebc..20142dd2144 100644
--- a/src/backend/access/heap/pruneheap.c
+++ b/src/backend/access/heap/pruneheap.c
@@ -235,8 +235,8 @@ heap_page_prune_opt(Relation relation, Buffer buffer)
 	 * important than sometimes getting a wrong answer in what is after all
 	 * just a heuristic estimate.
 	 */
-	minfree = RelationGetTargetPageFreeSpace(relation,
-											 HEAP_DEFAULT_FILLFACTOR);
+	minfree = HeapGetTargetPageFreeSpace(relation,
+										 HEAP_DEFAULT_FILLFACTOR);
 	minfree = Max(minfree, BLCKSZ / 10);
 
 	if (PageIsFull(page) || PageGetHeapFreeSpace(page) < minfree)
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index 473f3aa9bef..2bbf121146e 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -641,8 +641,8 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 						len, MaxHeapTupleSize)));
 
 	/* Compute desired extra freespace due to fillfactor option */
-	saveFreeSpace = RelationGetTargetPageFreeSpace(state->rs_new_rel,
-												   HEAP_DEFAULT_FILLFACTOR);
+	saveFreeSpace = HeapGetTargetPageFreeSpace(state->rs_new_rel,
+											   HEAP_DEFAULT_FILLFACTOR);
 
 	/* Now we can check to see if there's enough free space already. */
 	page = (Page) state->rs_buffer;
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index 805d222cebc..5d5f0e68fd7 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -750,7 +750,7 @@ table_block_relation_estimate_size(Relation rel, int32 *attr_widths,
 		 * The other branch considers it implicitly by calculating density
 		 * from actual relpages/reltuples statistics.
 		 */
-		fillfactor = RelationGetFillFactor(rel, HEAP_DEFAULT_FILLFACTOR);
+		fillfactor = HeapGetFillFactor(rel, HEAP_DEFAULT_FILLFACTOR);
 
 		tuple_width = get_rel_data_width(rel, attr_widths);
 		tuple_width += overhead_bytes_per_tuple;
diff --git a/src/backend/access/table/tableamapi.c b/src/backend/access/table/tableamapi.c
index 55b8caeadf2..d9e23ef3175 100644
--- a/src/backend/access/table/tableamapi.c
+++ b/src/backend/access/table/tableamapi.c
@@ -13,9 +13,11 @@
 
 #include "access/tableam.h"
 #include "access/xact.h"
+#include "catalog/pg_am.h"
 #include "commands/defrem.h"
 #include "miscadmin.h"
 #include "utils/guc_hooks.h"
+#include "utils/syscache.h"
 
 
 /*
@@ -98,6 +100,29 @@ GetTableAmRoutine(Oid amhandler)
 	return routine;
 }
 
+/*
+ * GetTableAmRoutineByAmOid
+ *		Given the table access method oid get its TableAmRoutine struct, which
+ *		will be palloc'd in the caller's memory context.
+ */
+const TableAmRoutine *
+GetTableAmRoutineByAmOid(Oid amoid)
+{
+	HeapTuple	ht_am;
+	Form_pg_am	amrec;
+	const TableAmRoutine *tableam = NULL;
+
+	ht_am = SearchSysCache1(AMOID, ObjectIdGetDatum(amoid));
+	if (!HeapTupleIsValid(ht_am))
+		elog(ERROR, "cache lookup failed for access method %u",
+			 amoid);
+	amrec = (Form_pg_am) GETSTRUCT(ht_am);
+
+	tableam = GetTableAmRoutine(amrec->amhandler);
+	ReleaseSysCache(ht_am);
+	return tableam;
+}
+
 /* check_hook: validate new default_table_access_method */
 bool
 check_default_table_access_method(char **newval, void **extra, GucSource source)