v1-0002-Prototype-Allow-tablespaces-to-specify-which-SMGR.patch

From 8db3e73a6fe60c114335a47432a80ecb447b9357 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Fri, 30 Jun 2023 14:15:36 +0200
Subject: [PATCH v1 2/2] Prototype: Allow tablespaces to specify which SMGR
 they use

This allows for tablespaces that are not present on the local file system.

For now, the default tablespaces (pg_default and pg_global) are still
dependent on the md.c smgr, but in the future this may change as well.
---
 src/backend/access/rmgrdesc/tblspcdesc.c |   2 +-
 src/backend/commands/tablespace.c        | 182 +++----------------
 src/backend/parser/gram.y                |  32 +++-
 src/backend/storage/smgr/md.c            | 214 ++++++++++++++++++++++-
 src/backend/storage/smgr/smgr.c          |  72 +++++++-
 src/backend/utils/cache/spccache.c       |  38 +++-
 src/include/catalog/pg_tablespace.dat    |   6 +-
 src/include/catalog/pg_tablespace.h      |   1 +
 src/include/commands/tablespace.h        |   3 +-
 src/include/nodes/parsenodes.h           |   3 +-
 src/include/storage/md.h                 |  10 ++
 src/include/storage/smgr.h               |  20 ++-
 src/include/utils/spccache.h             |   2 +
 13 files changed, 407 insertions(+), 178 deletions(-)

diff --git a/src/backend/access/rmgrdesc/tblspcdesc.c b/src/backend/access/rmgrdesc/tblspcdesc.c
index b8c89f8c54..04cc15e121 100644
--- a/src/backend/access/rmgrdesc/tblspcdesc.c
+++ b/src/backend/access/rmgrdesc/tblspcdesc.c
@@ -27,7 +27,7 @@ tblspc_desc(StringInfo buf, XLogReaderState *record)
 	{
 		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) rec;
 
-		appendStringInfo(buf, "%u \"%s\"", xlrec->ts_id, xlrec->ts_path);
+		appendStringInfo(buf, "%u \"%s\"", xlrec->ts_id, NameStr(xlrec->ts_smgr));
 	}
 	else if (info == XLOG_TBLSPC_DROP)
 	{
diff --git a/src/backend/commands/tablespace.c b/src/backend/commands/tablespace.c
index 13b0dee146..b3da4a1b93 100644
--- a/src/backend/commands/tablespace.c
+++ b/src/backend/commands/tablespace.c
@@ -74,6 +74,7 @@
 #include "miscadmin.h"
 #include "postmaster/bgwriter.h"
 #include "storage/fd.h"
+#include "storage/md.h"
 #include "storage/lmgr.h"
 #include "storage/standby.h"
 #include "utils/acl.h"
@@ -92,8 +93,6 @@ bool		allow_in_place_tablespaces = false;
 
 Oid			binary_upgrade_next_pg_tablespace_oid = InvalidOid;
 
-static void create_tablespace_directories(const char *location,
-										  const Oid tablespaceoid);
 static bool destroy_tablespace_directories(Oid tablespaceoid, bool redo);
 
 
@@ -218,10 +217,8 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	bool		nulls[Natts_pg_tablespace] = {0};
 	HeapTuple	tuple;
 	Oid			tablespaceoid;
-	char	   *location;
 	Oid			ownerId;
 	Datum		newOptions;
-	bool		in_place;
 
 	/* Must be superuser */
 	if (!superuser())
@@ -237,47 +234,7 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	else
 		ownerId = GetUserId();
 
-	/* Unix-ify the offered path, and strip any trailing slashes */
-	location = pstrdup(stmt->location);
-	canonicalize_path(location);
-
-	/* disallow quotes, else CREATE DATABASE would be at risk */
-	if (strchr(location, '\''))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_NAME),
-				 errmsg("tablespace location cannot contain single quotes")));
-
-	in_place = allow_in_place_tablespaces && strlen(location) == 0;
-
-	/*
-	 * Allowing relative paths seems risky
-	 *
-	 * This also helps us ensure that location is not empty or whitespace,
-	 * unless specifying a developer-only in-place tablespace.
-	 */
-	if (!in_place && !is_absolute_path(location))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-				 errmsg("tablespace location must be an absolute path")));
-
-	/*
-	 * Check that location isn't too long. Remember that we're going to append
-	 * 'PG_XXX/<dboid>/<relid>_<fork>.<nnn>'.  FYI, we never actually
-	 * reference the whole path here, but MakePGDirectory() uses the first two
-	 * parts.
-	 */
-	if (strlen(location) + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) + 1 +
-		OIDCHARS + 1 + OIDCHARS + 1 + FORKNAMECHARS + 1 + OIDCHARS > MAXPGPATH)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-				 errmsg("tablespace location \"%s\" is too long",
-						location)));
-
-	/* Warn if the tablespace is in the data directory. */
-	if (path_is_prefix_of_path(DataDir, location))
-		ereport(WARNING,
-				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
-				 errmsg("tablespace location should not be inside the data directory")));
+	smgrvalidatetspopts(stmt->smgr, stmt->smgropts);
 
 	/*
 	 * Disallow creation of tablespaces named "pg_xxx"; we reserve this
@@ -334,6 +291,8 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	values[Anum_pg_tablespace_oid - 1] = ObjectIdGetDatum(tablespaceoid);
 	values[Anum_pg_tablespace_spcname - 1] =
 		DirectFunctionCall1(namein, CStringGetDatum(stmt->tablespacename));
+	values[Anum_pg_tablespace_spcsmgr - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(stmt->smgr));
 	values[Anum_pg_tablespace_spcowner - 1] =
 		ObjectIdGetDatum(ownerId);
 	nulls[Anum_pg_tablespace_spcacl - 1] = true;
@@ -360,18 +319,22 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	/* Post creation hook for new tablespace */
 	InvokeObjectPostCreateHook(TableSpaceRelationId, tablespaceoid, 0);
 
