Add pg_get_tablespace_ddl() function

Add a new SQL-callable function that returns the DDL statements needed
to recreate a tablespace. It takes a tablespace name or OID and an
optional VARIADIC text argument for options that are specified as
alternating name/value pairs. The following options are supported: pretty
(boolean) for formatted output and owner (boolean) to include OWNER.
(It includes two variants because there is no regtablespace pseudotype.)
The return is one or multiple rows where the first row is a CREATE
TABLESPACE statement and subsequent rows are ALTER TABLESPACE statements
to set some tablespace properties.

The caller must have SELECT privilege on pg_tablespace.

get_reloptions() in ruleutils.c is made non-static so it can be called
from the new ddlutils.c file.

Author: Nishant Sharma <nishant.sharma@enterprisedb.com>
Author: Manni Wood <manni.wood@enterprisedb.com>
Co-authored-by: Andrew Dunstan <andrew@dunslane.net>
Co-authored-by: Euler Taveira <euler@eulerto.com>
Reviewed-by: Jim Jones <jim.jones@uni-muenster.de>
Reviewed-by: Álvaro Herrera <alvherre@kurilemu.de>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://postgr.es/m/CAKWEB6rmnmGKUA87Zmq-s=b3Scsnj02C0kObQjnbL2ajfPWGEw@mail.gmail.com
Discussion: https://postgr.es/m/e247c261-e3fb-4810-81e0-a65893170e94@dunslane.net

doc/src/sgml/func/func-info.sgml

@@ -3910,6 +3910,34 @@ acl      | {postgres=arwdDxtm/postgres,foo=r/postgres}
         role memberships and their options.
        </para></entry>
       </row>
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_get_tablespace_ddl</primary>
+        </indexterm>
+        <function>pg_get_tablespace_ddl</function>
+        ( <parameter>tablespace</parameter> <type>oid</type>
+        <optional>, <literal>VARIADIC</literal> <parameter>options</parameter>
+        <type>text</type> </optional> )
+        <returnvalue>setof text</returnvalue>
+       </para>
+       <para>
+        <function>pg_get_tablespace_ddl</function>
+        ( <parameter>tablespace</parameter> <type>name</type>
+        <optional>, <literal>VARIADIC</literal> <parameter>options</parameter>
+        <type>text</type> </optional> )
+        <returnvalue>setof text</returnvalue>
+       </para>
+       <para>
+        Reconstructs the <command>CREATE TABLESPACE</command> statement for
+        the specified tablespace (by OID or name).  If the tablespace has
+        options set, an <command>ALTER TABLESPACE ... SET</command> statement
+        is also returned.  Each statement is returned as a separate row.
+        The following options are supported: <literal>pretty</literal> (boolean)
+        for formatted output and <literal>owner</literal> (boolean) to include
+        <literal>OWNER</literal>.
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>

src/backend/utils/adt/ddlutils.c

@@ -20,11 +20,13 @@
 
 #include "access/genam.h"
 #include "access/htup_details.h"
-#include "access/relation.h"
 #include "access/table.h"
 #include "catalog/pg_auth_members.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_db_role_setting.h"
+#include "catalog/pg_tablespace.h"
+#include "commands/tablespace.h"
+#include "common/relpath.h"
 #include "funcapi.h"
 #include "miscadmin.h"
 #include "utils/acl.h"
@@ -35,6 +37,7 @@
 #include "utils/guc.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
+#include "utils/ruleutils.h"
 #include "utils/syscache.h"
 #include "utils/timestamp.h"
 #include "utils/varlena.h"
@@ -75,6 +78,8 @@ static void append_guc_value(StringInfo buf, const char *name,
 							 const char *value);
 static List *pg_get_role_ddl_internal(Oid roleid, bool pretty,
 									  bool memberships);
