0009-Custom-reloptions-for-table-AM-v1.patch

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

Let table AM define custom reloptions for its tables.  Also, let table AM
override reloptions for indexes built on its tables.
---
 src/backend/access/common/reloptions.c   |  9 ++--
 src/backend/access/heap/heapam_handler.c | 21 +++++++++
 src/backend/access/table/tableamapi.c    | 18 +++++++
 src/backend/commands/indexcmds.c         |  3 +-
 src/backend/commands/tablecmds.c         | 60 +++++++++++++++---------
 src/backend/postmaster/autovacuum.c      |  4 +-
 src/backend/utils/cache/relcache.c       | 30 ++++++++++--
 src/include/access/reloptions.h          |  2 +
 src/include/access/tableam.h             | 52 ++++++++++++++++++++
 9 files changed, 169 insertions(+), 30 deletions(-)

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index c32bb7d2b64..2f1ebaacd47 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"
@@ -1379,7 +1380,7 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				  const TableAmRoutine *tableam, amoptions_function amoptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1401,7 +1402,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);
@@ -1411,7 +1413,8 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
-			options = index_reloptions(amoptions, datum, false);
+			options = tableam_indexoptions(tableam, amoptions, classForm->relkind,
+										   datum, false);
 			break;
 		case RELKIND_FOREIGN_TABLE:
 			options = NULL;
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 5f25244ce43..d3b8edc73ee 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"
@@ -2719,6 +2720,24 @@ heapam_relation_toast_am(Relation rel)
 	return rel->rd_rel->relam;
 }
 
+static bytea *
+heapam_reloptions(char relkind, Datum reloptions, bool validate)
+{
+	if (relkind == RELKIND_RELATION ||
+		relkind == RELKIND_TOASTVALUE ||
+		relkind == RELKIND_MATVIEW)
+		return heap_reloptions(relkind, reloptions, validate);
+
+	return NULL;
+}
+
+static bytea *
+heapam_indexoptions(amoptions_function amoptions, char relkind,
+					Datum reloptions, bool validate)
+{
+	return index_reloptions(amoptions, reloptions, validate);
+}
+
 
 /* ------------------------------------------------------------------------
  * Planner related callbacks for the heap AM
@@ -3224,6 +3243,8 @@ 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,
+	.indexoptions = heapam_indexoptions,
 
 	.relation_estimate_size = heapam_estimate_rel_size,
 
diff --git a/src/backend/access/table/tableamapi.c b/src/backend/access/table/tableamapi.c
index b233f585258..1bffedfc10a 100644
--- a/src/backend/access/table/tableamapi.c
+++ b/src/backend/access/table/tableamapi.c
@@ -106,6 +106,24 @@ GetTableAmRoutine(Oid amhandler)
 	return routine;
 }
 
+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/indexcmds.c b/src/backend/commands/indexcmds.c
index 0b3b8e98b80..43a914f44b1 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -884,7 +884,8 @@ DefineIndex(Oid tableId,
 	reloptions = transformRelOptions((Datum) 0, stmt->options,
 									 NULL, NULL, false, false);
 
-	(void) index_reloptions(amoptions, reloptions, true);
+	(void) tableam_indexoptions(rel->rd_tableam, amoptions, RELKIND_INDEX,
+								reloptions, true);
 
 	/*
 	 * Prepare arguments for index_create, primarily an IndexInfo structure.
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index d3ee23d08ba..0d8c3bb25d8 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -696,6 +696,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	LOCKMODE	parentLockmode;
 	const char *accessMethod = NULL;
 	Oid			accessMethodId = InvalidOid;
+	const TableAmRoutine *tableam = NULL;
 
 	/*
 	 * Truncate relname to appropriate length (probably a waste of time, as
@@ -835,6 +836,26 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
 
+	/*
+	 * If the statement hasn't specified an access method, but we're defining
+	 * a type of relation that needs one, use the default.
+	 */
+	if (stmt->accessMethod != NULL)
+	{
+		accessMethod = stmt->accessMethod;
+
+		if (partitioned)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("specifying a table access method is not supported on a partitioned table")));
+	}
+	else if (RELKIND_HAS_TABLE_AM(relkind))
+		accessMethod = default_table_access_method;
+
+	/* look up the access method, verify it is for a table */
+	if (accessMethod != NULL)
+		accessMethodId = get_table_am_oid(accessMethod, false);
+
 	/*
 	 * Parse and validate reloptions, if any.
 	 */
@@ -843,6 +864,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;
@@ -851,6 +878,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			break;
 		default:
 			(void) heap_reloptions(relkind, reloptions, true);
+			break;
 	}
 
 	if (stmt->ofTypename)
@@ -942,26 +970,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		}
 	}
 