-	create_tablespace_directories(location, tablespaceoid);
+	smgrcreatetsp(stmt->smgr, tablespaceoid, stmt->smgropts, 0);
 
 	/* Record the filesystem change in XLOG */
 	{
-		xl_tblspc_create_rec xlrec;
+		xl_tblspc_create_rec xlrec = {0};
+		Datum	smgropts;
 
 		xlrec.ts_id = tablespaceoid;
+		memcpy(&xlrec.ts_smgr, stmt->smgr, strlen(stmt->smgr));
+		smgropts = transformRelOptions((Datum) 0, stmt->smgropts,
+									   NULL, NULL, false, false);
 
 		XLogBeginInsert();
 		XLogRegisterData((char *) &xlrec,
-						 offsetof(xl_tblspc_create_rec, ts_path));
-		XLogRegisterData((char *) location, strlen(location) + 1);
+						 offsetof(xl_tblspc_create_rec, ts_smgropts));
+		XLogRegisterData((char *) smgropts, VARSIZE_ANY(smgropts));
 
 		(void) XLogInsert(RM_TBLSPC_ID, XLOG_TBLSPC_CREATE);
 	}
@@ -384,8 +347,6 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	 */
 	ForceSyncCommit();
 
-	pfree(location);
-
 	/* We keep the lock on pg_tablespace until commit */
 	table_close(rel, NoLock);
 
@@ -401,6 +362,7 @@ void
 DropTableSpace(DropTableSpaceStmt *stmt)
 {
 	char	   *tablespacename = stmt->tablespacename;
+	char	   *smgrname;
 	TableScanDesc scandesc;
 	Relation	rel;
 	HeapTuple	tuple;
@@ -444,6 +406,7 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 
 	spcform = (Form_pg_tablespace) GETSTRUCT(tuple);
 	tablespaceoid = spcform->oid;
+	smgrname = pstrdup(NameStr(spcform->spcsmgr));
 
 	/* Must be tablespace owner */
 	if (!object_ownercheck(TableSpaceRelationId, tablespaceoid, GetUserId()))
@@ -492,6 +455,8 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	 */
 	LWLockAcquire(TablespaceCreateLock, LW_EXCLUSIVE);
 
+	smgrdroptsp(smgrname, tablespaceoid, false);
+
 	/*
 	 * Try to remove the physical infrastructure.
 	 */
@@ -567,114 +532,6 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	table_close(rel, NoLock);
 }
 
