Per-table storage parameters for TableAM/IndexAM extensions

From 6e09f9e5c95f3c88c0240437f99b854a92355e50 Mon Sep 17 00:00:00 2001
From: B Sadhu Prasad Patro <b.sadhuprasadp@enterprisedb.com>
Date: Tue, 18 Jan 2022 08:36:28 -0800
Subject: [PATCH v2] [PATCH v2] Per-table storage parameters for
 TableAM/IndexAM extensions
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Currently all the storage options for a table are very much specific
to the heap but a different AM might need some user defined AM
specific parameters to help tune the AM. So here is a patch which
provides an AM level routine so that instead of getting parameters
validated using “heap_reloptions” it will call the registered AM
routine.
---
 src/backend/access/common/reloptions.c   | 30 +++++++++++++--
 src/backend/access/heap/heapam_handler.c |  1 +
 src/backend/commands/tablecmds.c         | 66 +++++++++++++++++++++++---------
 src/backend/postmaster/autovacuum.c      | 18 ++++++---
 src/backend/utils/cache/relcache.c       | 11 +++++-
 src/include/access/reloptions.h          |  6 ++-
 src/include/access/tableam.h             |  8 +++-
 7 files changed, 108 insertions(+), 32 deletions(-)

diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d592655..bcb08d7 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -1372,7 +1372,8 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				 amoptions_function amoptions,
+				 reloptions_function reloptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1394,7 +1395,9 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			options = heap_reloptions(classForm->relkind, datum, false);
+			options = table_reloptions(reloptions,
+							classForm->relkind,
+							datum, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
@@ -2007,7 +2010,8 @@ view_reloptions(Datum reloptions, bool validate)
 }
 
 /*
- * Parse options for heaps, views and toast tables.
+ * Parse options for heaps, views and toast tables. This is
+ * implementation of relOptions for access method heap.
  */
 bytea *
 heap_reloptions(char relkind, Datum reloptions, bool validate)
@@ -2038,6 +2042,26 @@ heap_reloptions(char relkind, Datum reloptions, bool validate)
 
 
 /*
+ * Parse options for tables.
+ *
+ *	reloptions	tables AM's option parser function
+ *	reloptions	options as text[] datum
+ *	validate	error flag
+ */
+bytea *
+table_reloptions(reloptions_function reloptsfun, char relkind,
+				 Datum reloptions, bool validate)
+{
+	Assert(reloptsfun != NULL);
+
+	/* Assume function is strict */
+	if (!PointerIsValid(DatumGetPointer(reloptions)))
+		return NULL;
+
+	return reloptsfun(relkind, reloptions, validate);
+}
+
+/*
  * Parse options for indexes.
  *
  *	amoptions	index AM's option parser function
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 39ef8a0..4f7f110 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -2581,6 +2581,7 @@ static const TableAmRoutine heapam_methods = {
 	.index_build_range_scan = heapam_index_build_range_scan,
 	.index_validate_scan = heapam_index_validate_scan,
 
+	.relation_options = heap_reloptions,
 	.relation_size = table_block_relation_size,
 	.relation_needs_toast_table = heapam_relation_needs_toast_table,
 	.relation_toast_am = heapam_relation_toast_am,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1f0654c..2d4bb5d 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -808,24 +808,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
 
-	/*
-	 * Parse and validate reloptions, if any.
-	 */
-	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
-									 true, false);
-
-	switch (relkind)
-	{
-		case RELKIND_VIEW:
-			(void) view_reloptions(reloptions, true);
-			break;
-		case RELKIND_PARTITIONED_TABLE:
-			(void) partitioned_table_reloptions(reloptions, true);
-			break;
-		default:
-			(void) heap_reloptions(relkind, reloptions, true);
-	}
-
 	if (stmt->ofTypename)
 	{
 		AclResult	aclresult;
@@ -946,6 +928,52 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		accessMethodId = get_table_am_oid(accessMethod, false);
 
 	/*
+	 * Parse and validate reloptions, if any.
+	 */
+	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
+									 true, false);
+	switch (relkind)
+	{
+		case RELKIND_VIEW:
+			(void) view_reloptions(reloptions, true);
+			break;
+		case RELKIND_PARTITIONED_TABLE:
+			(void) partitioned_table_reloptions(reloptions, true);
+			break;
+		case RELKIND_RELATION:
+		case RELKIND_TOASTVALUE:
+		case RELKIND_MATVIEW:
+			{
+				const TableAmRoutine *routine;
+				HeapTuple	tuple;
+				Form_pg_am	aform;
+
+				tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(accessMethodId));
+				if (!HeapTupleIsValid(tuple))
+				{
+					elog(ERROR, "cache lookup failed for access method %u",
+						 accessMethodId);
+				}
+
+				aform = (Form_pg_am) GETSTRUCT(tuple);
+				routine = GetTableAmRoutine(aform->amhandler);
+				if (routine->relation_options == NULL)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("specifying a table access method is not supported")));
+				}
+
+				(void) routine->relation_options(relkind, reloptions, true);
+				ReleaseSysCache(tuple);
+				break;
+			}
+
+		default:
+			(void) heap_reloptions(relkind, reloptions, true);
+	}
+
+	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
 	 * stored immediately.
@@ -14136,7 +14164,7 @@ 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);
+			rel->rd_tableam->relation_options(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 681ef91..cd9c0bb 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -327,6 +327,7 @@ static void FreeWorkerInfo(int code, Datum arg);
 
 static autovac_table *table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 											TupleDesc pg_class_desc,
+											reloptions_function reloptions,
 											int effective_multixact_freeze_max_age);
 static void recheck_relation_needs_vacanalyze(Oid relid, AutoVacOpts *avopts,
 											  Form_pg_class classForm,
@@ -341,7 +342,7 @@ static void relation_needs_vacanalyze(Oid relid, AutoVacOpts *relopts,
 static void autovacuum_do_vac_analyze(autovac_table *tab,
 									  BufferAccessStrategy bstrategy);
 static AutoVacOpts *extract_autovac_opts(HeapTuple tup,
-										 TupleDesc pg_class_desc);
+										 TupleDesc pg_class_desc, reloptions_function reloptions);
 static PgStat_StatTabEntry *get_pgstat_tabentry_relid(Oid relid, bool isshared,
 													  PgStat_StatDBEntry *shared,
 													  PgStat_StatDBEntry *dbentry);
@@ -2118,7 +2119,8 @@ do_autovacuum(void)
 		}
 
 		/* Fetch reloptions and the pgstat entry for this table */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		tabentry = get_pgstat_tabentry_relid(relid, classForm->relisshared,
 											 shared, dbentry);
 
@@ -2191,7 +2193,8 @@ do_autovacuum(void)
 		 * fetch reloptions -- if this toast table does not have them, try the
 		 * main rel
 		 */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		if (relopts == NULL)
 		{
 			av_relation *hentry;
@@ -2427,6 +2430,7 @@ do_autovacuum(void)
 		 */
 		MemoryContextSwitchTo(AutovacMemCxt);
 		tab = table_recheck_autovac(relid, table_toast_map, pg_class_desc,
+									classRel->rd_tableam->relation_options,
 									effective_multixact_freeze_max_age);
 		if (tab == NULL)
 		{
@@ -2748,7 +2752,8 @@ deleted2:
  * be a risk; fortunately, it doesn't.
  */
 static AutoVacOpts *
-extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc)
+extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc,
+					 reloptions_function reloptions)
 {
 	bytea	   *relopts;
 	AutoVacOpts *av;
@@ -2757,7 +2762,7 @@ 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, NULL, reloptions);
 	if (relopts == NULL)
 		return NULL;
 