-	/*
-	 * If the statement hasn't specified an access method, but we're defining
-	 * a type of relation that needs one, use the default.
-	 */
-	if (stmt->accessMethod != NULL)
-	{
-		accessMethod = stmt->accessMethod;
-
-		if (partitioned)
-			ereport(ERROR,
-					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("specifying a table access method is not supported on a partitioned table")));
-	}
-	else if (RELKIND_HAS_TABLE_AM(relkind))
-		accessMethod = default_table_access_method;
-
-	/* look up the access method, verify it is for a table */
-	if (accessMethod != NULL)
-		accessMethodId = get_table_am_oid(accessMethod, false);
-
 	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
@@ -14989,7 +14997,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);
@@ -14999,7 +15008,14 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
-			(void) index_reloptions(rel->rd_indam->amoptions, newOptions, true);
+			{
+				Relation	tbl = relation_open(rel->rd_index->indrelid,
+												AccessShareLock);
+
+				tableam_indexoptions(tbl->rd_tableam, rel->rd_indam->amoptions,
+									 rel->rd_rel->relkind, newOptions, true);
+				relation_close(tbl, AccessShareLock);
+			}
 			break;
 		default:
 			ereport(ERROR,
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 86a3b3d8be2..efb7cff985f 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -2814,7 +2814,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 c802f33ac59..bc04d68478d 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -34,6 +34,7 @@
 #include "access/multixact.h"
 #include "access/nbtree.h"
 #include "access/parallel.h"
+#include "access/relation.h"
 #include "access/reloptions.h"
 #include "access/sysattr.h"
 #include "access/table.h"
@@ -466,6 +467,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
 	amoptions_function amoptsfn;
+	const TableAmRoutine *tableam = NULL;
 
 	relation->rd_options = NULL;
 
@@ -477,14 +479,35 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
-		case RELKIND_VIEW:
 		case RELKIND_MATVIEW:
+		case RELKIND_VIEW:
 		case RELKIND_PARTITIONED_TABLE:
+			tableam = relation->rd_tableam;
 			amoptsfn = NULL;
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
-			amoptsfn = relation->rd_indam->amoptions;
+			{
+				Form_pg_class classForm;
+				HeapTuple	classTup;
+
+				/* fetch the relation's relcache entry */
+				if (relation->rd_index->indrelid >= FirstNormalObjectId)
+				{
+					classTup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relation->rd_index->indrelid));
+					classForm = (Form_pg_class) GETSTRUCT(classTup);
+					if (classForm->relam >= FirstNormalObjectId)
+						tableam = GetTableAmRoutineByAmOid(classForm->relam);
+					else
+						tableam = GetHeapamTableAmRoutine();
+					heap_freetuple(classTup);
+				}
+				else
+				{
+					tableam = GetHeapamTableAmRoutine();
+				}
+				amoptsfn = relation->rd_indam->amoptions;
+			}
 			break;
 		default:
 			return;
@@ -495,7 +518,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 3602397cf51..2e08edbc54c 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 58ca3ecaedb..32afea79c86 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -17,6 +17,7 @@
 #ifndef TABLEAM_H
 #define TABLEAM_H
 
+#include "access/amapi.h"
 #include "access/relscan.h"
 #include "access/sdir.h"
 #include "access/xact.h"
@@ -740,6 +741,18 @@ typedef struct TableAmRoutine
 											   int32 slicelength,
 											   struct varlena *result);
 
+	/*
+	 * Parse table AM-specific table options.
+	 */
+	bytea	   *(*reloptions) (char relkind, Datum reloptions, bool validate);
+
+	/*
+	 * Parse table AM-specific index options.  Useful for table AM to define
+	 * new index options or override existing index options.
+	 */
+	bytea	   *(*indexoptions) (amoptions_function amoptions, char relkind,
+								 Datum reloptions, bool validate);
+
 
 	/* ------------------------------------------------------------------------
 	 * Planner related functions.
@@ -1956,6 +1969,44 @@ 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);
+}
+
+extern bytea *index_reloptions(amoptions_function amoptions, Datum reloptions,
+							   bool validate);
+
+/*
+ * Parse index options.  Gives table AM a chance to override index-specific
+ * options defined in 'amoptions'.
+ */
+static inline bytea *
+tableam_indexoptions(const TableAmRoutine *tableam,
+					 amoptions_function amoptions, char relkind,
+					 Datum reloptions, bool validate)
+{
+	if (tableam)
+		return tableam->indexoptions(amoptions, relkind, reloptions, validate);
+	else
+		return index_reloptions(amoptions, reloptions, validate);
+}
+
 
 /* ----------------------------------------------------------------------------
  * Planner related functionality
@@ -2134,6 +2185,7 @@ extern void table_block_relation_estimate_size(Relation rel,
  */
 
 extern const TableAmRoutine *GetTableAmRoutine(Oid amhandler);
+extern const TableAmRoutine *GetTableAmRoutineByAmOid(Oid amoid);
 extern const TableAmRoutine *GetHeapamTableAmRoutine(void);
 
 #endif							/* TABLEAM_H */
-- 
2.39.3 (Apple Git-145)