-
-/*
- * create_tablespace_directories
- *
- *	Attempt to create filesystem infrastructure linking $PGDATA/pg_tblspc/
- *	to the specified directory
- */
-static void
-create_tablespace_directories(const char *location, const Oid tablespaceoid)
-{
-	char	   *linkloc;
-	char	   *location_with_version_dir;
-	struct stat st;
-	bool		in_place;
-
-	linkloc = psprintf("pg_tblspc/%u", tablespaceoid);
-
-	/*
-	 * If we're asked to make an 'in place' tablespace, create the directory
-	 * directly where the symlink would normally go.  This is a developer-only
-	 * option for now, to facilitate regression testing.
-	 */
-	in_place = strlen(location) == 0;
-
-	if (in_place)
-	{
-		if (MakePGDirectory(linkloc) < 0 && errno != EEXIST)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not create directory \"%s\": %m",
-							linkloc)));
-	}
-
-	location_with_version_dir = psprintf("%s/%s", in_place ? linkloc : location,
-										 TABLESPACE_VERSION_DIRECTORY);
-
-	/*
-	 * Attempt to coerce target directory to safe permissions.  If this fails,
-	 * it doesn't exist or has the wrong owner.  Not needed for in-place mode,
-	 * because in that case we created the directory with the desired
-	 * permissions.
-	 */
-	if (!in_place && chmod(location, pg_dir_create_mode) != 0)
-	{
-		if (errno == ENOENT)
-			ereport(ERROR,
-					(errcode(ERRCODE_UNDEFINED_FILE),
-					 errmsg("directory \"%s\" does not exist", location),
-					 InRecovery ? errhint("Create this directory for the tablespace before "
-										  "restarting the server.") : 0));
-		else
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not set permissions on directory \"%s\": %m",
-							location)));
-	}
-
-	/*
-	 * The creation of the version directory prevents more than one tablespace
-	 * in a single location.  This imitates TablespaceCreateDbspace(), but it
-	 * ignores concurrency and missing parent directories.  The chmod() would
-	 * have failed in the absence of a parent.  pg_tablespace_spcname_index
-	 * prevents concurrency.
-	 */
-	if (stat(location_with_version_dir, &st) < 0)
-	{
-		if (errno != ENOENT)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not stat directory \"%s\": %m",
-							location_with_version_dir)));
-		else if (MakePGDirectory(location_with_version_dir) < 0)
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not create directory \"%s\": %m",
-							location_with_version_dir)));
-	}
-	else if (!S_ISDIR(st.st_mode))
-		ereport(ERROR,
-				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-				 errmsg("\"%s\" exists but is not a directory",
-						location_with_version_dir)));
-	else if (!InRecovery)
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_IN_USE),
-				 errmsg("directory \"%s\" already in use as a tablespace",
-						location_with_version_dir)));
-
-	/*
-	 * In recovery, remove old symlink, in case it points to the wrong place.
-	 */
-	if (!in_place && InRecovery)
-		remove_tablespace_symlink(linkloc);
-
-	/*
-	 * Create the symlink under PGDATA
-	 */
-	if (!in_place && symlink(location, linkloc) < 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not create symbolic link \"%s\": %m",
-						linkloc)));
-
-	pfree(linkloc);
-	pfree(location_with_version_dir);
-}
-
-
 /*
  * destroy_tablespace_directories
  *
@@ -1524,9 +1381,12 @@ tblspc_redo(XLogReaderState *record)
 	if (info == XLOG_TBLSPC_CREATE)
 	{
 		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) XLogRecGetData(record);
-		char	   *location = xlrec->ts_path;
+		smgrcreatetsp(NameStr(xlrec->ts_smgr), xlrec->ts_id,
+					  untransformRelOptions((Datum) &xlrec->ts_smgropts), true);
 
-		create_tablespace_directories(location, xlrec->ts_id);
+		/*
+		 * create_tablespace_directories(location, xlrec->ts_id);
+		 */
 	}
 	else if (info == XLOG_TBLSPC_DROP)
 	{
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 39ab7eac0d..49742553d4 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -60,6 +60,7 @@
 #include "nodes/nodeFuncs.h"
 #include "parser/parser.h"
 #include "storage/lmgr.h"
+#include "storage/md.h"
 #include "utils/date.h"
 #include "utils/datetime.h"
 #include "utils/numeric.h"
@@ -394,6 +395,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				opt_inline_handler opt_validator validator_clause
 				opt_collate
 
+%type <node>	OptTableSpaceStorage
+
 %type <range>	qualified_name insert_target OptConstrFromTable
 
 %type <str>		all_Op MathOp
@@ -4931,18 +4934,35 @@ opt_procedural:
 /*****************************************************************************
  *
  *		QUERY:
- *             CREATE TABLESPACE tablespace LOCATION '/path/to/tablespace/'
+ *             CREATE TABLESPACE tablespace
+ *                 [ OWNER role ]
+ *                 [ LOCATION '/path/to/tablespace/' | USING smgr ( option [, ...] ) ]
+ *                 [ WITH ( option [ , ... ] ) ]
  *
  *****************************************************************************/
 
-CreateTableSpaceStmt: CREATE TABLESPACE name OptTableSpaceOwner LOCATION Sconst opt_reloptions
+CreateTableSpaceStmt: CREATE TABLESPACE name OptTableSpaceOwner OptTableSpaceStorage opt_reloptions
 				{
-					CreateTableSpaceStmt *n = makeNode(CreateTableSpaceStmt);
-
+					CreateTableSpaceStmt *n = (CreateTableSpaceStmt *) $5;
 					n->tablespacename = $3;
 					n->owner = $4;
-					n->location = $6;
-					n->options = $7;
+					n->options = $6;
+					$$ = (Node *) n;
+				}
+		;
+
+OptTableSpaceStorage: LOCATION Sconst
+				{
+					CreateTableSpaceStmt *n = makeNode(CreateTableSpaceStmt);
+					n->smgr = MD_SMGR_NAME;
+					n->smgropts = list_make1(makeDefElem("location", (Node *) makeString($2), @1));
+					$$ = (Node *) n;
+				}
+			| USING name '(' utility_option_list ')'
+				{
+					CreateTableSpaceStmt *n = makeNode(CreateTableSpaceStmt);
+					n->smgr = $2;
+					n->smgropts = $4;
 					$$ = (Node *) n;
 				}
 		;
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 690bdd27c5..dfc5a11da4 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -22,12 +22,16 @@
 #include "postgres.h"
 
 #include <unistd.h>
+#include <dirent.h>
 #include <fcntl.h>
 #include <sys/file.h>
+#include <sys/stat.h>
 
 #include "access/xlog.h"
 #include "access/xlogutils.h"
+#include "commands/defrem.h"
 #include "commands/tablespace.h"
+#include "common/file_perm.h"
 #include "miscadmin.h"
 #include "pg_trace.h"
 #include "pgstat.h"
@@ -156,6 +160,9 @@ void mdsmgr_register(void)
 		.smgr_nblocks = mdnblocks,
 		.smgr_truncate = mdtruncate,
 		.smgr_immedsync = mdimmedsync,
+		.smgr_validate_tspopts = mdvalidatetspopts,
+		.smgr_create_tsp = mdcreatetsp,
+		.smgr_drop_tsp = mddroptsp,
 	};
 
 	MdSMgrId = smgr_register(&md_smgr, sizeof(MdSMgrRelationData));