+static List *pg_get_tablespace_ddl_internal(Oid tsid, bool pretty, bool no_owner);
+static Datum pg_get_tablespace_ddl_srf(FunctionCallInfo fcinfo, Oid tsid, bool isnull);
 
 
 /*
@@ -634,3 +639,202 @@ pg_get_role_ddl(PG_FUNCTION_ARGS)
 		SRF_RETURN_DONE(funcctx);
 	}
 }
+
+/*
+ * pg_get_tablespace_ddl_internal
+ *		Generate DDL statements to recreate a tablespace.
+ *
+ * Returns a List of palloc'd strings.  The first element is the
+ * CREATE TABLESPACE statement; if the tablespace has reloptions,
+ * a second element with ALTER TABLESPACE SET (...) is appended.
+ */
+static List *
+pg_get_tablespace_ddl_internal(Oid tsid, bool pretty, bool no_owner)
+{
+	HeapTuple	tuple;
+	Form_pg_tablespace tspForm;
+	StringInfoData buf;
+	char	   *spcname;
+	char	   *spcowner;
+	char	   *path;
+	bool		isNull;
+	Datum		datum;
+	List	   *statements = NIL;
+
+	tuple = SearchSysCache1(TABLESPACEOID, ObjectIdGetDatum(tsid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("tablespace with OID %u does not exist",
+						tsid)));
+
+	tspForm = (Form_pg_tablespace) GETSTRUCT(tuple);
+	spcname = pstrdup(NameStr(tspForm->spcname));
+
+	/* User must have SELECT privilege on pg_tablespace. */
+	if (pg_class_aclcheck(TableSpaceRelationId, GetUserId(), ACL_SELECT) != ACLCHECK_OK)
+	{
+		ReleaseSysCache(tuple);
+		aclcheck_error(ACLCHECK_NO_PRIV, OBJECT_TABLESPACE, spcname);
+	}
+
+	/*
+	 * We don't support generating DDL for system tablespaces.  The primary
+	 * reason for this is that users shouldn't be recreating them.
+	 */
+	if (IsReservedName(spcname))
+		ereport(ERROR,
+				(errcode(ERRCODE_RESERVED_NAME),
+				 errmsg("tablespace name \"%s\" is reserved", spcname),
+				 errdetail("Tablespace names starting with \"pg_\" are reserved for system tablespaces.")));
+
+	initStringInfo(&buf);
+
+	/* Start building the CREATE TABLESPACE statement */
+	appendStringInfo(&buf, "CREATE TABLESPACE %s", quote_identifier(spcname));
+
+	/* Add OWNER clause */
+	if (!no_owner)
+	{
+		spcowner = GetUserNameFromId(tspForm->spcowner, false);
+		append_ddl_option(&buf, pretty, 4, "OWNER %s",
+						  quote_identifier(spcowner));
+		pfree(spcowner);
+	}
+
+	/* Find tablespace directory path */
+	path = get_tablespace_location(tsid);
+
+	/* Add directory LOCATION (path), if it exists */
+	if (path[0] != '\0')
+	{
+		/*
+		 * Special case: if the tablespace was created with GUC
+		 * "allow_in_place_tablespaces = true" and "LOCATION ''", path will
+		 * begin with "pg_tblspc/". In that case, show "LOCATION ''" as the
+		 * user originally specified.
+		 */
+		if (strncmp(PG_TBLSPC_DIR_SLASH, path, strlen(PG_TBLSPC_DIR_SLASH)) == 0)
+			append_ddl_option(&buf, pretty, 4, "LOCATION ''");
+		else
+			append_ddl_option(&buf, pretty, 4, "LOCATION %s",
+							  quote_literal_cstr(path));
+	}
+	pfree(path);
+
+	appendStringInfoChar(&buf, ';');
+	statements = lappend(statements, pstrdup(buf.data));
+
+	/* Check for tablespace options */
+	datum = SysCacheGetAttr(TABLESPACEOID, tuple,
+							Anum_pg_tablespace_spcoptions, &isNull);
+	if (!isNull)
+	{
+		resetStringInfo(&buf);
+		appendStringInfo(&buf, "ALTER TABLESPACE %s SET (",
+						 quote_identifier(spcname));
+		get_reloptions(&buf, datum);
+		appendStringInfoString(&buf, ");");
+		statements = lappend(statements, pstrdup(buf.data));
+	}
+
+	ReleaseSysCache(tuple);
+	pfree(spcname);
+	pfree(buf.data);
+
+	return statements;
+}
+
+/*
+ * pg_get_tablespace_ddl_srf - common SRF logic for tablespace DDL
+ */
+static Datum
+pg_get_tablespace_ddl_srf(FunctionCallInfo fcinfo, Oid tsid, bool isnull)
+{
+	FuncCallContext *funcctx;
+	List	   *statements;
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		MemoryContext oldcontext;
+		DdlOption	opts[] = {
+			{"pretty", DDL_OPT_BOOL},
+			{"owner", DDL_OPT_BOOL},
+		};
+
+		funcctx = SRF_FIRSTCALL_INIT();
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		if (isnull)
+		{
+			MemoryContextSwitchTo(oldcontext);
+			SRF_RETURN_DONE(funcctx);
+		}
+
+		parse_ddl_options(fcinfo, 1, opts, lengthof(opts));
+
+		statements = pg_get_tablespace_ddl_internal(tsid,
+													opts[0].isset && opts[0].boolval,
+													opts[1].isset && !opts[1].boolval);
+		funcctx->user_fctx = statements;
+		funcctx->max_calls = list_length(statements);
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	funcctx = SRF_PERCALL_SETUP();
+	statements = (List *) funcctx->user_fctx;
+
+	if (funcctx->call_cntr < funcctx->max_calls)
+	{
+		char	   *stmt;
+
+		stmt = (char *) list_nth(statements, funcctx->call_cntr);
+
+		SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(stmt));
+	}
+	else
+	{
+		list_free_deep(statements);
+		SRF_RETURN_DONE(funcctx);
+	}
+}
+
+/*
+ * pg_get_tablespace_ddl_oid
+ *		Return DDL to recreate a tablespace, taking OID.
+ */
+Datum
+pg_get_tablespace_ddl_oid(PG_FUNCTION_ARGS)
+{
+	Oid			tsid = InvalidOid;
+	bool		isnull;
+
+	isnull = PG_ARGISNULL(0);
+	if (!isnull)
+		tsid = PG_GETARG_OID(0);
+
+	return pg_get_tablespace_ddl_srf(fcinfo, tsid, isnull);
+}
+
+/*
+ * pg_get_tablespace_ddl_name
+ *		Return DDL to recreate a tablespace, taking name.
+ */
+Datum
+pg_get_tablespace_ddl_name(PG_FUNCTION_ARGS)
+{
+	Oid			tsid = InvalidOid;
+	Name		tspname;
+	bool		isnull;
+
+	isnull = PG_ARGISNULL(0);
+
+	if (!isnull)
+	{
+		tspname = PG_GETARG_NAME(0);
+		tsid = get_tablespace_oid(NameStr(*tspname), false);
+	}
+
+	return pg_get_tablespace_ddl_srf(fcinfo, tsid, isnull);
+}

