alter-type-set-storage-v1.patch

diff --git a/doc/src/sgml/ref/alter_type.sgml b/doc/src/sgml/ref/alter_type.sgml
index 67be1dd568..03a6a59468 100644
--- a/doc/src/sgml/ref/alter_type.sgml
+++ b/doc/src/sgml/ref/alter_type.sgml
@@ -30,6 +30,7 @@ ALTER TYPE <replaceable class="parameter">name</replaceable> RENAME TO <replacea
 ALTER TYPE <replaceable class="parameter">name</replaceable> SET SCHEMA <replaceable class="parameter">new_schema</replaceable>
 ALTER TYPE <replaceable class="parameter">name</replaceable> ADD VALUE [ IF NOT EXISTS ] <replaceable class="parameter">new_enum_value</replaceable> [ { BEFORE | AFTER } <replaceable class="parameter">neighbor_enum_value</replaceable> ]
 ALTER TYPE <replaceable class="parameter">name</replaceable> RENAME VALUE <replaceable class="parameter">existing_enum_value</replaceable> TO <replaceable class="parameter">new_enum_value</replaceable>
+ALTER TYPE <replaceable class="parameter">name</replaceable> SET STORAGE { PLAIN | EXTERNAL | EXTENDED | MAIN }
 
 <phrase>where <replaceable class="parameter">action</replaceable> is one of:</phrase>
 
@@ -97,6 +98,38 @@ ALTER TYPE <replaceable class="parameter">name</replaceable> RENAME VALUE <repla
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term>
+     <literal>SET STORAGE</literal>
+     <indexterm>
+      <primary>TOAST</primary>
+      <secondary>per-type storage settings</secondary>
+     </indexterm>
+    </term>
+    <listitem>
+     <para>
+      This form sets the storage mode for a data type, controlling whether the
+      values are held inline or in a secondary <acronym>TOAST</acronym> table,
+      and whether the data should be compressed or not.
+      <literal>PLAIN</literal> must be used for fixed-length values such as
+      <type>integer</type> and is inline, uncompressed. <literal>MAIN</literal>
+      is for inline, compressible data. <literal>EXTERNAL</literal> is for
+      external, uncompressed data, and <literal>EXTENDED</literal> is for
+      external, compressed data.  <literal>EXTENDED</literal> is the default
+      for most data types that support non-<literal>PLAIN</literal> storage.
+      Use of <literal>EXTERNAL</literal> will make substring operations on
+      very large <type>text</type> and <type>bytea</type> values run faster,
+      at the penalty of increased storage space.  Note that
+      <literal>SET STORAGE</literal> doesn't itself change anything in the
+      tables, it just sets the strategy to be used by tables created in the
+      future. See <xref linkend="storage-toast"/> for more information.
+      Note that this merely modifies the default for a data type, but each
+      attribute specifies it's own strategy that overrides this value.
+      See <xref linkend="sql-altertable"/> for how to change that.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>SET SCHEMA</literal></term>
     <listitem>
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index 1cb84182b0..d021b1d272 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -1042,3 +1042,48 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
 
 	InvokeObjectPostAlterHook(classId, objectId, 0);
 }
+
+/*
+ * Executes an ALTER TYPE / SET STORAGE statement.  At the moment this is
+ * supported only for OBJECT_TYPE nad OBJECT_DOMAIN.
+ */
+ObjectAddress
+ExecAlterTypeStorageStmt(AlterTypeStorageStmt *stmt)
+{
+	char   *storagemode;
+	char	newstorage;
+
+	Assert(IsA(stmt->newstorage, String));
+	storagemode = strVal(stmt->newstorage);
+
+	if (pg_strcasecmp(storagemode, "plain") == 0)
+		newstorage = 'p';
+	else if (pg_strcasecmp(storagemode, "external") == 0)
+		newstorage = 'e';
+	else if (pg_strcasecmp(storagemode, "extended") == 0)
+		newstorage = 'x';
+	else if (pg_strcasecmp(storagemode, "main") == 0)
+		newstorage = 'm';
+	else
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("invalid storage type \"%s\"",
+						storagemode)));
+		newstorage = 0;			/* keep compiler quiet */
+	}
+
+	switch (stmt->objectType)
+	{
+		case OBJECT_TYPE:
+		case OBJECT_DOMAIN:		/* same as TYPE */
+			return AlterTypeStorage(castNode(List, stmt->object),
+									newstorage, stmt->objectType);
+			break;
+
+		default:
+			elog(ERROR, "unrecognized AlterTypeStorageStmt type: %d",
+				 (int) stmt->objectType);
+			return InvalidObjectAddress;	/* keep compiler happy */
+	}
+}
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 52097363fd..3909ed6b72 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -41,6 +41,7 @@
 #include "catalog/heap.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_attribute.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