@@ -213,6 +220,7 @@ mdinit(void)
 bool
 mdexists(SMgrRelation reln, ForkNumber forknum)
 {
+	MdSMgrRelation mdreln = (MdSMgrRelation) reln;
 	/*
 	 * Close it first, to ensure that we notice if the fork has been unlinked
 	 * since we opened it.  As an optimization, we can skip that in recovery,
@@ -221,7 +229,7 @@ mdexists(SMgrRelation reln, ForkNumber forknum)
 	if (!InRecovery)
 		mdclose(reln, forknum);
 
-	return (mdopenfork(reln, forknum, EXTENSION_RETURN_NULL) != NULL);
+	return (mdopenfork(mdreln, forknum, EXTENSION_RETURN_NULL) != NULL);
 }
 
 /*
@@ -1672,3 +1680,207 @@ mdfiletagmatches(const FileTag *ftag, const FileTag *candidate)
 	 */
 	return ftag->rlocator.dbOid == candidate->rlocator.dbOid;
 }
+
+void mdvalidatetspopts(List *opts)
+{
+	ListCell   *option;
+	char	   *location;
+	bool		in_place;
+
+	if (list_length(opts) != 1)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_NAME),
+				 errmsg("too many storage options for the %s storage manager", MD_SMGR_NAME),
+				 errhint("Only LOCATION is supported")));
+
+	foreach(option, opts)
+	{
+		DefElem    *defel = lfirst_node(DefElem, option);
+
+		if (strcmp(defel->defname, "location") == 0)
+		{
+			location = pstrdup(defGetString(defel));
+		}
+		else
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("unrecognised option '%s' for to the %s storage manager",
+							defel->defname, MD_SMGR_NAME),
+					 errhint("Only 'location' is supported")),
+					 errposition(defel->location));
+		}
+	}
+
+	/* Unix-ify the offered path, and strip any trailing slashes */
+	canonicalize_path(location);
+
+	/* disallow quotes, else CREATE DATABASE would be at risk */
+	if (strchr(location, '\''))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_NAME),
+					errmsg("tablespace location cannot contain single quotes")));
+
+	in_place = allow_in_place_tablespaces && strlen(location) == 0;
+
+	/*
+	 * Allowing relative paths seems risky
+	 *
+	 * This also helps us ensure that location is not empty or whitespace,
+	 * unless specifying a developer-only in-place tablespace.
+	 */
+	if (!in_place && !is_absolute_path(location))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					errmsg("tablespace location must be an absolute path")));
+
+	/*
+	 * Check that location isn't too long. Remember that we're going to append
+	 * 'PG_XXX/<dboid>/<relid>_<fork>.<nnn>'.  FYI, we never actually
+	 * reference the whole path here, but MakePGDirectory() uses the first two
+	 * parts.
+	 */
+	if (strlen(location) + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) + 1 +
+		OIDCHARS + 1 + OIDCHARS + 1 + FORKNAMECHARS + 1 + OIDCHARS > MAXPGPATH)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					errmsg("tablespace location \"%s\" is too long",
+						   location)));
+
+	/* Warn if the tablespace is in the data directory. */
+	if (path_is_prefix_of_path(DataDir, location))
+		ereport(WARNING,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					errmsg("tablespace location should not be inside the data directory")));
+
+	pfree(location);
+}
+
+void mdcreatetsp(Oid tablespaceoid, List *opts, bool isredo)
+{
+	char	   *location;
+	DefElem	   *defel = (DefElem *) linitial_node(DefElem, opts);
+
+	Assert(strcmp(defel->defname, "location") == 0);
+	Assert(list_length(opts) == 1);
+
+	location = pstrdup(defGetString(defel));
+
+	/* Unix-ify the offered path, and strip any trailing slashes */
+	canonicalize_path(location);
+
+	create_tablespace_directories(location, tablespaceoid);
+
+	pfree(location);
+}
+
+void mddroptsp(Oid tsp, bool isredo)
+{
+	
+}
+
+/*
+ * create_tablespace_directories
+ *
+ *	Attempt to create filesystem infrastructure linking $PGDATA/pg_tblspc/
+ *	to the specified directory
+ */
+void
+create_tablespace_directories(const char *location, const Oid tablespaceoid)
+{
+	char	   *linkloc;
+	char	   *location_with_version_dir;
+	struct stat st;
+	bool		in_place;
+
+	linkloc = psprintf("pg_tblspc/%u", tablespaceoid);
+
+	/*
+	 * If we're asked to make an 'in place' tablespace, create the directory
+	 * directly where the symlink would normally go.  This is a developer-only
+	 * option for now, to facilitate regression testing.
+	 */
+	in_place = strlen(location) == 0;
+
+	if (in_place)
+	{
+		if (MakePGDirectory(linkloc) < 0 && errno != EEXIST)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+						errmsg("could not create directory \"%s\": %m",
+							   linkloc)));
+	}
+
+	location_with_version_dir = psprintf("%s/%s", in_place ? linkloc : location,
+										 TABLESPACE_VERSION_DIRECTORY);
+
+	/*
+	 * Attempt to coerce target directory to safe permissions.  If this fails,
+	 * it doesn't exist or has the wrong owner.  Not needed for in-place mode,
+	 * because in that case we created the directory with the desired
+	 * permissions.
+	 */
+	if (!in_place && chmod(location, pg_dir_create_mode) != 0)
+	{
+		if (errno == ENOENT)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_FILE),
+						errmsg("directory \"%s\" does not exist", location),
+						InRecovery ? errhint("Create this directory for the tablespace before "
+											 "restarting the server.") : 0));
+		else
+			ereport(ERROR,
+					(errcode_for_file_access(),
+						errmsg("could not set permissions on directory \"%s\": %m",
+							   location)));
+	}
+
+	/*
+	 * The creation of the version directory prevents more than one tablespace
+	 * in a single location.  This imitates TablespaceCreateDbspace(), but it
+	 * ignores concurrency and missing parent directories.  The chmod() would
+	 * have failed in the absence of a parent.  pg_tablespace_spcname_index
+	 * prevents concurrency.
+	 */
+	if (stat(location_with_version_dir, &st) < 0)
+	{
+		if (errno != ENOENT)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+						errmsg("could not stat directory \"%s\": %m",
+							   location_with_version_dir)));
+		else if (MakePGDirectory(location_with_version_dir) < 0)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+						errmsg("could not create directory \"%s\": %m",
+							   location_with_version_dir)));
+	}
+	else if (!S_ISDIR(st.st_mode))
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+					errmsg("\"%s\" exists but is not a directory",
+						   location_with_version_dir)));
+	else if (!InRecovery)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_IN_USE),
+					errmsg("directory \"%s\" already in use as a tablespace",
+						   location_with_version_dir)));
+
+	/*
+	 * In recovery, remove old symlink, in case it points to the wrong place.
+	 */
+	if (!in_place && InRecovery)
+		remove_tablespace_symlink(linkloc);
+
+	/*
+	 * Create the symlink under PGDATA
+	 */
+	if (!in_place && symlink(location, linkloc) < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+					errmsg("could not create symbolic link \"%s\": %m",
+						   linkloc)));
+
+	pfree(linkloc);
+	pfree(location_with_version_dir);
+}
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index d37202609f..b5cb720064 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -18,6 +18,7 @@
 #include "postgres.h"
 
 #include "access/xlogutils.h"