src/backend/utils/adt/ruleutils.c

@@ -547,7 +547,7 @@ static void add_cast_to(StringInfo buf, Oid typid);
 static char *generate_qualified_type_name(Oid typid);
 static text *string_to_text(char *str);
 static char *flatten_reloptions(Oid relid);
-static void get_reloptions(StringInfo buf, Datum reloptions);
+void		get_reloptions(StringInfo buf, Datum reloptions);
 static void get_json_path_spec(Node *path_spec, deparse_context *context,
 							   bool showimplicit);
 static void get_json_table_columns(TableFunc *tf, JsonTablePathScan *scan,
@@ -14240,7 +14240,7 @@ string_to_text(char *str)
 /*
  * Generate a C string representing a relation options from text[] datum.
  */
-static void
+void
 get_reloptions(StringInfo buf, Datum reloptions)
 {
 	Datum	   *options;

src/include/catalog/pg_proc.dat

@@ -8611,6 +8611,22 @@
   proallargtypes => '{regrole,text}',
   pronargdefaults => '1', proargdefaults => '{NULL}',
   prosrc => 'pg_get_role_ddl' },
+{ oid => '8758', descr => 'get DDL to recreate a tablespace',
+  proname => 'pg_get_tablespace_ddl', provariadic => 'text', proisstrict => 'f',
+  provolatile => 's', proretset => 't', prorows => '10', prorettype => 'text',
+  proargtypes => 'oid text',
+  proargmodes => '{i,v}',
+  proallargtypes => '{oid,text}',
+  pronargdefaults => '1', proargdefaults => '{NULL}',
+  prosrc => 'pg_get_tablespace_ddl_oid' },
+{ oid => '8759', descr => 'get DDL to recreate a tablespace',
+  proname => 'pg_get_tablespace_ddl', provariadic => 'text', proisstrict => 'f',
+  provolatile => 's', proretset => 't', prorows => '10', prorettype => 'text',
+  proargtypes => 'name text',
+  proargmodes => '{i,v}',
+  proallargtypes => '{name,text}',
+  pronargdefaults => '1', proargdefaults => '{NULL}',
+  prosrc => 'pg_get_tablespace_ddl_name' },
 { oid => '2509',
   descr => 'deparse an encoded expression with pretty-print option',
   proname => 'pg_get_expr', provolatile => 's', prorettype => 'text',

src/include/utils/ruleutils.h

@@ -51,6 +51,7 @@ extern char *get_window_frame_options_for_explain(int frameOptions,
 extern char *generate_collation_name(Oid collid);
 extern char *generate_opclass_name(Oid opclass);
 extern char *get_range_partbound_string(List *bound_datums);
+extern void get_reloptions(StringInfo buf, Datum reloptions);
 
 extern char *pg_get_statisticsobjdef_string(Oid statextid);
 

src/test/regress/expected/tablespace_ddl.out

@@ -0,0 +1,84 @@
+--
+-- Tests for pg_get_tablespace_ddl()
+--
+SET allow_in_place_tablespaces = true;
+CREATE ROLE regress_tblspc_ddl_user;
+-- error: non-existent tablespace by name
+SELECT * FROM pg_get_tablespace_ddl('regress_nonexistent_tblsp');
+ERROR:  tablespace "regress_nonexistent_tblsp" does not exist
+-- error: non-existent tablespace by OID
+SELECT * FROM pg_get_tablespace_ddl(0::oid);
+ERROR:  tablespace with OID 0 does not exist
+-- NULL input returns no rows (name variant)
+SELECT * FROM pg_get_tablespace_ddl(NULL::name);
+ pg_get_tablespace_ddl 
+-----------------------
+(0 rows)
+
+-- NULL input returns no rows (OID variant)
+SELECT * FROM pg_get_tablespace_ddl(NULL::oid);
+ pg_get_tablespace_ddl 
+-----------------------
+(0 rows)
+
+-- tablespace name requiring quoting
+CREATE TABLESPACE "regress_ tblsp" OWNER regress_tblspc_ddl_user LOCATION '';
+SELECT * FROM pg_get_tablespace_ddl('regress_ tblsp');
+                             pg_get_tablespace_ddl                             
+-------------------------------------------------------------------------------
+ CREATE TABLESPACE "regress_ tblsp" OWNER regress_tblspc_ddl_user LOCATION '';
+(1 row)
+
+DROP TABLESPACE "regress_ tblsp";
+-- tablespace with multiple options
+CREATE TABLESPACE regress_allopt_tblsp OWNER regress_tblspc_ddl_user LOCATION ''
+  WITH (seq_page_cost = '1.5', random_page_cost = '1.1234567890',
+        effective_io_concurrency = '17', maintenance_io_concurrency = '18');
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp');
+                                                                       pg_get_tablespace_ddl                                                                       
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLESPACE regress_allopt_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+ ALTER TABLESPACE regress_allopt_tblsp SET (seq_page_cost='1.5', random_page_cost='1.1234567890', effective_io_concurrency='17', maintenance_io_concurrency='18');
+(2 rows)
+
+-- pretty-printed output
+\pset format unaligned
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp', 'pretty', 'true');
+pg_get_tablespace_ddl
+CREATE TABLESPACE regress_allopt_tblsp
+    OWNER regress_tblspc_ddl_user
+    LOCATION '';
+ALTER TABLESPACE regress_allopt_tblsp SET (seq_page_cost='1.5', random_page_cost='1.1234567890', effective_io_concurrency='17', maintenance_io_concurrency='18');
+(2 rows)
+\pset format aligned
+-- tablespace with owner suppressed
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp', 'owner', 'false');
+                                                                       pg_get_tablespace_ddl                                                                       
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ CREATE TABLESPACE regress_allopt_tblsp LOCATION '';
+ ALTER TABLESPACE regress_allopt_tblsp SET (seq_page_cost='1.5', random_page_cost='1.1234567890', effective_io_concurrency='17', maintenance_io_concurrency='18');
+(2 rows)
+
+DROP TABLESPACE regress_allopt_tblsp;
+-- test by OID
+CREATE TABLESPACE regress_oid_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+SELECT oid AS tsid FROM pg_tablespace WHERE spcname = 'regress_oid_tblsp' \gset
+SELECT * FROM pg_get_tablespace_ddl(:tsid);
+                             pg_get_tablespace_ddl                              
+--------------------------------------------------------------------------------
+ CREATE TABLESPACE regress_oid_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+(1 row)
+
+DROP TABLESPACE regress_oid_tblsp;
+-- Permission check: revoke SELECT on pg_tablespace
+CREATE TABLESPACE regress_acl_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+CREATE ROLE regress_tblspc_ddl_noaccess;
+REVOKE SELECT ON pg_tablespace FROM PUBLIC;
+SET ROLE regress_tblspc_ddl_noaccess;
+SELECT * FROM pg_get_tablespace_ddl('regress_acl_tblsp');  -- should fail
+ERROR:  permission denied for tablespace regress_acl_tblsp
+RESET ROLE;
+GRANT SELECT ON pg_tablespace TO PUBLIC;
+DROP TABLESPACE regress_acl_tblsp;
+DROP ROLE regress_tblspc_ddl_noaccess;
+DROP ROLE regress_tblspc_ddl_user;

src/test/regress/parallel_schedule

@@ -130,7 +130,7 @@ test: partition_merge partition_split partition_join partition_prune reloptions
 # oidjoins is read-only, though, and should run late for best coverage
 test: oidjoins event_trigger
 
-test: role_ddl
+test: role_ddl tablespace_ddl
 
 # event_trigger_login cannot run concurrently with any other tests because
 # on-login event handling could catch connection of a concurrent test.

src/test/regress/sql/tablespace_ddl.sql

@@ -0,0 +1,58 @@
+--
+-- Tests for pg_get_tablespace_ddl()
+--
+
+SET allow_in_place_tablespaces = true;
+CREATE ROLE regress_tblspc_ddl_user;
+
+-- error: non-existent tablespace by name
+SELECT * FROM pg_get_tablespace_ddl('regress_nonexistent_tblsp');
+
+-- error: non-existent tablespace by OID
+SELECT * FROM pg_get_tablespace_ddl(0::oid);
+
+-- NULL input returns no rows (name variant)
+SELECT * FROM pg_get_tablespace_ddl(NULL::name);
+
+-- NULL input returns no rows (OID variant)
+SELECT * FROM pg_get_tablespace_ddl(NULL::oid);
+
+-- tablespace name requiring quoting
+CREATE TABLESPACE "regress_ tblsp" OWNER regress_tblspc_ddl_user LOCATION '';
+SELECT * FROM pg_get_tablespace_ddl('regress_ tblsp');
+DROP TABLESPACE "regress_ tblsp";
+
+-- tablespace with multiple options
+CREATE TABLESPACE regress_allopt_tblsp OWNER regress_tblspc_ddl_user LOCATION ''
+  WITH (seq_page_cost = '1.5', random_page_cost = '1.1234567890',
+        effective_io_concurrency = '17', maintenance_io_concurrency = '18');
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp');
+
+-- pretty-printed output
+\pset format unaligned
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp', 'pretty', 'true');
+\pset format aligned
+
+-- tablespace with owner suppressed
+SELECT * FROM pg_get_tablespace_ddl('regress_allopt_tblsp', 'owner', 'false');
+
+DROP TABLESPACE regress_allopt_tblsp;
+
+-- test by OID
+CREATE TABLESPACE regress_oid_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+SELECT oid AS tsid FROM pg_tablespace WHERE spcname = 'regress_oid_tblsp' \gset
+SELECT * FROM pg_get_tablespace_ddl(:tsid);
+DROP TABLESPACE regress_oid_tblsp;
+
+-- Permission check: revoke SELECT on pg_tablespace
+CREATE TABLESPACE regress_acl_tblsp OWNER regress_tblspc_ddl_user LOCATION '';
+CREATE ROLE regress_tblspc_ddl_noaccess;
+REVOKE SELECT ON pg_tablespace FROM PUBLIC;
+SET ROLE regress_tblspc_ddl_noaccess;
+SELECT * FROM pg_get_tablespace_ddl('regress_acl_tblsp');  -- should fail
+RESET ROLE;
+GRANT SELECT ON pg_tablespace TO PUBLIC;
+DROP TABLESPACE regress_acl_tblsp;
+DROP ROLE regress_tblspc_ddl_noaccess;
+
+DROP ROLE regress_tblspc_ddl_user;