@@ -2803,6 +2808,7 @@ get_pgstat_tabentry_relid(Oid relid, bool isshared, PgStat_StatDBEntry *shared,
 static autovac_table *
 table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 					  TupleDesc pg_class_desc,
+					  reloptions_function reloptions,
 					  int effective_multixact_freeze_max_age)
 {
 	Form_pg_class classForm;
@@ -2824,7 +2830,7 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 	 * Get the applicable reloptions.  If it is a TOAST table, try to get the
 	 * main table reloptions if the toast table itself doesn't have.
 	 */
-	avopts = extract_autovac_opts(classTup, pg_class_desc);
+	avopts = extract_autovac_opts(classTup, pg_class_desc, reloptions);
 	if (classForm->relkind == RELKIND_TOASTVALUE &&
 		avopts == NULL && table_toast_map != NULL)
 	{
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 2e760e8..17bdfa3 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -460,6 +460,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
 	amoptions_function amoptsfn;
+	reloptions_function reloptsfn;
 
 	relation->rd_options = NULL;
 
@@ -471,13 +472,18 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
-		case RELKIND_VIEW:
 		case RELKIND_MATVIEW:
+			reloptsfn = relation->rd_tableam->relation_options;
+			amoptsfn = NULL;
+			break;
+		case RELKIND_VIEW:
 		case RELKIND_PARTITIONED_TABLE:
+			reloptsfn = NULL;
 			amoptsfn = NULL;
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
+			reloptsfn = NULL;
 			amoptsfn = relation->rd_indam->amoptions;
 			break;
 		default:
@@ -489,7 +495,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(),
+								amoptsfn, reloptsfn);
 
 	/*
 	 * Copy parsed data into CacheMemoryContext.  To guard against the
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index f740513..c42d5e9 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,7 +225,8 @@ extern Datum transformRelOptions(Datum oldOptions, List *defList,
 								 bool acceptOidsOff, bool isReset);
 extern List *untransformRelOptions(Datum options);
 extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-								amoptions_function amoptions);
+								amoptions_function amoptions,
+								reloptions_function reloptions);
 extern void *build_reloptions(Datum reloptions, bool validate,
 							  relopt_kind kind,
 							  Size relopt_struct_size,
@@ -238,6 +240,8 @@ extern bytea *default_reloptions(Datum reloptions, bool validate,
 extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 extern bytea *view_reloptions(Datum reloptions, bool validate);
 extern bytea *partitioned_table_reloptions(Datum reloptions, bool validate);
+extern bytea *table_reloptions(reloptions_function reloptsfun, char relkind,
+							   Datum reloptions, bool validate);
 extern bytea *index_reloptions(amoptions_function amoptions, Datum reloptions,
 							   bool validate);
 extern bytea *attribute_reloptions(Datum reloptions, bool validate);
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index bb36573..e110085 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -252,6 +252,10 @@ typedef void (*IndexBuildCallback) (Relation index,
 									bool tupleIsAlive,
 									void *state);
 
+/* This callback parse the table reloptions and returns in bytea format */
+typedef bytea *(*reloptions_function) (char relkind,
+									   Datum reloptions, bool validate);
+
 /*
  * API struct for a table AM.  Note this must be allocated in a
  * server-lifetime manner, typically as a static const struct, which then gets
@@ -692,6 +696,8 @@ typedef struct TableAmRoutine
 	 * ------------------------------------------------------------------------
 	 */
 
+	reloptions_function relation_options;
+
 	/*
 	 * See table_relation_size().
 	 *
@@ -702,7 +708,6 @@ typedef struct TableAmRoutine
 	 */
 	uint64		(*relation_size) (Relation rel, ForkNumber forkNumber);
 
-
 	/*
 	 * This callback should return true if the relation requires a TOAST table
 	 * and false if it does not.  It may wish to examine the relation's tuple
@@ -2073,5 +2078,6 @@ extern const TableAmRoutine *GetTableAmRoutine(Oid amhandler);
 extern const TableAmRoutine *GetHeapamTableAmRoutine(void);
 extern bool check_default_table_access_method(char **newval, void **extra,
 											  GucSource source);
+extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 
 #endif							/* TABLEAM_H */
-- 
1.8.3.1

From 7b8b9afcd2c528f7bf826c7596e7e35e7c8f0aa1 Mon Sep 17 00:00:00 2001
From: B Sadhu Prasad Patro <b.sadhuprasadp@enterprisedb.com>
Date: Wed, 23 Feb 2022 21:59:47 -0800
Subject: [PATCH v3] [PATCH V3] Per-table storage parameters for TableAM
 extensions
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Currently all the storage options for a table are very much specific
to the heap but a different AM might need some user defined AM
specific parameters to help tune the AM. So here is a patch which
provides an AM level routine so that instead of getting parameters
validated using “heap_reloptions” it will call the registered AM
routine.

I have added new test module, which shows way to register a new table
access method.
---
 doc/src/sgml/ref/create_table.sgml                 |   3 +-
 src/backend/access/common/reloptions.c             |  30 +-
 src/backend/access/heap/heapam_handler.c           |   1 +
 src/backend/commands/tablecmds.c                   |  66 ++-
 src/backend/postmaster/autovacuum.c                |  18 +-
 src/backend/utils/cache/relcache.c                 |  11 +-
 src/include/access/reloptions.h                    |   6 +-
 src/include/access/tableam.h                       |   8 +-
 src/test/modules/dummy_table_am/Makefile           |  20 +
 src/test/modules/dummy_table_am/README             |  12 +
 .../modules/dummy_table_am/dummy_table_am--1.0.sql |  19 +
 src/test/modules/dummy_table_am/dummy_table_am.c   | 491 +++++++++++++++++++++
 .../modules/dummy_table_am/dummy_table_am.control  |   6 +
 .../modules/dummy_table_am/expected/reloptions.out |  10 +
 src/test/modules/dummy_table_am/sql/reloptions.sql |  14 +
 15 files changed, 682 insertions(+), 33 deletions(-)
 create mode 100644 src/test/modules/dummy_table_am/Makefile
 create mode 100644 src/test/modules/dummy_table_am/README
 create mode 100644 src/test/modules/dummy_table_am/dummy_table_am--1.0.sql
 create mode 100644 src/test/modules/dummy_table_am/dummy_table_am.c
 create mode 100644 src/test/modules/dummy_table_am/dummy_table_am.control
 create mode 100644 src/test/modules/dummy_table_am/expected/reloptions.out
 create mode 100644 src/test/modules/dummy_table_am/sql/reloptions.sql

diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index 7e4ef31..615bcad 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -1375,7 +1375,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     Storage parameters for
     indexes are documented in <xref linkend="sql-createindex"/>.
     The storage parameters currently
-    available for tables are listed below.  For many of these parameters, as
+    available for tables are listed below. Each table may have different set of storage
+    parameters through different access methods. For many of these parameters, as
     shown, there is an additional parameter with the same name prefixed with
     <literal>toast.</literal>, which controls the behavior of the
     table's secondary <acronym>TOAST</acronym> table, if any
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d592655..bcb08d7 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -1372,7 +1372,8 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				 amoptions_function amoptions,
+				 reloptions_function reloptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1394,7 +1395,9 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			options = heap_reloptions(classForm->relkind, datum, false);
+			options = table_reloptions(reloptions,
+							classForm->relkind,
+							datum, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
@@ -2007,7 +2010,8 @@ view_reloptions(Datum reloptions, bool validate)
 }
 
 /*
- * Parse options for heaps, views and toast tables.
+ * Parse options for heaps, views and toast tables. This is
+ * implementation of relOptions for access method heap.
  */
 bytea *
 heap_reloptions(char relkind, Datum reloptions, bool validate)
@@ -2038,6 +2042,26 @@ heap_reloptions(char relkind, Datum reloptions, bool validate)
 
 
 /*
+ * Parse options for tables.
+ *
+ *	reloptions	tables AM's option parser function
+ *	reloptions	options as text[] datum
+ *	validate	error flag
+ */
+bytea *
+table_reloptions(reloptions_function reloptsfun, char relkind,
+				 Datum reloptions, bool validate)
+{
+	Assert(reloptsfun != NULL);
+
+	/* Assume function is strict */
+	if (!PointerIsValid(DatumGetPointer(reloptions)))
+		return NULL;
+
+	return reloptsfun(relkind, reloptions, validate);
+}
+
+/*
  * Parse options for indexes.
  *
  *	amoptions	index AM's option parser function
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 39ef8a0..4f7f110 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -2581,6 +2581,7 @@ static const TableAmRoutine heapam_methods = {
 	.index_build_range_scan = heapam_index_build_range_scan,
 	.index_validate_scan = heapam_index_validate_scan,
 
+	.relation_options = heap_reloptions,
 	.relation_size = table_block_relation_size,
 	.relation_needs_toast_table = heapam_relation_needs_toast_table,
 	.relation_toast_am = heapam_relation_toast_am,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 3e83f37..1d6168f 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -809,24 +809,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
 
-	/*
-	 * Parse and validate reloptions, if any.
-	 */
-	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
-									 true, false);
-
-	switch (relkind)
-	{
-		case RELKIND_VIEW:
-			(void) view_reloptions(reloptions, true);
-			break;
-		case RELKIND_PARTITIONED_TABLE:
-			(void) partitioned_table_reloptions(reloptions, true);
-			break;
-		default:
-			(void) heap_reloptions(relkind, reloptions, true);
-	}
-
 	if (stmt->ofTypename)
 	{
 		AclResult	aclresult;
@@ -947,6 +929,52 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		accessMethodId = get_table_am_oid(accessMethod, false);
 
 	/*
+	 * Parse and validate reloptions, if any.
+	 */
+	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
+									 true, false);
+	switch (relkind)
+	{
+		case RELKIND_VIEW:
+			(void) view_reloptions(reloptions, true);
+			break;
+		case RELKIND_PARTITIONED_TABLE:
+			(void) partitioned_table_reloptions(reloptions, true);
+			break;
+		case RELKIND_RELATION:
+		case RELKIND_TOASTVALUE:
+		case RELKIND_MATVIEW:
+			{
+				const TableAmRoutine *routine;
+				HeapTuple	tuple;
+				Form_pg_am	aform;
+
+				tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(accessMethodId));
+				if (!HeapTupleIsValid(tuple))
+				{
+					elog(ERROR, "cache lookup failed for access method %u",
+						 accessMethodId);
+				}
+
+				aform = (Form_pg_am) GETSTRUCT(tuple);
+				routine = GetTableAmRoutine(aform->amhandler);
+				if (routine->relation_options == NULL)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("specifying a table access method is not supported")));
+				}
+
+				(void) routine->relation_options(relkind, reloptions, true);
+				ReleaseSysCache(tuple);
+				break;
+			}
+
+		default:
+			(void) heap_reloptions(relkind, reloptions, true);
+	}
+
+	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
 	 * stored immediately.