+#include "catalog/pg_tablespace_d.h"
 #include "lib/ilist.h"
 #include "miscadmin.h"
 #include "storage/bufmgr.h"
@@ -29,7 +30,7 @@
 #include "utils/hsearch.h"
 #include "utils/inval.h"
 #include "utils/memutils.h"
-
+#include "utils/spccache.h"
 
 static f_smgr *smgrsw;
 
@@ -174,13 +175,25 @@ smgropen(RelFileLocator rlocator, BackendId backend)
 	/* Initialize it if not present before */
 	if (!found)
 	{
+		Oid		tspid = reln->smgr_rlocator.locator.spcOid;
 		/* hash_search already filled in the lookup key */
 		reln->smgr_owner = NULL;
 		reln->smgr_targblock = InvalidBlockNumber;
 		for (int i = 0; i <= MAX_FORKNUM; ++i)
 			reln->smgr_cached_nblocks[i] = InvalidBlockNumber;
 
-		reln->smgr_which = MdSMgrId;	/* we only have md.c at present */
+		/*
+		 * There is a chicken-and-egg problem for determining which storage
+		 * manager to use for the global tablespace, as that holds the
+		 * pg_tablespace table which we'd use to look up this information.
+		 *
+		 * As the global tablespace can't be replaced, the default is used
+		 * instead, which is the md.c smgr (MD_SMGR_NAME).
+		 */
+		if (tspid == GLOBALTABLESPACE_OID || tspid == DEFAULTTABLESPACE_OID)
+			reln->smgr_which = get_smgr_id(MD_SMGR_NAME, false);
+		else
+			reln->smgr_which = get_tablespace_smgrid(tspid);
 
 		/* implementation-specific initialization */
 		smgrsw[reln->smgr_which].smgr_open(reln);
@@ -722,6 +735,61 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum)
 	smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum);
 }
 
