From dd263df5bd0ddab823829fdbb52e7c76027fcea1 Mon Sep 17 00:00:00 2001
From: Alexander Korotkov <akorotkov@postgresql.org>
Date: Mon, 12 Jun 2023 23:16:01 +0300
Subject: [PATCH 2/8] Custom reloptions for table AM

Let table AM define custom reloptions for its tables.
---
 src/backend/access/common/reloptions.c   |  6 ++-
 src/backend/access/heap/heapam_handler.c | 12 ++++++
 src/backend/access/table/tableamapi.c    | 25 +++++++++++++
 src/backend/commands/tablecmds.c         | 47 ++++++++++++++----------
 src/backend/postmaster/autovacuum.c      |  4 +-
 src/backend/utils/cache/relcache.c       |  6 ++-
 src/include/access/reloptions.h          |  2 +
 src/include/access/tableam.h             | 43 ++++++++++++++++++++++
 8 files changed, 122 insertions(+), 23 deletions(-)

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d6eb5d85599..963995388bb 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"
@@ -1377,7 +1378,7 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				  const TableAmRoutine *tableam, amoptions_function amoptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1399,7 +1400,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, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 30ac13b9751..a586850ea2b 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -25,6 +25,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"
@@ -2436,6 +2437,16 @@ heapam_relation_toast_am(Relation rel)
 	return rel->rd_rel->relam;
 }
 
+static bytea *
+heapam_reloptions(char relkind, Datum reloptions, bool validate)
+{
+	Assert(relkind == RELKIND_RELATION ||
+		   relkind == RELKIND_TOASTVALUE ||
+		   relkind == RELKIND_MATVIEW);
+
+	return heap_reloptions(relkind, reloptions, validate);
+}
+
 
 /* ------------------------------------------------------------------------
  * Planner related callbacks for the heap AM
@@ -2941,6 +2952,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/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)
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 8a02c5b05b6..6fc815666bf 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -715,6 +715,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	ObjectAddress address;
 	LOCKMODE	parentLockmode;
 	Oid			accessMethodId = InvalidOid;
+	const TableAmRoutine *tableam = NULL;
 
 	/*
 	 * Truncate relname to appropriate length (probably a waste of time, as
@@ -850,6 +851,24 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
 
+	/*
+	 * Select access method to use: an explicitly indicated one, or (in the
+	 * case of a partitioned table) the parent's, if it has one.
+	 */
+	if (stmt->accessMethod != NULL)
+		accessMethodId = get_table_am_oid(stmt->accessMethod, false);
+	else if (stmt->partbound)
+	{
+		Assert(list_length(inheritOids) == 1);
+		accessMethodId = get_rel_relam(linitial_oid(inheritOids));
+	}
+	else
+		accessMethodId = InvalidOid;
+
+	/* still nothing? use the default */
+	if (RELKIND_HAS_TABLE_AM(relkind) && !OidIsValid(accessMethodId))
+		accessMethodId = get_table_am_oid(default_table_access_method, false);
+
 	/*
 	 * Parse and validate reloptions, if any.
 	 */
@@ -858,6 +877,12 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 
 	switch (relkind)
 	{
+		case RELKIND_RELATION:
+		case RELKIND_TOASTVALUE:
+		case RELKIND_MATVIEW:
+			tableam = GetTableAmRoutineByAmOid(accessMethodId);
+			(void) tableam_reloptions(tableam, relkind, reloptions, true);
+			break;
 		case RELKIND_VIEW:
 			(void) view_reloptions(reloptions, true);
 			break;
@@ -866,6 +891,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			break;
 		default:
 			(void) heap_reloptions(relkind, reloptions, true);
+			break;
 	}
 
 	if (stmt->ofTypename)
@@ -957,24 +983,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		}
 	}
 
-	/*
-	 * Select access method to use: an explicitly indicated one, or (in the
-	 * case of a partitioned table) the parent's, if it has one.
-	 */
-	if (stmt->accessMethod != NULL)
-		accessMethodId = get_table_am_oid(stmt->accessMethod, false);
-	else if (stmt->partbound)
-	{
-		Assert(list_length(inheritOids) == 1);
-		accessMethodId = get_rel_relam(linitial_oid(inheritOids));
-	}
-	else
-		accessMethodId = InvalidOid;
-
-	/* still nothing? use the default */
-	if (RELKIND_HAS_TABLE_AM(relkind) && !OidIsValid(accessMethodId))
-		accessMethodId = get_table_am_oid(default_table_access_method, false);
-
 	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
@@ -15520,7 +15528,8 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			(void) heap_reloptions(rel->rd_rel->relkind, newOptions, true);
+			(void) table_reloptions(rel, rel->rd_rel->relkind,
+									newOptions, true);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			(void) partitioned_table_reloptions(newOptions, true);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 71e8a6f2584..d1d76016ab4 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -2661,7 +2661,9 @@ extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc)
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_MATVIEW ||
 		   ((Form_pg_class) GETSTRUCT(tup))->relkind == RELKIND_TOASTVALUE);
 