@@ -14137,7 +14165,7 @@ 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);
+			rel->rd_tableam->relation_options(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 681ef91..cd9c0bb 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -327,6 +327,7 @@ static void FreeWorkerInfo(int code, Datum arg);
 
 static autovac_table *table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 											TupleDesc pg_class_desc,
+											reloptions_function reloptions,
 											int effective_multixact_freeze_max_age);
 static void recheck_relation_needs_vacanalyze(Oid relid, AutoVacOpts *avopts,
 											  Form_pg_class classForm,
@@ -341,7 +342,7 @@ static void relation_needs_vacanalyze(Oid relid, AutoVacOpts *relopts,
 static void autovacuum_do_vac_analyze(autovac_table *tab,
 									  BufferAccessStrategy bstrategy);
 static AutoVacOpts *extract_autovac_opts(HeapTuple tup,
-										 TupleDesc pg_class_desc);
+										 TupleDesc pg_class_desc, reloptions_function reloptions);
 static PgStat_StatTabEntry *get_pgstat_tabentry_relid(Oid relid, bool isshared,
 													  PgStat_StatDBEntry *shared,
 													  PgStat_StatDBEntry *dbentry);
@@ -2118,7 +2119,8 @@ do_autovacuum(void)
 		}
 
 		/* Fetch reloptions and the pgstat entry for this table */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		tabentry = get_pgstat_tabentry_relid(relid, classForm->relisshared,
 											 shared, dbentry);
 
@@ -2191,7 +2193,8 @@ do_autovacuum(void)
 		 * fetch reloptions -- if this toast table does not have them, try the
 		 * main rel
 		 */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		if (relopts == NULL)
 		{
 			av_relation *hentry;
@@ -2427,6 +2430,7 @@ do_autovacuum(void)
 		 */
 		MemoryContextSwitchTo(AutovacMemCxt);
 		tab = table_recheck_autovac(relid, table_toast_map, pg_class_desc,
+									classRel->rd_tableam->relation_options,
 									effective_multixact_freeze_max_age);
 		if (tab == NULL)
 		{
@@ -2748,7 +2752,8 @@ deleted2:
  * be a risk; fortunately, it doesn't.
  */
 static AutoVacOpts *
-extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc)
+extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc,
+					 reloptions_function reloptions)
 {
 	bytea	   *relopts;
 	AutoVacOpts *av;
@@ -2757,7 +2762,7 @@ 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, NULL, reloptions);
 	if (relopts == NULL)
 		return NULL;
 