+static const char *recent_smgrname = NULL;
+static SMgrId recent_smgrid = -1;
+
+static SMgrId get_smgr_by_name(const char *smgrname, bool missing_ok)
+{
+	if (recent_smgrname != NULL && strcmp(smgrname, recent_smgrname) == 0)
+		return recent_smgrid;
+
+	for (SMgrId id = 0; id < NSmgr; id++)
+	{
+		f_smgr *smgr = &smgrsw[id];
+
+		if (strcmp(smgrname, smgr->name) == 0)
+		{
+			recent_smgrname = smgr->name;
+			recent_smgrid = id;
+			return id;
+		}
+	}
+
+	if (missing_ok)
+		return InvalidSmgrId;
+
+	ereport(ERROR,
+			(errcode(ERRCODE_INVALID_NAME),
+			 errmsg("invalid smgr '%s'", smgrname)));
+}
+
+
+SMgrId get_smgr_id(const char *smgrname, bool missing_ok)
+{
+	return get_smgr_by_name(smgrname, missing_ok);
+}
+
+void smgrvalidatetspopts(const char *smgrname, List *opts)
+{
+	SMgrId smgrid = get_smgr_by_name(smgrname, false);
+
+	smgrsw[smgrid].smgr_validate_tspopts(opts);
+}
+
+void smgrcreatetsp(const char *smgrname, Oid tsp, List *opts, bool isredo)
+{
+	SMgrId smgrid = get_smgr_by_name(smgrname, false);
+
+	smgrsw[smgrid].smgr_create_tsp(tsp, opts, isredo);
+}
+
+void smgrdroptsp(const char *smgrname, Oid tsp, bool isredo)
+{
+	SMgrId smgrid = get_smgr_by_name(smgrname, false);
+
+	smgrsw[smgrid].smgr_drop_tsp(tsp, isredo);
+}
+
 /*
  * AtEOXact_SMgr
  *
diff --git a/src/backend/utils/cache/spccache.c b/src/backend/utils/cache/spccache.c
index 136fd737d3..ce7e403b53 100644
--- a/src/backend/utils/cache/spccache.c
+++ b/src/backend/utils/cache/spccache.c
@@ -24,6 +24,8 @@
 #include "miscadmin.h"
 #include "optimizer/optimizer.h"
 #include "storage/bufmgr.h"
+#include "storage/smgr.h"
+#include "storage/md.h"
 #include "utils/catcache.h"
 #include "utils/hsearch.h"
 #include "utils/inval.h"
@@ -38,6 +40,7 @@ static HTAB *TableSpaceCacheHash = NULL;
 typedef struct
 {
 	Oid			oid;			/* lookup key - must be first */
+	SMgrId		smgrid;			/* cached storage manager id */
 	TableSpaceOpts *opts;		/* options, or NULL if none */
 } TableSpaceCacheEntry;
 