@@ -3394,6 +3395,187 @@ AlterTypeOwner(List *names, Oid newOwnerId, ObjectType objecttype)
 	return address;
 }
 
+static bool
+type_has_toasted_attributes(Oid typeoid)
+{
+	Relation	attrel;
+	SysScanDesc scan;
+	ScanKeyData key[1];
+	HeapTuple	tuple;
+	bool		found = false;	/* no TOAST-ed attributes found */
+
+	/*
+	 * Must scan pg_attribute.  Right now, it is a seqscan because there is
+	 * no available index on atttypid.
+	 */
+
+	attrel = table_open(AttributeRelationId, AccessShareLock);
+
+	/* Use the index to scan only attributes of the target relation */
+	ScanKeyInit(&key[0],
+				Anum_pg_attribute_atttypid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(typeoid));
+
+	scan = systable_beginscan(attrel, InvalidOid, true,
+							  NULL, 1, key);
+
+	/* There should be at most one matching tuple, but we loop anyway */
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+	{
+		Form_pg_attribute attr = (Form_pg_attribute) GETSTRUCT(tuple);
+
+		if (attr->attstorage != 'p')
+		{
+			found = true;
+			break;
+		}
+	}
+
+	/* Clean up after the scan */
+	systable_endscan(scan);
+	table_close(attrel, AccessShareLock);
+
+	return found;
+}
+
+/*
+ * Change the storage of a type.
+ */
+ObjectAddress
+AlterTypeStorage(List *names, char newStorage, ObjectType objecttype)
+{
+	TypeName   *typename;
+	Oid			typeOid;
+	Relation	rel;
+	HeapTuple	tup;
+	HeapTuple	newtup;
+	Form_pg_type typTup;
+	ObjectAddress address;
+
+	rel = table_open(TypeRelationId, RowExclusiveLock);
+
+	/* Make a TypeName so we can use standard type lookup machinery */
+	typename = makeTypeNameFromNameList(names);
+
+	/* Use LookupTypeName here so that shell types can be processed */
+	tup = LookupTypeName(NULL, typename, NULL, false);
+	if (tup == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("type \"%s\" does not exist",
+						TypeNameToString(typename))));
+	typeOid = typeTypeId(tup);
+
+	/* Copy the syscache entry so we can scribble on it below */
+	newtup = heap_copytuple(tup);
+	ReleaseSysCache(tup);
+	tup = newtup;
+	typTup = (Form_pg_type) GETSTRUCT(tup);
+
+	/* Don't allow ALTER DOMAIN on a type */
+	if (objecttype == OBJECT_DOMAIN && typTup->typtype != TYPTYPE_DOMAIN)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("%s is not a domain",
+						format_type_be(typeOid))));
+
+	/*
+	 * If it's a composite type, we need to check that it really is a
+	 * free-standing composite type, and not a table's rowtype. We want people
+	 * to use ALTER TABLE not ALTER TYPE for that case.
+	 *
+	 * XXX Maybe this should be allowed?
+	 */
+	if (typTup->typtype == TYPTYPE_COMPOSITE &&
+		get_rel_relkind(typTup->typrelid) != RELKIND_COMPOSITE_TYPE)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("%s is a table's row type",
+						format_type_be(typeOid)),
+				 errhint("Use ALTER TABLE instead.")));
+
+	/*
+	 * don't allow direct alteration of array types, either
+	 *
+	 * XXX Maybe this should be allowed?
+	 */
+	if (OidIsValid(typTup->typelem) &&
+		get_array_type(typTup->typelem) == typeOid)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("cannot alter array type %s",
+						format_type_be(typeOid)),
+				 errhint("You can alter type %s, which will alter the array type as well.",
+						 format_type_be(typTup->typelem))));
+
+	/*
+	 * If the new storage is the same as the existing storage, consider the
+	 * command to have succeeded.
+	 */
+	if (typTup->typstorage != newStorage)
+	{
+		Datum	repl_val[Natts_pg_type];
+		bool	repl_null[Natts_pg_type];
+		bool	repl_repl[Natts_pg_type];
+
+		/* Check the user has right to do ALTER TYPE */
+		if (!pg_type_ownercheck(typTup->oid, GetUserId()))
+			aclcheck_error_type(ACLCHECK_NOT_OWNER, typTup->oid);
+
+		/*
+		 * Verify the transition to the new storage value is allowed given the
+		 * other type parameters (especially length) and existing attributes
+		 * using the type.
+		 */
+
+		/* Only varlena types can be toasted */
+		if (newStorage != 'p' && typTup->typlen != -1)
+			ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("fixed-size types must have storage PLAIN")));
+
+		/*
+		 * We must not allow switching to PLAIN when there are attributes with
+		 * this type using some other storage value. PLAIN may indicate the
+		 * type code does not expect the values to be TOAST-ed, so allowing
+		 * that at the attribute level might lead to crashes.
+		 *
+		 * Any other transition is correct, because it makes the type TOAST
+		 * aware. But it's allowed to restrict it at the attribute level, or
+		 * enable/disable compression.
+		 *
+		 * So maybe 
+		 *
+		 * XXX Maybe this could also support CASCADE mode, i.e. we'd set the
+		 * storage strategy for all attributes using the data type.
+		 */
+		if (newStorage == 'p' && type_has_toasted_attributes(typeOid))
+			ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("can't set storage to PLAIN when there are attributes with non-PLAIN storage values")));
+
+		/* do the actual change */
+		memset(repl_null, false, sizeof(repl_null));
+		memset(repl_repl, false, sizeof(repl_repl));
+
+		repl_repl[Anum_pg_type_typstorage - 1] = true;
+		repl_val[Anum_pg_type_typstorage - 1] = CharGetDatum(newStorage);
+
+		tup = heap_modify_tuple(tup, RelationGetDescr(rel), repl_val, repl_null,
+								repl_repl);
+
+		CatalogTupleUpdate(rel, &tup->t_self, tup);
+	}
+
+	ObjectAddressSet(address, TypeRelationId, typeOid);
+
+	/* Clean up */
+	table_close(rel, RowExclusiveLock);
+
+	return address;
+}
+
 /*
  * AlterTypeOwner_oid - change type owner unconditionally
  *
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 96e7fdbcfe..3249a67eca 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -253,7 +253,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 		AlterTblSpcStmt AlterExtensionStmt AlterExtensionContentsStmt AlterForeignTableStmt
 		AlterCompositeTypeStmt AlterUserMappingStmt
 		AlterRoleStmt AlterRoleSetStmt AlterPolicyStmt AlterStatsStmt
-		AlterDefaultPrivilegesStmt DefACLAction
+		AlterDefaultPrivilegesStmt AlterTypeStorageStmt DefACLAction
 		AnalyzeStmt CallStmt ClosePortalStmt ClusterStmt CommentStmt
 		ConstraintsSetStmt CopyStmt CreateAsStmt CreateCastStmt
 		CreateDomainStmt CreateExtensionStmt CreateGroupStmt CreateOpClassStmt
@@ -847,6 +847,7 @@ stmt :
 			| AlterObjectSchemaStmt
 			| AlterOwnerStmt
 			| AlterOperatorStmt
+			| AlterTypeStorageStmt
 			| AlterPolicyStmt
 			| AlterSeqStmt
 			| AlterSystemStmt
@@ -9566,6 +9567,31 @@ AlterOwnerStmt: ALTER AGGREGATE aggregate_with_argtypes OWNER TO RoleSpec
 		;
 
 
+/*****************************************************************************
+ *
+ * ALTER TTYPE name SET STORAGE storage
+ *
+ *****************************************************************************/
+
+AlterTypeStorageStmt:
+			ALTER TYPE_P any_name SET STORAGE ColId
+				{
+					AlterTypeStorageStmt *n = makeNode(AlterTypeStorageStmt);
+					n->objectType = OBJECT_TYPE;
+					n->object = (Node *) $3;
+					n->newstorage = (Node *) makeString($6);
+					$$ = (Node *)n;
+				}
+			| ALTER DOMAIN_P any_name SET STORAGE ColId
+				{
+					AlterTypeStorageStmt *n = makeNode(AlterTypeStorageStmt);
+					n->objectType = OBJECT_DOMAIN;
+					n->object = (Node *) $3;
+					n->newstorage = (Node *) makeString($6);
+					$$ = (Node *)n;
+				}
+		;
+
 /*****************************************************************************
  *
  * CREATE PUBLICATION name [ FOR TABLE ] [ WITH options ]
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index bb85b5e52a..3f3da70b88 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -158,6 +158,7 @@ ClassifyUtilityCommandAsReadOnly(Node *parsetree)
 		case T_AlterSeqStmt:
 		case T_AlterStatsStmt:
 		case T_AlterSubscriptionStmt:
+		case T_AlterTypeStorageStmt:
 		case T_AlterTSConfigurationStmt:
 		case T_AlterTSDictionaryStmt:
 		case T_AlterTableMoveAllStmt:
@@ -1045,6 +1046,19 @@ standard_ProcessUtility(PlannedStmt *pstmt,
 			}
 			break;
 
+		case T_AlterTypeStorageStmt:
+			{
+				AlterTypeStorageStmt *stmt = (AlterTypeStorageStmt *) parsetree;
+
+				if (EventTriggerSupportsObjectType(stmt->objectType))
+					ProcessUtilitySlow(pstate, pstmt, queryString,
+									   context, params, queryEnv,
+									   dest, completionTag);
+				else
+					ExecAlterTypeStorageStmt(stmt);
+			}
+			break;
+
 		case T_CommentStmt:
 			{
 				CommentStmt *stmt = (CommentStmt *) parsetree;
@@ -1723,6 +1737,10 @@ ProcessUtilitySlow(ParseState *pstate,
 				address = AlterOperator((AlterOperatorStmt *) parsetree);
 				break;
 
+			case T_AlterTypeStorageStmt:
+				address = ExecAlterTypeStorageStmt((AlterTypeStorageStmt *) parsetree);
+				break;
+
 			case T_CommentStmt:
 				address = CommentObject((CommentStmt *) parsetree);
 				break;
@@ -2575,6 +2593,10 @@ CreateCommandTag(Node *parsetree)
 			tag = AlterObjectTypeCommandTag(((AlterOwnerStmt *) parsetree)->objectType);
 			break;
 
+		case T_AlterTypeStorageStmt:
+			tag = "ALTER TYPE";
+			break;
+
 		case T_AlterTableMoveAllStmt:
 			tag = AlterObjectTypeCommandTag(((AlterTableMoveAllStmt *) parsetree)->objtype);
 			break;
@@ -3264,6 +3286,10 @@ GetCommandLogLevel(Node *parsetree)
 			lev = LOGSTMT_DDL;
 			break;
 
+		case T_AlterTypeStorageStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
 		case T_AlterTableMoveAllStmt:
 		case T_AlterTableStmt:
 			lev = LOGSTMT_DDL;
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index b6b08d0ccb..9a98306b06 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2150,7 +2150,7 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("ALTER", "TYPE", MatchAny))
 		COMPLETE_WITH("ADD ATTRIBUTE", "ADD VALUE", "ALTER ATTRIBUTE",
 					  "DROP ATTRIBUTE",
-					  "OWNER TO", "RENAME", "SET SCHEMA");
+					  "OWNER TO", "RENAME", "SET SCHEMA", "SET STORAGE");
 	/* complete ALTER TYPE <foo> ADD with actions */
 	else if (Matches("ALTER", "TYPE", MatchAny, "ADD"))
 		COMPLETE_WITH("ATTRIBUTE", "VALUE");