@@ -2803,6 +2808,7 @@ get_pgstat_tabentry_relid(Oid relid, bool isshared, PgStat_StatDBEntry *shared,
 static autovac_table *
 table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 					  TupleDesc pg_class_desc,
+					  reloptions_function reloptions,
 					  int effective_multixact_freeze_max_age)
 {
 	Form_pg_class classForm;
@@ -2824,7 +2830,7 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 	 * Get the applicable reloptions.  If it is a TOAST table, try to get the
 	 * main table reloptions if the toast table itself doesn't have.
 	 */
-	avopts = extract_autovac_opts(classTup, pg_class_desc);
+	avopts = extract_autovac_opts(classTup, pg_class_desc, reloptions);
 	if (classForm->relkind == RELKIND_TOASTVALUE &&
 		avopts == NULL && table_toast_map != NULL)
 	{
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index fccffce..91ec1b7 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -461,6 +461,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
 	amoptions_function amoptsfn;
+	reloptions_function reloptsfn;
 
 	relation->rd_options = NULL;
 
@@ -472,13 +473,18 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
-		case RELKIND_VIEW:
 		case RELKIND_MATVIEW:
+			reloptsfn = relation->rd_tableam->relation_options;
+			amoptsfn = NULL;
+			break;
+		case RELKIND_VIEW:
 		case RELKIND_PARTITIONED_TABLE:
+			reloptsfn = NULL;
 			amoptsfn = NULL;
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
+			reloptsfn = NULL;
 			amoptsfn = relation->rd_indam->amoptions;
 			break;
 		default:
@@ -490,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(),
+								amoptsfn, reloptsfn);
 
 	/*
 	 * Copy parsed data into CacheMemoryContext.  To guard against the
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index f740513..c42d5e9 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,7 +225,8 @@ extern Datum transformRelOptions(Datum oldOptions, List *defList,
 								 bool acceptOidsOff, bool isReset);
 extern List *untransformRelOptions(Datum options);
 extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-								amoptions_function amoptions);
+								amoptions_function amoptions,
+								reloptions_function reloptions);
 extern void *build_reloptions(Datum reloptions, bool validate,
 							  relopt_kind kind,
 							  Size relopt_struct_size,
@@ -238,6 +240,8 @@ extern bytea *default_reloptions(Datum reloptions, bool validate,
 extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 extern bytea *view_reloptions(Datum reloptions, bool validate);
 extern bytea *partitioned_table_reloptions(Datum reloptions, bool validate);
+extern bytea *table_reloptions(reloptions_function reloptsfun, char relkind,
+							   Datum reloptions, bool validate);
 extern bytea *index_reloptions(amoptions_function amoptions, Datum reloptions,
 							   bool validate);
 extern bytea *attribute_reloptions(Datum reloptions, bool validate);
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index bb36573..e110085 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -252,6 +252,10 @@ typedef void (*IndexBuildCallback) (Relation index,
 									bool tupleIsAlive,
 									void *state);
 
+/* This callback parse the table reloptions and returns in bytea format */
+typedef bytea *(*reloptions_function) (char relkind,
+									   Datum reloptions, bool validate);
+
 /*
  * API struct for a table AM.  Note this must be allocated in a
  * server-lifetime manner, typically as a static const struct, which then gets
@@ -692,6 +696,8 @@ typedef struct TableAmRoutine
 	 * ------------------------------------------------------------------------
 	 */
 
+	reloptions_function relation_options;
+
 	/*
 	 * See table_relation_size().
 	 *
@@ -702,7 +708,6 @@ typedef struct TableAmRoutine
 	 */
 	uint64		(*relation_size) (Relation rel, ForkNumber forkNumber);
 
-
 	/*
 	 * This callback should return true if the relation requires a TOAST table
 	 * and false if it does not.  It may wish to examine the relation's tuple
@@ -2073,5 +2078,6 @@ extern const TableAmRoutine *GetTableAmRoutine(Oid amhandler);
 extern const TableAmRoutine *GetHeapamTableAmRoutine(void);
 extern bool check_default_table_access_method(char **newval, void **extra,
 											  GucSource source);
+extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 
 #endif							/* TABLEAM_H */
diff --git a/src/test/modules/dummy_table_am/Makefile b/src/test/modules/dummy_table_am/Makefile
new file mode 100644
index 0000000..94837df
--- /dev/null
+++ b/src/test/modules/dummy_table_am/Makefile
@@ -0,0 +1,20 @@
+# src/test/modules/dummy_table_am/Makefile
+
+MODULES = dummy_table_am
+
+EXTENSION = dummy_table_am
+DATA = dummy_table_am--1.0.sql
+PGFILEDESC = "dummy_table_am - table access method template"
+
+REGRESS = reloptions
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/dummy_table_am
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/dummy_table_am/README b/src/test/modules/dummy_table_am/README
new file mode 100644
index 0000000..c1a9642
--- /dev/null
+++ b/src/test/modules/dummy_table_am/README
@@ -0,0 +1,12 @@
+Dummy Table AM
+==============
+
+Dummy table AM is a module for testing any facility usable by an table
+access method, whose code is kept a maximum simple.
+
+This includes tests for all relation option types:
+- boolean
+- enum
+- integer
+- real
+- strings (with and without NULL as default)
diff --git a/src/test/modules/dummy_table_am/dummy_table_am--1.0.sql b/src/test/modules/dummy_table_am/dummy_table_am--1.0.sql
new file mode 100644
index 0000000..aca35b7
--- /dev/null
+++ b/src/test/modules/dummy_table_am/dummy_table_am--1.0.sql
@@ -0,0 +1,19 @@
+/* src/test/modules/dummy_table_am/dummy_table_am--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION dummy_table_am" to load this file. \quit
+
+CREATE FUNCTION dthandler(internal)
+RETURNS table_am_handler
+AS 'MODULE_PATHNAME'
+LANGUAGE C;
+
+-- Access method
+CREATE ACCESS METHOD dummy_table_am TYPE TABLE HANDLER dthandler;
+COMMENT ON ACCESS METHOD dummy_table_am IS 'dummy table access method';
+
+-- Operator classes
+--CREATE OPERATOR CLASS int4_ops
+--DEFAULT FOR TYPE int4 USING dummy_table_am AS
+--  OPERATOR 1 = (int4, int4),
+-- FUNCTION 1 hashint4(int4);
diff --git a/src/test/modules/dummy_table_am/dummy_table_am.c b/src/test/modules/dummy_table_am/dummy_table_am.c
new file mode 100644
index 0000000..fb3c0ae
--- /dev/null
+++ b/src/test/modules/dummy_table_am/dummy_table_am.c
@@ -0,0 +1,491 @@
+/*-------------------------------------------------------------------------
+ *
+ * dummy_table_am.c
+ *              Table AM template main file.
+ *
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *        src/test/modules/dummy_tablen_am/dummy_table_am.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/amapi.h"
+#include "access/reloptions.h"
+#include "catalog/index.h"
+#include "commands/vacuum.h"
+#include "nodes/pathnodes.h"
+#include "utils/guc.h"
+#include "utils/rel.h"
+
+PG_MODULE_MAGIC;
+
+void            _PG_init(void);
+
+/* parse table for fillRelOptions */
+relopt_parse_elt di_relopt_tab[2];
+
+/* Dummy table options */
+typedef struct DummyTableOptions
+{
+	StdRdOptions	stdOptions;
+        bool		isCompression;
+}DummyTableOptions;
+
+typedef struct BulkInsertStateData *BulkInsertState;
+
+/* Handler for table AM */
+PG_FUNCTION_INFO_V1(dthandler);
+
+/*
+ * This function creates a full set of relation option types,
+ * with various patterns.
+ */
+static bytea *
+create_reloptions_table(char relkind, Datum reloptions, bool validate)
+{
+	DummyTableOptions* options = (DummyTableOptions *)DatumGetPointer(reloptions);
+	if (options->isCompression)
+	{
+		printf("COMPRESSION ENABLED");
+	}
+	else
+	{
+		printf("COMPRESSION DISABLED");
+	}
+
+	return NULL;
+}
+
+static bool
+tblam_scan_bitmap_next_block(TableScanDesc scan,
+							  TBMIterateResult *tbmres)
+{
+	return false;
+}
+
+static bool
+tblam_scan_bitmap_next_tuple(TableScanDesc scan,
+							  TBMIterateResult *tbmres,
+							  TupleTableSlot *slot)
+{
+	return false;
+}
+
+static bool
+tblam_scan_sample_next_block(TableScanDesc scan, SampleScanState *scanstate)
+{
+	return false;
+}
+
+static bool
+tblam_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate,
+							  TupleTableSlot *slot)
+{
+	return false;
+}
+
+static void
+tblam_estimate_rel_size(Relation rel, int32 *attr_widths,
+						 BlockNumber *pages, double *tuples,
+						 double *allvisfrac)
+{
+	*allvisfrac = 10 * 8192 /* assuming BLCKSZ */;
+	return;
+}
+
+static void
+tblam_fetch_toast_slice(Relation toastrel, Oid valueid, int32 attrsize,
+                                           int32 sliceoffset, int32 slicelength,
+                                           struct varlena *result)
+{
+	return;
+}
+
+static Oid
+tblam_relation_toast_am(Relation rel)
+{
+	return rel->rd_rel->relam;
+}
+
+static bool
+tblam_relation_needs_toast_table(Relation rel)
+{
+	return false;
+}
+
+static uint64
+tblam_block_relation_size(Relation rel, ForkNumber forkNumber)
+{
+	uint64		nblocks = 10;
+	return nblocks * 8192 /* assuming BLCKSZ*/;
+}
+
+static void
+tblam_index_validate_scan(Relation heapRelation,
+						   Relation indexRelation,
+						   IndexInfo *indexInfo,
+						   Snapshot snapshot,
+						   ValidateIndexState *state)
+{
+	return;
+}
+
+static double
+tblam_index_build_range_scan(Relation heapRelation,
+							  Relation indexRelation,
+							  IndexInfo *indexInfo,
+							  bool allow_sync,
+							  bool anyvisible,
+							  bool progress,
+							  BlockNumber start_blockno,
+							  BlockNumber numblocks,
+							  IndexBuildCallback callback,
+							  void *callback_state,
+							  TableScanDesc scan)
+{
+	return 0;
+}
+
+static bool
+tblam_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
+							   double *liverows, double *deadrows,
+							   TupleTableSlot *slot)
+{
+	return false;
+}
+
+static void
+tblam_vacuum_rel(Relation rel, VacuumParams *params,
+                                BufferAccessStrategy bstrategy)
+{
+	return;
+}
+
+static bool
+tblam_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
+							   BufferAccessStrategy bstrategy)
+{
+	return false;
+}
+
+static void
+tblam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
+								 Relation OldIndex, bool use_sort,
+								 TransactionId OldestXmin,
+								 TransactionId *xid_cutoff,
+								 MultiXactId *multi_cutoff,
+								 double *num_tuples,
+								 double *tups_vacuumed,
+								 double *tups_recently_dead)
+{
+	return;
+}
+
+static void
+tblam_relation_copy_data(Relation rel, const RelFileNode *newrnode)
+{
+	return;
+}
+
+static void
+tblam_relation_nontransactional_truncate(Relation rel)
+{
+	return;
+}
+
+static void
+tblam_relation_set_new_filenode(Relation rel,
+								 const RelFileNode *newrnode,
+								 char persistence,
+								 TransactionId *freezeXid,
+								 MultiXactId *minmulti)
+{
+	return;
+}
+
+static TransactionId
+tblam_index_delete_tuples(Relation rel, TM_IndexDeleteOp *delstate)
+{
+	return InvalidTransactionId;
+}
+
+static bool
+tblam_tuple_satisfies_snapshot(Relation rel, TupleTableSlot *slot,
+								Snapshot snapshot)
+{
+	return false;
+}
+
+static bool
+tblam_tuple_tid_valid(TableScanDesc scan, ItemPointer tid)
+{
+	return false;
+}
+
+static void
+tblam_get_latest_tid(TableScanDesc sscan,
+					ItemPointer tid)
+{
+	return;
+}
+
+static bool
+tblam_fetch_row_version(Relation relation,
+						 ItemPointer tid,
+						 Snapshot snapshot,
+						 TupleTableSlot *slot)
+{
+	return false;
+}
+
+static TM_Result
+tblam_tuple_lock(Relation relation, ItemPointer tid, Snapshot snapshot,
+				  TupleTableSlot *slot, CommandId cid, LockTupleMode mode,
+				  LockWaitPolicy wait_policy, uint8 flags,
+				  TM_FailureData *tmfd)
+{
+	return TM_Ok;
+}
+
+static TM_Result
+tblam_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot,
+					CommandId cid, Snapshot snapshot, Snapshot crosscheck,
+					bool wait, TM_FailureData *tmfd,
+					LockTupleMode *lockmode, bool *update_indexes)
+{
+	return TM_Ok;
+}
+
+static TM_Result
+tblam_tuple_delete(Relation relation, ItemPointer tid, CommandId cid,
+					Snapshot snapshot, Snapshot crosscheck, bool wait,
+					TM_FailureData *tmfd, bool changingPart)
+{
+	return TM_Ok;
+}
+
+static void
+tblam_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
+				  CommandId cid, int options, BulkInsertState bistate)
+{
+	return;
+}
+
+static void
+tblam_tuple_complete_speculative(Relation relation, TupleTableSlot *slot,
+								  uint32 specToken, bool succeeded)
+{
+	return;
+}
+
+static void
+tblam_tuple_insert_speculative(Relation relation, TupleTableSlot *slot,
+								CommandId cid, int options,
+								BulkInsertState bistate, uint32 specToken)
+{
+	return;
+}
+
+static void
+tblam_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
+					int options, BulkInsertState bistate)
+{
+	return;
+}
+
+static IndexFetchTableData *
+tblam_index_fetch_begin(Relation rel)
+{
+	return NULL;
+}
+
+static void
+tblam_index_fetch_reset(IndexFetchTableData *scan)
+{
+	return;
+}
+
+static void
+tblam_index_fetch_end(IndexFetchTableData *scan)
+{
+	return;
+}
+
+static bool
+tblam_index_fetch_tuple(struct IndexFetchTableData *scan,
+						 ItemPointer tid,
+						 Snapshot snapshot,
+						 TupleTableSlot *slot,
+						 bool *call_again, bool *all_dead)
+{
+	return false;
+}
+
+static Size
+tblam_block_parallelscan_estimate(Relation rel)
+{
+	return sizeof(ParallelBlockTableScanDescData);
+}
+
+static Size
+tblam_block_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan)
+{
+	ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
+
+	bpscan->base.phs_relid = RelationGetRelid(rel);
+	bpscan->phs_nblocks = 10; //RelationGetNumberOfBlocks(rel);
+	/* compare phs_syncscan initialization to similar logic in initscan */
+	bpscan->base.phs_syncscan = synchronize_seqscans &&
+		!RelationUsesLocalBuffers(rel) &&
+		bpscan->phs_nblocks > 10 / 4;
+	SpinLockInit(&bpscan->phs_mutex);
+	bpscan->phs_startblock = InvalidBlockNumber;
+	pg_atomic_init_u64(&bpscan->phs_nallocated, 0);
+
+	return sizeof(ParallelBlockTableScanDescData);
+}
+
+static void
+tblam_block_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
+{
+	ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
+
+	pg_atomic_write_u64(&bpscan->phs_nallocated, 0);
+}
+
+static bool
+tblam_getnextslot_tidrange(TableScanDesc sscan, ScanDirection direction,
+						  TupleTableSlot *slot)
+{
+	return false;
+}
+
+static void
+tblam_set_tidrange(TableScanDesc sscan, ItemPointer mintid,
+				  ItemPointer maxtid)
+{
+	return;
+}
+
+static bool
+tblam_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
+{
+	return false;
+}
+
+static void
+tblam_rescan(TableScanDesc sscan, ScanKey key, bool set_params,
+			bool allow_strat, bool allow_sync, bool allow_pagemode)
+{
+	return;
+}
+
+static void
+tblam_endscan(TableScanDesc sscan)
+{
+	return;
+}
+
+static TableScanDesc
+tblam_beginscan(Relation relation, Snapshot snapshot,
+			   int nkeys, ScanKey key,
+			   ParallelTableScanDesc parallel_scan,
+			   uint32 flags)
+{
+	return NULL;
+}
+
+static const TupleTableSlotOps *
+tblam_slot_callbacks(Relation relation)
+{
+	return &TTSOpsBufferHeapTuple;
+}
+
+/*
+ * Table AM handler function: returns TableAmRoutine with access method
+ * parameters and callbacks.
+ */
+Datum
+dthandler(PG_FUNCTION_ARGS)
+{
+	TableAmRoutine *amroutine = makeNode(TableAmRoutine);
+
+	amroutine->type = T_TableAmRoutine;
+
+	amroutine->slot_callbacks = tblam_slot_callbacks;
+
+	amroutine->scan_begin = tblam_beginscan;
+	amroutine->scan_end = tblam_endscan;
+	amroutine->scan_rescan = tblam_rescan;
+	amroutine->scan_getnextslot = tblam_getnextslot;
+
+	amroutine->scan_set_tidrange = tblam_set_tidrange;
+	amroutine->scan_getnextslot_tidrange = tblam_getnextslot_tidrange;
+
+	amroutine->parallelscan_estimate = tblam_block_parallelscan_estimate;
+	amroutine->parallelscan_initialize = tblam_block_parallelscan_initialize;
+	amroutine->parallelscan_reinitialize = tblam_block_parallelscan_reinitialize;
+
+	amroutine->index_fetch_begin = tblam_index_fetch_begin;
+	amroutine->index_fetch_reset = tblam_index_fetch_reset;
+	amroutine->index_fetch_end = tblam_index_fetch_end;
+	amroutine->index_fetch_tuple = tblam_index_fetch_tuple;
+
+	amroutine->tuple_insert = tblam_tuple_insert;
+	amroutine->tuple_insert_speculative = tblam_tuple_insert_speculative;
+	amroutine->tuple_complete_speculative = tblam_tuple_complete_speculative;
+	amroutine->multi_insert = tblam_multi_insert;
+	amroutine->tuple_delete = tblam_tuple_delete;
+	amroutine->tuple_update = tblam_tuple_update;
+	amroutine->tuple_lock = tblam_tuple_lock;
+
+	amroutine->tuple_fetch_row_version = tblam_fetch_row_version;
+	amroutine->tuple_get_latest_tid = tblam_get_latest_tid;
+	amroutine->tuple_tid_valid = tblam_tuple_tid_valid;
+	amroutine->tuple_satisfies_snapshot = tblam_tuple_satisfies_snapshot;
+	amroutine->index_delete_tuples = tblam_index_delete_tuples;
+
+	amroutine->relation_set_new_filenode = tblam_relation_set_new_filenode;
+	amroutine->relation_nontransactional_truncate = tblam_relation_nontransactional_truncate;
+	amroutine->relation_copy_data = tblam_relation_copy_data;
+	amroutine->relation_copy_for_cluster = tblam_relation_copy_for_cluster;
+	amroutine->relation_vacuum = tblam_vacuum_rel;
+	amroutine->scan_analyze_next_block = tblam_scan_analyze_next_block;
+	amroutine->scan_analyze_next_tuple = tblam_scan_analyze_next_tuple;
+	amroutine->index_build_range_scan = tblam_index_build_range_scan;
+	amroutine->index_validate_scan = tblam_index_validate_scan;
+
+	amroutine->relation_options = create_reloptions_table;
+	amroutine->relation_size = tblam_block_relation_size;
+	amroutine->relation_needs_toast_table = tblam_relation_needs_toast_table;
+	amroutine->relation_toast_am = tblam_relation_toast_am;
+	amroutine->relation_fetch_toast_slice = tblam_fetch_toast_slice;
+
+	amroutine->relation_estimate_size = tblam_estimate_rel_size;
+
+	amroutine->scan_bitmap_next_block = tblam_scan_bitmap_next_block;
+	amroutine->scan_bitmap_next_tuple = tblam_scan_bitmap_next_tuple;
+	amroutine->scan_sample_next_block = tblam_scan_sample_next_block;
+	amroutine->scan_sample_next_tuple = tblam_scan_sample_next_tuple;
+
+	PG_RETURN_POINTER(amroutine);
+}
+
+void
+_PG_init(void)
+{
+	DummyTableOptions tabAmStr = {{0}};
+
+	tabAmStr.isCompression = false;
+        create_reloptions_table(RELKIND_RELATION, (Datum)&tabAmStr, false);
+
+	{
+		Relation rel;
+		tblam_relation_set_new_filenode(rel, NULL, 'p', NULL, NULL);
+	}
+}
+
diff --git a/src/test/modules/dummy_table_am/dummy_table_am.control b/src/test/modules/dummy_table_am/dummy_table_am.control
new file mode 100644
index 0000000..a81d361
--- /dev/null
+++ b/src/test/modules/dummy_table_am/dummy_table_am.control
@@ -0,0 +1,6 @@
+# dummy_table_am extension
+comment = 'dummy_table_am - table access method template'
+default_version = '1.0'
+module_pathname = '$libdir/dummy_table_am'
+relocatable = true
+
diff --git a/src/test/modules/dummy_table_am/expected/reloptions.out b/src/test/modules/dummy_table_am/expected/reloptions.out
new file mode 100644
index 0000000..2e75881
--- /dev/null
+++ b/src/test/modules/dummy_table_am/expected/reloptions.out
@@ -0,0 +1,10 @@
+-- Tests for relation options
+CREATE EXTENSION dummy_table_am;
+CREATE TABLE dummy_test_tab (i int4);
+-- Silence validation checks for strings
+SET client_min_messages TO 'warning';
+-- Test for new AM for dummy_table_am
+CREATE TABLE mytest (a int) USING dummy_table_am WITH (is_compression='TRUE');
+ALTER TABLE mytest SET (isCompression = 'FALSE');
+insert into mytest values(10);
+DROP TABLE mytest;
diff --git a/src/test/modules/dummy_table_am/sql/reloptions.sql b/src/test/modules/dummy_table_am/sql/reloptions.sql
new file mode 100644
index 0000000..7eb0372
--- /dev/null
+++ b/src/test/modules/dummy_table_am/sql/reloptions.sql
@@ -0,0 +1,14 @@
+-- Tests for relation options
+CREATE EXTENSION dummy_table_am;
+CREATE TABLE dummy_test_tab (i int4);
+
+-- Silence validation checks for strings
+SET client_min_messages TO 'warning';
+
+-- Test for new AM for dummy_table_am
+CREATE TABLE mytest (a int) USING dummy_table_am WITH (is_compression='TRUE');
+ALTER TABLE mytest SET (isCompression = 'FALSE');
+
+insert into mytest values(10);
+
+DROP TABLE mytest;
-- 
1.8.3.1