@@ -98,7 +101,7 @@ InitializeTableSpaceCache(void)
 
 /*
  * get_tablespace
- *		Fetch TableSpaceCacheEntry structure for a specified table OID.
+ *		Fetch TableSpaceCacheEntry structure for a specified tablespace OID.
  *
  * Pointers returned by this function should not be stored, since a cache
  * flush will invalidate them.
@@ -109,6 +112,7 @@ get_tablespace(Oid spcid)
 	TableSpaceCacheEntry *spc;
 	HeapTuple	tp;
 	TableSpaceOpts *opts;
+	SMgrId		smgrid;
 
 	/*
 	 * Since spcid is always from a pg_class tuple, InvalidOid implies the
@@ -135,18 +139,32 @@ get_tablespace(Oid spcid)
 	 */
 	tp = SearchSysCache1(TABLESPACEOID, ObjectIdGetDatum(spcid));
 	if (!HeapTupleIsValid(tp))
+	{
 		opts = NULL;
+		smgrid = InvalidSmgrId;
+	}
 	else
 	{
 		Datum		datum;
 		bool		isNull;
+		char	   *smgrname;
+		
+		smgrname = NameStr(*DatumGetName(SysCacheGetAttr(TABLESPACEOID,
+														 tp,
+														 Anum_pg_tablespace_spcsmgr,
+														 &isNull)));
+
+		Assert(!isNull);
+		smgrid = get_smgr_id(smgrname, false);
 
 		datum = SysCacheGetAttr(TABLESPACEOID,
 								tp,
 								Anum_pg_tablespace_spcoptions,
 								&isNull);
 		if (isNull)
+		{
 			opts = NULL;
+		}
 		else
 		{
 			bytea	   *bytea_opts = tablespace_reloptions(datum, false);
@@ -167,6 +185,8 @@ get_tablespace(Oid spcid)
 											   HASH_ENTER,
 											   NULL);
 	spc->opts = opts;
+	spc->smgrid = smgrid;
+
 	return spc;
 }
 
@@ -235,3 +255,19 @@ get_tablespace_maintenance_io_concurrency(Oid spcid)
 	else
 		return spc->opts->maintenance_io_concurrency;
 }
+
+/*
+ * get_tablespace_smgrid
+ */
+SMgrId
+get_tablespace_smgrid(Oid spcid)
+{
+	TableSpaceCacheEntry *spc;
+	
+	if (spcid == GLOBALTABLESPACE_OID || spcid == DEFAULTTABLESPACE_OID)
+		return get_smgr_id(MD_SMGR_NAME, false);
+
+	spc = get_tablespace(spcid);
+
+	return spc->smgrid;
+}
diff --git a/src/include/catalog/pg_tablespace.dat b/src/include/catalog/pg_tablespace.dat
index 9fbc98a44d..5e20429619 100644
--- a/src/include/catalog/pg_tablespace.dat
+++ b/src/include/catalog/pg_tablespace.dat
@@ -13,8 +13,10 @@
 [
 
 { oid => '1663', oid_symbol => 'DEFAULTTABLESPACE_OID',
-  spcname => 'pg_default', spcacl => '_null_', spcoptions => '_null_' },
+  spcname => 'pg_default', spcacl => '_null_', spcsmgr => 'md',
+  spcoptions => '_null_' },
 { oid => '1664', oid_symbol => 'GLOBALTABLESPACE_OID',
-  spcname => 'pg_global', spcacl => '_null_', spcoptions => '_null_' },
+  spcname => 'pg_global', spcacl => '_null_', spcsmgr => 'md',
+  spcoptions => '_null_' },
 
 ]