-	relopts = extractRelOptions(tup, pg_class_desc, NULL);
+	relopts = extractRelOptions(tup, pg_class_desc,
+								GetTableAmRoutineByAmOid(((Form_pg_class) GETSTRUCT(tup))->relam),
+								NULL);
 	if (relopts == NULL)
 		return NULL;
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 1f419c2a6dd..039c0d3eef4 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -33,6 +33,7 @@
 #include "access/htup_details.h"
 #include "access/multixact.h"
 #include "access/parallel.h"
+#include "access/relation.h"
 #include "access/reloptions.h"
 #include "access/sysattr.h"
 #include "access/table.h"
@@ -464,6 +465,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
 	amoptions_function amoptsfn;
+	const TableAmRoutine *tableam = NULL;
 
 	relation->rd_options = NULL;
 
@@ -478,6 +480,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 		case RELKIND_VIEW:
 		case RELKIND_MATVIEW:
 		case RELKIND_PARTITIONED_TABLE:
+			tableam = relation->rd_tableam;
 			amoptsfn = NULL;
 			break;
 		case RELKIND_INDEX:
@@ -493,7 +496,8 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	 * we might not have any other for pg_class yet (consider executing this
 	 * code for pg_class itself)
 	 */
-	options = extractRelOptions(tuple, GetPgClassDescriptor(), amoptsfn);
+	options = extractRelOptions(tuple, GetPgClassDescriptor(),
+								tableam, amoptsfn);
 
 	/*
 	 * Copy parsed data into CacheMemoryContext.  To guard against the
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index 81829b8270a..8ddc75df287 100644
--- a/src/include/access/reloptions.h
+++ b/src/include/access/reloptions.h
@@ -21,6 +21,7 @@
 
 #include "access/amapi.h"
 #include "access/htup.h"
+#include "access/tableam.h"
 #include "access/tupdesc.h"
 #include "nodes/pg_list.h"
 #include "storage/lock.h"
@@ -224,6 +225,7 @@ extern Datum transformRelOptions(Datum oldOptions, List *defList,
 								 bool acceptOidsOff, bool isReset);
 extern List *untransformRelOptions(Datum options);
 extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
+								const TableAmRoutine *tableam,
 								amoptions_function amoptions);
 extern void *build_reloptions(Datum reloptions, bool validate,
 							  relopt_kind kind,
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index 8ed4e7295ad..c882d7b8ad1 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -737,6 +737,28 @@ typedef struct TableAmRoutine
 											   int32 slicelength,
 											   struct varlena *result);
 
+	/*
+	 * This callback parses and validate sthe reloptions array for a table.
+	 *
+	 * This is called only when a non-null reloptions array exists for the
+	 * table.  'reloptions' is a text array containing entries of the form
+	 * "name=value".  The function should construct a bytea value, which will
+	 * be copied into the rd_options field of the table's relcache entry. The
+	 * data contents of the bytea value are open for the access method to
+	 * define.
+	 *
+	 * When 'validate' is true, the function should report a suitable error
+	 * message if any of the options are unrecognized or have invalid values;
+	 * when 'validate' is false, invalid entries should be silently ignored.
+	 * ('validate' is false when loading options already stored in pg_catalog;
+	 * an invalid entry could only be found if the access method has changed
+	 * its rules for options, and in that case ignoring obsolete entries is
+	 * appropriate.)
+	 *
+	 * It is OK to return NULL if default behavior is wanted.
+	 */
+	bytea	   *(*reloptions) (char relkind, Datum reloptions, bool validate);
+
 
 	/* ------------------------------------------------------------------------
 	 * Planner related functions.
@@ -1925,6 +1947,26 @@ table_relation_fetch_toast_slice(Relation toastrel, Oid valueid,
 													 result);
 }
 
+/*
+ * Parse options for given table.
+ */
+static inline bytea *
+table_reloptions(Relation rel, char relkind,
+				 Datum reloptions, bool validate)
+{
+	return rel->rd_tableam->reloptions(relkind, reloptions, validate);
+}
+
+/*
+ * Parse table options without knowledge of particular table.
+ */
+static inline bytea *
+tableam_reloptions(const TableAmRoutine *tableam, char relkind,
+				   Datum reloptions, bool validate)
+{
+	return tableam->reloptions(relkind, reloptions, validate);
+}
+
 
 /* ----------------------------------------------------------------------------
  * Planner related functionality
@@ -2102,6 +2144,7 @@ extern void table_block_relation_estimate_size(Relation rel,
  */
 
 extern const TableAmRoutine *GetTableAmRoutine(Oid amhandler);
+extern const TableAmRoutine *GetTableAmRoutineByAmOid(Oid amoid);
 
 /* ----------------------------------------------------------------------------
  * Functions in heapam_handler.c
-- 
2.39.3 (Apple Git-145)