From 782e7d98320c6e7d9bf5c0f9aa33fa1c6047101f Mon Sep 17 00:00:00 2001
From: B Sadhu Prasad Patro <b.sadhuprasadp@enterprisedb.com>
Date: Mon, 21 Mar 2022 20:23:22 -0700
Subject: [PATCH v4] [PATCH V4] Per-table storage parameters for TableAM
 extensions
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Currently all the storage options for a table are very much specific
to the heap but a different AM might need some user defined AM
specific parameters to help tune the AM. So here is a patch which
provides an AM level routine so that instead of getting parameters
validated using “heap_reloptions” it will call the registered AM
routine.

I have added new test module, which shows way to register a new table
access method.

New Command supported as:
ALTER TABLE name SET ACCESS METHOD amname [ OPTIONS ( ADD | DROP option 'value' [, ... ] ) ];
---
 doc/src/sgml/ref/create_table.sgml       |   3 +-
 src/backend/access/common/reloptions.c   |  30 ++++-
 src/backend/access/heap/heapam_handler.c |   1 +
 src/backend/commands/tablecmds.c         | 214 ++++++++++++++++++++++++++++---
 src/backend/parser/gram.y                |  43 ++++++-
 src/backend/postmaster/autovacuum.c      |  18 ++-
 src/backend/utils/cache/relcache.c       |  11 +-
 src/include/access/reloptions.h          |   6 +-
 src/include/access/tableam.h             |   8 +-
 9 files changed, 298 insertions(+), 36 deletions(-)

diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index 7e4ef31..615bcad 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -1375,7 +1375,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     Storage parameters for
     indexes are documented in <xref linkend="sql-createindex"/>.
     The storage parameters currently
-    available for tables are listed below.  For many of these parameters, as
+    available for tables are listed below. Each table may have different set of storage
+    parameters through different access methods. For many of these parameters, as
     shown, there is an additional parameter with the same name prefixed with
     <literal>toast.</literal>, which controls the behavior of the
     table's secondary <acronym>TOAST</acronym> table, if any
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index d592655..bcb08d7 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -1372,7 +1372,8 @@ untransformRelOptions(Datum options)
  */
 bytea *
 extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-				  amoptions_function amoptions)
+				 amoptions_function amoptions,
+				 reloptions_function reloptions)
 {
 	bytea	   *options;
 	bool		isnull;
@@ -1394,7 +1395,9 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_MATVIEW:
-			options = heap_reloptions(classForm->relkind, datum, false);
+			options = table_reloptions(reloptions,
+							classForm->relkind,
+							datum, false);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			options = partitioned_table_reloptions(datum, false);
@@ -2007,7 +2010,8 @@ view_reloptions(Datum reloptions, bool validate)
 }
 
 /*
- * Parse options for heaps, views and toast tables.
+ * Parse options for heaps, views and toast tables. This is
+ * implementation of relOptions for access method heap.
  */
 bytea *
 heap_reloptions(char relkind, Datum reloptions, bool validate)
@@ -2038,6 +2042,26 @@ heap_reloptions(char relkind, Datum reloptions, bool validate)
 
 
 /*
+ * Parse options for tables.
+ *
+ *	reloptions	tables AM's option parser function
+ *	reloptions	options as text[] datum
+ *	validate	error flag
+ */
+bytea *
+table_reloptions(reloptions_function reloptsfun, char relkind,
+				 Datum reloptions, bool validate)
+{
+	Assert(reloptsfun != NULL);
+
+	/* Assume function is strict */
+	if (!PointerIsValid(DatumGetPointer(reloptions)))
+		return NULL;
+
+	return reloptsfun(relkind, reloptions, validate);
+}
+
+/*
  * Parse options for indexes.
  *
  *	amoptions	index AM's option parser function
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 39ef8a0..4f7f110 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -2581,6 +2581,7 @@ static const TableAmRoutine heapam_methods = {
 	.index_build_range_scan = heapam_index_build_range_scan,
 	.index_validate_scan = heapam_index_validate_scan,
 
+	.relation_options = heap_reloptions,
 	.relation_size = table_block_relation_size,
 	.relation_needs_toast_table = heapam_relation_needs_toast_table,
 	.relation_toast_am = heapam_relation_toast_am,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 80faae9..e2d5172 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -574,6 +574,7 @@ static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
 								const char *tablespacename, LOCKMODE lockmode);
 static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
 static void ATExecSetTableSpaceNoStorage(Relation rel, Oid newTableSpace);
+static void ATExecSetAMOptions(Relation rel, List *defList, AlterTableType operation, LOCKMODE lockmode);
 static void ATExecSetRelOptions(Relation rel, List *defList,
 								AlterTableType operation,
 								LOCKMODE lockmode);
@@ -816,24 +817,6 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	if (!OidIsValid(ownerId))
 		ownerId = GetUserId();
 
-	/*
-	 * Parse and validate reloptions, if any.
-	 */
-	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
-									 true, false);
-
-	switch (relkind)
-	{
-		case RELKIND_VIEW:
-			(void) view_reloptions(reloptions, true);
-			break;
-		case RELKIND_PARTITIONED_TABLE:
-			(void) partitioned_table_reloptions(reloptions, true);
-			break;
-		default:
-			(void) heap_reloptions(relkind, reloptions, true);
-	}
-
 	if (stmt->ofTypename)
 	{
 		AclResult	aclresult;
@@ -954,6 +937,52 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		accessMethodId = get_table_am_oid(accessMethod, false);
 
 	/*
+	 * Parse and validate reloptions, if any.
+	 */
+	reloptions = transformRelOptions((Datum) 0, stmt->options, NULL, validnsps,
+									 true, false);
+	switch (relkind)
+	{
+		case RELKIND_VIEW:
+			(void) view_reloptions(reloptions, true);
+			break;
+		case RELKIND_PARTITIONED_TABLE:
+			(void) partitioned_table_reloptions(reloptions, true);
+			break;
+		case RELKIND_RELATION:
+		case RELKIND_TOASTVALUE:
+		case RELKIND_MATVIEW:
+			{
+				const TableAmRoutine *routine;
+				HeapTuple	tuple;
+				Form_pg_am	aform;
+
+				tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(accessMethodId));
+				if (!HeapTupleIsValid(tuple))
+				{
+					elog(ERROR, "cache lookup failed for access method %u",
+						 accessMethodId);
+				}
+
+				aform = (Form_pg_am) GETSTRUCT(tuple);
+				routine = GetTableAmRoutine(aform->amhandler);
+				if (routine->relation_options == NULL)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("specifying a table access method is not supported")));
+				}
+
+				(void) routine->relation_options(relkind, reloptions, true);
+				ReleaseSysCache(tuple);
+				break;
+			}
+
+		default:
+			(void) heap_reloptions(relkind, reloptions, true);
+	}
+
+	/*
 	 * Create the relation.  Inherited defaults and constraints are passed in
 	 * for immediate handling --- since they don't need parsing, they can be
 	 * stored immediately.
@@ -5087,6 +5116,7 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab,
 			break;
 		case AT_SetAccessMethod:	/* SET ACCESS METHOD */
 			/* handled specially in Phase 3 */