diff --git a/src/include/commands/alter.h b/src/include/commands/alter.h
index f9d6ba13c9..c55a1da0af 100644
--- a/src/include/commands/alter.h
+++ b/src/include/commands/alter.h
@@ -31,5 +31,6 @@ extern Oid	AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 extern ObjectAddress ExecAlterOwnerStmt(AlterOwnerStmt *stmt);
 extern void AlterObjectOwner_internal(Relation catalog, Oid objectId,
 									  Oid new_ownerId);
+extern ObjectAddress ExecAlterTypeStorageStmt(AlterTypeStorageStmt *stmt);
 
 #endif							/* ALTER_H */
diff --git a/src/include/commands/typecmds.h b/src/include/commands/typecmds.h
index fc18d64347..1f45fc670a 100644
--- a/src/include/commands/typecmds.h
+++ b/src/include/commands/typecmds.h
@@ -54,4 +54,6 @@ extern Oid	AlterTypeNamespaceInternal(Oid typeOid, Oid nspOid,
 									   bool errorOnTableType,
 									   ObjectAddresses *objsMoved);
 
+extern ObjectAddress AlterTypeStorage(List *names, char newStorage, ObjectType objecttype);
+
 #endif							/* TYPECMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index baced7eec0..2eb227efdc 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -380,6 +380,7 @@ typedef enum NodeTag
 	T_AlterObjectSchemaStmt,
 	T_AlterOwnerStmt,
 	T_AlterOperatorStmt,
+	T_AlterTypeStorageStmt,
 	T_DropOwnedStmt,
 	T_ReassignOwnedStmt,
 	T_CompositeTypeStmt,
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index da0706add5..4d40acc097 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2959,6 +2959,18 @@ typedef struct AlterOwnerStmt
 	RoleSpec   *newowner;		/* the new owner */
 } AlterOwnerStmt;
 
+/* ----------------------
+ *		Alter Type Storage Statement
+ * ----------------------
+ */
+typedef struct AlterTypeStorageStmt
+{
+	NodeTag		type;
+	ObjectType	objectType;		/* OBJECT_TYPE or OBJECT_DOMAIN */
+	Node	   *object;
+	Node	   *newstorage;		/* the new storage */
+} AlterTypeStorageStmt;
+
 
 /* ----------------------
  *		Alter Operator Set Restrict, Join