diff --git a/src/include/catalog/pg_tablespace.h b/src/include/catalog/pg_tablespace.h
index ea1593d874..9385933c05 100644
--- a/src/include/catalog/pg_tablespace.h
+++ b/src/include/catalog/pg_tablespace.h
@@ -30,6 +30,7 @@ CATALOG(pg_tablespace,1213,TableSpaceRelationId) BKI_SHARED_RELATION
 {
 	Oid			oid;			/* oid */
 	NameData	spcname;		/* tablespace name */
+	NameData	spcsmgr;		/* tablespace storage manager */
 
 	/* owner of tablespace */
 	Oid			spcowner BKI_DEFAULT(POSTGRES) BKI_LOOKUP(pg_authid);
diff --git a/src/include/commands/tablespace.h b/src/include/commands/tablespace.h
index f1961c1813..15220ffb99 100644
--- a/src/include/commands/tablespace.h
+++ b/src/include/commands/tablespace.h
@@ -28,7 +28,8 @@ extern PGDLLIMPORT bool allow_in_place_tablespaces;
 typedef struct xl_tblspc_create_rec
 {
 	Oid			ts_id;
-	char		ts_path[FLEXIBLE_ARRAY_MEMBER]; /* null-terminated string */
+	NameData	ts_smgr;
+	char		ts_smgropts[FLEXIBLE_ARRAY_MEMBER];
 } xl_tblspc_create_rec;
 
 typedef struct xl_tblspc_drop_rec
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index b3bec90e52..e167acec7d 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2613,8 +2613,9 @@ typedef struct CreateTableSpaceStmt
 {
 	NodeTag		type;
 	char	   *tablespacename;
+	char	   *smgr;
+	List	   *smgropts; /* list of DefElem nodes */
 	RoleSpec   *owner;
-	char	   *location;
 	List	   *options;
 } CreateTableSpaceStmt;
 
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index beeddfd373..a397aa1c10 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -21,6 +21,8 @@
 
 /* registration function for md storage manager */
 extern void mdsmgr_register(void);
+
+#define MD_SMGR_NAME "md"
 extern SMgrId MdSMgrId;
 
 /* md storage manager functionality */
@@ -55,4 +57,12 @@ extern int	mdsyncfiletag(const FileTag *ftag, char *path);
 extern int	mdunlinkfiletag(const FileTag *ftag, char *path);
 extern bool mdfiletagmatches(const FileTag *ftag, const FileTag *candidate);
 
+/* md tsp callbacks */
+extern void mdvalidatetspopts(List *opts);
+extern void mdcreatetsp(Oid tsp, List *opts, bool isredo);
+extern void mddroptsp(Oid tsp, bool isredo);
+void create_tablespace_directories(const char *location,
+								   const Oid tablespaceoid);
+
+
 #endif							/* MD_H */
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index 5ad1d50e0c..12a9b5f00e 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -15,12 +15,18 @@
 #define SMGR_H
 
 #include "lib/ilist.h"
+#include "nodes/pg_list.h"
 #include "storage/block.h"
 #include "storage/relfilelocator.h"
 
-typedef uint8 SMgrId;
+/*
+ * volatile ID of the smgr. Across various configurations IDs may vary,
+ * true identity is the name of each smgr. 
+ */
+typedef int SMgrId;
 
-#define MaxSMgrId UINT8_MAX
+#define MaxSMgrId		INT_MAX
+#define InvalidSmgrId	(-1)
 
 /*
  * smgr.c maintains a table of SMgrRelation objects, which are essentially
@@ -113,8 +119,13 @@ typedef struct f_smgr
 	void		(*smgr_truncate) (SMgrRelation reln, ForkNumber forknum,
 								  BlockNumber nblocks);
 	void		(*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum);
+
+	void		(*smgr_validate_tspopts) (List *tspopts);
+	void		(*smgr_create_tsp) (Oid tspoid, List *tspopts, bool isredo);
+	void		(*smgr_drop_tsp) (Oid tspoid, bool isredo);
 } f_smgr;
 
+extern SMgrId get_smgr_id(const char *smgrname, bool missing_ok);
 extern SMgrId smgr_register(const f_smgr *smgr, Size smgrrelation_size);
 
 extern void smgrinit(void);
@@ -147,6 +158,11 @@ extern BlockNumber smgrnblocks_cached(SMgrRelation reln, ForkNumber forknum);
 extern void smgrtruncate(SMgrRelation reln, ForkNumber *forknum,
 						 int nforks, BlockNumber *nblocks);
 extern void smgrimmedsync(SMgrRelation reln, ForkNumber forknum);
+
+extern void smgrvalidatetspopts(const char *smgrname, List *opts);
+extern void smgrcreatetsp(const char *smgrname, Oid tsp, List *opts, bool isredo);
+extern void smgrdroptsp(const char *smgrname, Oid tsp, bool isredo);
+
 extern void AtEOXact_SMgr(void);
 extern bool ProcessBarrierSmgrRelease(void);
 
diff --git a/src/include/utils/spccache.h b/src/include/utils/spccache.h
index c6c754a2ec..6569452e91 100644
--- a/src/include/utils/spccache.h
+++ b/src/include/utils/spccache.h
@@ -12,10 +12,12 @@
  */
 #ifndef SPCCACHE_H
 #define SPCCACHE_H
+#include "storage/smgr.h"
 
 extern void get_tablespace_page_costs(Oid spcid, float8 *spc_random_page_cost,
 									  float8 *spc_seq_page_cost);
 extern int	get_tablespace_io_concurrency(Oid spcid);
 extern int	get_tablespace_maintenance_io_concurrency(Oid spcid);
+extern SMgrId get_tablespace_smgrid(Oid spcid);
 
 #endif							/* SPCCACHE_H */
-- 
2.39.0