+			ATExecSetAMOptions(rel, (List *) cmd->def, cmd->subtype, lockmode);
 			break;
 		case AT_SetTableSpace:	/* SET TABLESPACE */
 
@@ -14103,6 +14133,152 @@ ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, const char *tablespacen
 	tab->newTableSpace = tablespaceId;
 }
 
+/*TODO: This needs to be removed. Same function is already present in foreigncmds.c*/
+static Datum
+optionListToArray(List *options)
+{
+	ArrayBuildState *astate = NULL;
+	ListCell   *cell;
+
+	foreach(cell, options)
+	{
+		DefElem    *def = lfirst(cell);
+		const char *value;
+		Size		len;
+		text	   *t;
+
+		value = defGetString(def);
+		len = VARHDRSZ + strlen(def->defname) + 1 + strlen(value);
+		t = palloc(len + 1);
+		SET_VARSIZE(t, len);
+		sprintf(VARDATA(t), "%s=%s", def->defname, value);
+
+		astate = accumArrayResult(astate, PointerGetDatum(t),
+								  false, TEXTOID,
+								  CurrentMemoryContext);
+	}
+
+	if (astate)
+		return makeArrayResult(astate, CurrentMemoryContext);
+
+	return PointerGetDatum(NULL);
+}
+
+/* ADD or DROP reloptions in SET ACCESS METHOD.*/
+static void
+ATExecSetAMOptions(Relation rel, List *defList, AlterTableType operation,
+					LOCKMODE lockmode)
+{
+	Oid			relid;
+	Relation	pgclass;
+	HeapTuple	tuple;
+	HeapTuple	newtuple;
+	Datum		datum;
+	bool		isnull;
+	Datum		newOptions;
+	Datum		repl_val[Natts_pg_class];
+	bool		repl_null[Natts_pg_class];
+	bool		repl_repl[Natts_pg_class];
+	List		*resultOptions;
+	ListCell	*optcell;
+
+	if (defList == NIL)
+		return; /* nothing to do */
+
+	pgclass = table_open(RelationRelationId, RowExclusiveLock);
+
+	/* Fetch heap tuple */
+	relid = RelationGetRelid(rel);
+	tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for relation %u", relid);
+
+	/* Get the old reloptions */
+	datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
+								&isnull);
+	resultOptions = untransformRelOptions(datum);
+
+	foreach(optcell, defList)
+	{
+		DefElem    *od = lfirst(optcell);
+		ListCell   *cell;
+
+		/* Search in existing options */
+		foreach(cell, resultOptions)
+		{
+			DefElem    *def = lfirst(cell);
+
+			if (strcmp(def->defname, od->defname) == 0)
+				break;
+		}
+
+		/*
+		 * It is possible to perform multiple ADD/DROP actions on the same
+		 * option.  The standard permits this, as long as the options to be
+		 * added are unique.
+		 */
+		switch (od->defaction)
+		{
+			case DEFELEM_ADD:
+				if (cell)
+					ereport(ERROR,
+							(errcode(ERRCODE_DUPLICATE_OBJECT),
+							 errmsg("option \"%s\" provided more than once",
+									od->defname)));
+				resultOptions = lappend(resultOptions, od);
+				break;
+
+			case DEFELEM_DROP:
+				if (!cell)
+					ereport(ERROR,
+							(errcode(ERRCODE_UNDEFINED_OBJECT),
+							 errmsg("option \"%s\" not found",
+									od->defname)));
+				resultOptions = list_delete_cell(resultOptions, cell);
+				break;
+
+			default:
+				elog(ERROR, "unrecognized action %d on option \"%s\"",
+					 (int) od->defaction, od->defname);
+				break;
+		}
+	}
+
+	newOptions = optionListToArray(resultOptions);
+
+	/* DO need to call the handler function to validate ?*/
+
+	/*
+	 * All we need do here is update the pg_class row; the new options will be
+	 * propagated into relcaches during post-commit cache inval.
+	 */
+	memset(repl_val, 0, sizeof(repl_val));
+	memset(repl_null, false, sizeof(repl_null));
+	memset(repl_repl, false, sizeof(repl_repl));
+
+	if (newOptions != (Datum) 0)
+		repl_val[Anum_pg_class_reloptions - 1] = newOptions;
+	else
+		repl_null[Anum_pg_class_reloptions - 1] = true;
+
+	repl_repl[Anum_pg_class_reloptions - 1] = true;
+
+	newtuple = heap_modify_tuple(tuple, RelationGetDescr(pgclass),
+								 repl_val, repl_null, repl_repl);
+
+	CatalogTupleUpdate(pgclass, &newtuple->t_self, newtuple);
+
+
+	InvokeObjectPostAlterHook(RelationRelationId, RelationGetRelid(rel),
+								  InvalidOid);
+
+	heap_freetuple(newtuple);
+
+	ReleaseSysCache(tuple);
+
+	table_close(pgclass, RowExclusiveLock);
+}
+
 /*
  * Set, reset, or replace reloptions.
  */
@@ -14160,7 +14336,7 @@ 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);
+			rel->rd_tableam->relation_options(rel->rd_rel->relkind, newOptions, true);
 			break;
 		case RELKIND_PARTITIONED_TABLE:
 			(void) partitioned_table_reloptions(newOptions, true);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 0036c2f..2b1d870 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -441,7 +441,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				prep_type_clause
 				execute_param_clause using_clause returning_clause
 				opt_enum_val_list enum_val_list table_func_column_list
-				create_generic_options alter_generic_options
+				create_generic_options alter_generic_options alter_am_options
 				relation_expr_list dostmt_opt_list
 				transform_element_list transform_type_list
 				TriggerTransitions TriggerReferencing
@@ -544,8 +544,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 %type <str>		generic_option_name
 %type <node>	generic_option_arg
-%type <defelt>	generic_option_elem alter_generic_option_elem
-%type <list>	generic_option_list alter_generic_option_list
+%type <defelt>	generic_option_elem alter_generic_option_elem alter_am_option_elem
+%type <list>	generic_option_list alter_generic_option_list alter_am_option_list
 
 %type <ival>	reindex_target_type reindex_target_multitable
 
@@ -2655,6 +2655,14 @@ alter_table_cmd:
 					n->name = $4;
 					$$ = (Node *)n;
 				}
+			/* ALTER TABLE <name> SET ACCESS METHOD <amname> [OPTIONS]*/
+			| SET ACCESS METHOD name alter_am_options
+				{
+					AlterTableCmd *n = makeNode(AlterTableCmd);
+					n->subtype = AT_SetAccessMethod;
+					n->name = $4;
+					$$ = (Node *)n;
+				}
 			/* ALTER TABLE <name> SET TABLESPACE <tablespacename> */
 			| SET TABLESPACE name
 				{
@@ -2724,6 +2732,35 @@ alter_table_cmd:
 				}
 		;
 
+/* Options definition for ALTER TABLE <name> SET ACCESS METHOD <amname> OPTIONS (...) */
+alter_am_options:
+			OPTIONS '(' alter_am_option_list ')'               { $$ = $3; }
+		;
+
+alter_am_option_list:
+			alter_am_option_elem
+				{
+					$$ = list_make1($1);
+				}
+			| alter_am_option_list ',' alter_am_option_elem
+				{
+					$$ = lappend($1, $3);
+				}
+		;
+
+alter_am_option_elem:
+			ADD_P generic_option_elem
+				{
+					$$ = $2;
+					$$->defaction = DEFELEM_ADD;
+				}
+			| DROP generic_option_elem
+				{
+					$$ = makeDefElemExtended(NULL, $2, NULL, DEFELEM_DROP, @2);
+				}
+		;
+
+
 alter_column_default:
 			SET DEFAULT a_expr			{ $$ = $3; }
 			| DROP DEFAULT				{ $$ = NULL; }
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 681ef91..cd9c0bb 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -327,6 +327,7 @@ static void FreeWorkerInfo(int code, Datum arg);
 
 static autovac_table *table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 											TupleDesc pg_class_desc,
+											reloptions_function reloptions,
 											int effective_multixact_freeze_max_age);
 static void recheck_relation_needs_vacanalyze(Oid relid, AutoVacOpts *avopts,
 											  Form_pg_class classForm,
@@ -341,7 +342,7 @@ static void relation_needs_vacanalyze(Oid relid, AutoVacOpts *relopts,
 static void autovacuum_do_vac_analyze(autovac_table *tab,
 									  BufferAccessStrategy bstrategy);
 static AutoVacOpts *extract_autovac_opts(HeapTuple tup,
-										 TupleDesc pg_class_desc);
+										 TupleDesc pg_class_desc, reloptions_function reloptions);
 static PgStat_StatTabEntry *get_pgstat_tabentry_relid(Oid relid, bool isshared,
 													  PgStat_StatDBEntry *shared,
 													  PgStat_StatDBEntry *dbentry);
@@ -2118,7 +2119,8 @@ do_autovacuum(void)
 		}
 
 		/* Fetch reloptions and the pgstat entry for this table */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		tabentry = get_pgstat_tabentry_relid(relid, classForm->relisshared,
 											 shared, dbentry);
 
@@ -2191,7 +2193,8 @@ do_autovacuum(void)
 		 * fetch reloptions -- if this toast table does not have them, try the
 		 * main rel
 		 */
-		relopts = extract_autovac_opts(tuple, pg_class_desc);
+		relopts = extract_autovac_opts(tuple, pg_class_desc,
+									   classRel->rd_tableam->relation_options);
 		if (relopts == NULL)
 		{
 			av_relation *hentry;
@@ -2427,6 +2430,7 @@ do_autovacuum(void)
 		 */
 		MemoryContextSwitchTo(AutovacMemCxt);
 		tab = table_recheck_autovac(relid, table_toast_map, pg_class_desc,
+									classRel->rd_tableam->relation_options,
 									effective_multixact_freeze_max_age);
 		if (tab == NULL)
 		{
@@ -2748,7 +2752,8 @@ deleted2:
  * be a risk; fortunately, it doesn't.
  */
 static AutoVacOpts *
-extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc)
+extract_autovac_opts(HeapTuple tup, TupleDesc pg_class_desc,
+					 reloptions_function reloptions)
 {
 	bytea	   *relopts;
 	AutoVacOpts *av;
@@ -2757,7 +2762,7 @@ 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, NULL, reloptions);
 	if (relopts == NULL)
 		return NULL;
 
@@ -2803,6 +2808,7 @@ get_pgstat_tabentry_relid(Oid relid, bool isshared, PgStat_StatDBEntry *shared,
 static autovac_table *
 table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 					  TupleDesc pg_class_desc,
+					  reloptions_function reloptions,
 					  int effective_multixact_freeze_max_age)
 {
 	Form_pg_class classForm;
@@ -2824,7 +2830,7 @@ table_recheck_autovac(Oid relid, HTAB *table_toast_map,
 	 * Get the applicable reloptions.  If it is a TOAST table, try to get the
 	 * main table reloptions if the toast table itself doesn't have.
 	 */
-	avopts = extract_autovac_opts(classTup, pg_class_desc);
+	avopts = extract_autovac_opts(classTup, pg_class_desc, reloptions);
 	if (classForm->relkind == RELKIND_TOASTVALUE &&
 		avopts == NULL && table_toast_map != NULL)
 	{
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index fccffce..91ec1b7 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -461,6 +461,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 {
 	bytea	   *options;
 	amoptions_function amoptsfn;
+	reloptions_function reloptsfn;
 
 	relation->rd_options = NULL;
 
@@ -472,13 +473,18 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
-		case RELKIND_VIEW:
 		case RELKIND_MATVIEW:
+			reloptsfn = relation->rd_tableam->relation_options;
+			amoptsfn = NULL;
+			break;
+		case RELKIND_VIEW:
 		case RELKIND_PARTITIONED_TABLE:
+			reloptsfn = NULL;
 			amoptsfn = NULL;
 			break;
 		case RELKIND_INDEX:
 		case RELKIND_PARTITIONED_INDEX:
+			reloptsfn = NULL;
 			amoptsfn = relation->rd_indam->amoptions;
 			break;
 		default:
@@ -490,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(),
+								amoptsfn, reloptsfn);
 
 	/*
 	 * Copy parsed data into CacheMemoryContext.  To guard against the
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index f740513..c42d5e9 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,7 +225,8 @@ extern Datum transformRelOptions(Datum oldOptions, List *defList,
 								 bool acceptOidsOff, bool isReset);
 extern List *untransformRelOptions(Datum options);
 extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
-								amoptions_function amoptions);
+								amoptions_function amoptions,
+								reloptions_function reloptions);
 extern void *build_reloptions(Datum reloptions, bool validate,
 							  relopt_kind kind,
 							  Size relopt_struct_size,
@@ -238,6 +240,8 @@ extern bytea *default_reloptions(Datum reloptions, bool validate,
 extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 extern bytea *view_reloptions(Datum reloptions, bool validate);
 extern bytea *partitioned_table_reloptions(Datum reloptions, bool validate);
+extern bytea *table_reloptions(reloptions_function reloptsfun, char relkind,
+							   Datum reloptions, bool validate);
 extern bytea *index_reloptions(amoptions_function amoptions, Datum reloptions,
 							   bool validate);
 extern bytea *attribute_reloptions(Datum reloptions, bool validate);
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index bb36573..e110085 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -252,6 +252,10 @@ typedef void (*IndexBuildCallback) (Relation index,
 									bool tupleIsAlive,
 									void *state);
 
+/* This callback parse the table reloptions and returns in bytea format */
+typedef bytea *(*reloptions_function) (char relkind,
+									   Datum reloptions, bool validate);
+
 /*
  * API struct for a table AM.  Note this must be allocated in a
  * server-lifetime manner, typically as a static const struct, which then gets
@@ -692,6 +696,8 @@ typedef struct TableAmRoutine
 	 * ------------------------------------------------------------------------
 	 */
 
+	reloptions_function relation_options;
+
 	/*
 	 * See table_relation_size().
 	 *
@@ -702,7 +708,6 @@ typedef struct TableAmRoutine
 	 */
 	uint64		(*relation_size) (Relation rel, ForkNumber forkNumber);
 
-
 	/*
 	 * This callback should return true if the relation requires a TOAST table
 	 * and false if it does not.  It may wish to examine the relation's tuple
@@ -2073,5 +2078,6 @@ extern const TableAmRoutine *GetTableAmRoutine(Oid amhandler);
 extern const TableAmRoutine *GetHeapamTableAmRoutine(void);
 extern bool check_default_table_access_method(char **newval, void **extra,
 											  GucSource source);
+extern bytea *heap_reloptions(char relkind, Datum reloptions, bool validate);
 
 #endif							/* TABLEAM_H */
-- 
1.8.3.1