v12-0003-Add-support-for-PRESERVE.patch

From f593daa79daa902c9a805c5aa985923288704a5a Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilipkumar@localhost.localdomain>
Date: Tue, 20 Oct 2020 17:18:34 +0530
Subject: [PATCH v12 3/6] Add support for PRESERVE

Now the compression method can be changed without forcing a table
rewrite, by including the old method in the PRESERVE list.
---
 doc/src/sgml/ref/alter_table.sgml          |  11 +-
 src/backend/catalog/dependency.c           |   8 +-
 src/backend/catalog/objectaddress.c        |  24 +++
 src/backend/commands/Makefile              |   1 +
 src/backend/commands/alter.c               |   1 +
 src/backend/commands/compressioncmds.c     | 222 +++++++++++++++++++++
 src/backend/commands/event_trigger.c       |   1 +
 src/backend/commands/tablecmds.c           | 116 ++++++-----
 src/backend/executor/nodeModifyTable.c     |   7 +-
 src/backend/nodes/copyfuncs.c              |  17 +-
 src/backend/nodes/equalfuncs.c             |  15 +-
 src/backend/nodes/outfuncs.c               |  15 +-
 src/backend/parser/gram.y                  |  52 ++++-
 src/backend/parser/parse_utilcmd.c         |   2 +-
 src/include/catalog/dependency.h           |   5 +-
 src/include/commands/defrem.h              |   7 +
 src/include/nodes/nodes.h                  |   1 +
 src/include/nodes/parsenodes.h             |  16 +-
 src/test/regress/expected/create_cm.out    |   9 +
 src/test/regress/expected/create_index.out |   6 +-
 src/test/regress/sql/create_cm.sql         |   4 +
 21 files changed, 470 insertions(+), 70 deletions(-)
 create mode 100644 src/backend/commands/compressioncmds.c

diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index 5c88c979af..733ee0cf20 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -54,7 +54,7 @@ ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET ( <replaceable class="parameter">attribute_option</replaceable> = <replaceable class="parameter">value</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> RESET ( <replaceable class="parameter">attribute_option</replaceable> [, ... ] )
     ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET STORAGE { PLAIN | EXTERNAL | EXTENDED | MAIN }
-    ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable>
+    ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]
     ADD <replaceable class="parameter">table_constraint</replaceable> [ NOT VALID ]
     ADD <replaceable class="parameter">table_constraint_using_index</replaceable>
     ALTER CONSTRAINT <replaceable class="parameter">constraint_name</replaceable> [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ]
@@ -386,12 +386,17 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
 
    <varlistentry>
     <term>
-     <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable></literal>
+     <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]</literal>
     </term>
     <listitem>
      <para>
       This clause adds compression to a column. Compression method can be
-      set from available built-in compression methods.
+      set from available built-in compression methods. The PRESERVE list
+      contains list of compression methods used on the column and determines
+      which of them should be kept on the column. Without PRESERVE or if all
+      the previous compression methods are not preserved then the table will
+      be rewritten. If PRESERVE ALL is specified then all the previous methods
+      will be preserved and the table will not be rewritten.
      </para>
     </listitem>
    </varlistentry>
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index f515e2c308..6e86c66456 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -29,6 +29,7 @@
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
 #include "catalog/pg_collation.h"
+#include "catalog/pg_compression.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
@@ -180,7 +181,8 @@ static const Oid object_classes[] = {
 	PublicationRelationId,		/* OCLASS_PUBLICATION */
 	PublicationRelRelationId,	/* OCLASS_PUBLICATION_REL */
 	SubscriptionRelationId,		/* OCLASS_SUBSCRIPTION */
-	TransformRelationId			/* OCLASS_TRANSFORM */
+	TransformRelationId,		/* OCLASS_TRANSFORM */
+	CompressionRelationId		/* OCLASS_COMPRESSION */
 };
 
 
@@ -1506,6 +1508,7 @@ doDeletion(const ObjectAddress *object, int flags)
 		case OCLASS_DATABASE:
 		case OCLASS_TBLSPACE:
 		case OCLASS_SUBSCRIPTION:
+		case OCLASS_COMPRESSION:
 			elog(ERROR, "global objects cannot be deleted by doDeletion");
 			break;
 
@@ -2843,6 +2846,9 @@ getObjectClass(const ObjectAddress *object)
 
 		case TransformRelationId:
 			return OCLASS_TRANSFORM;
+
+		case CompressionRelationId:
+			return OCLASS_COMPRESSION;
 	}
 
 	/* shouldn't get here */
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 4815f6ca7e..ed40d78d8d 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -15,6 +15,7 @@
 
 #include "postgres.h"
 
+#include "access/compressionapi.h"
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/relation.h"
@@ -30,6 +31,7 @@
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
 #include "catalog/pg_collation.h"
+#include "catalog/pg_compression.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
@@ -3907,6 +3909,15 @@ getObjectDescription(const ObjectAddress *object, bool missing_ok)
 				break;
 			}
 
+		case OCLASS_COMPRESSION:
+			{
+				char *cmname = GetCompressionNameFromOid(object->objectId);
+
+				if (cmname)
+					appendStringInfo(&buffer, _("compression %s"), cmname);
+				break;
+			}
+
 		case OCLASS_TRANSFORM:
 			{
 				HeapTuple	trfTup;
@@ -4479,6 +4490,10 @@ getObjectTypeDescription(const ObjectAddress *object, bool missing_ok)
 			appendStringInfoString(&buffer, "transform");
 			break;
 
+		case OCLASS_COMPRESSION:
+			appendStringInfoString(&buffer, "compression");
+			break;
+
 			/*
 			 * There's intentionally no default: case here; we want the
 			 * compiler to warn if a new OCLASS hasn't been handled above.
@@ -5763,6 +5778,15 @@ getObjectIdentityParts(const ObjectAddress *object,
 			}
 			break;
 
+		case OCLASS_COMPRESSION:
+			{
+				appendStringInfo(&buffer, "%u",
+								 object->objectId);
+				if (objname)
+					*objname = list_make1(psprintf("%u", object->objectId));
+				break;
+			}
+
 			/*
 			 * There's intentionally no default: case here; we want the
 			 * compiler to warn if a new OCLASS hasn't been handled above.
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index d4815d3ce6..770df27b90 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -21,6 +21,7 @@ OBJS = \
 	cluster.o \
 	collationcmds.o \
 	comment.o \
+	compressioncmds.o \
 	constraint.o \
 	conversioncmds.o \
 	copy.o \
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index b11ebf0f61..8192429360 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -663,6 +663,7 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_COMPRESSION:
 			/* ignore object types that don't have schema-qualified names */
 			break;
 
diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c
new file mode 100644
index 0000000000..a1be6ef7be
--- /dev/null
+++ b/src/backend/commands/compressioncmds.c
@@ -0,0 +1,222 @@
+/*-------------------------------------------------------------------------
+ *
+ * compressioncmds.c
+ *	  Routines for SQL commands for attribute compression methods
+ *
+ * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/commands/compressioncmds.c
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/compressionapi.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/reloptions.h"
+#include "catalog/catalog.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_attribute.h"
+#include "catalog/pg_depend.h"
+#include "commands/defrem.h"
+#include "nodes/parsenodes.h"
+#include "utils/fmgroids.h"
+
+/*
+ * get list of all supported compression methods for the given attribute.
+ *
+ * If oldcmoids list is passed then it will delete the attribute dependency
+ * on the compression methods passed in the oldcmoids, otherwise it will
+ * return the list of all the compression method on which the attribute has
+ * dependency.
+ */
+static List *
+lookup_attribute_compression(Oid attrelid, AttrNumber attnum, List *oldcmoids)
+{
+	LOCKMODE	lock = AccessShareLock;
+	HeapTuple	tup;
+	Relation	rel;
+	SysScanDesc scan;
+	ScanKeyData key[3];
+	List	   *cmoids = NIL;
+
+	rel = table_open(DependRelationId, lock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_classid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationRelationId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_objid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(attrelid));
+	ScanKeyInit(&key[2],
+				Anum_pg_depend_objsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum((int32) attnum));
+
+	scan = systable_beginscan(rel, DependDependerIndexId, true,
+							  NULL, 3, key);
+
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+
+		if (depform->refclassid == CompressionRelationId)
+		{
+			if (oldcmoids && list_member_oid(oldcmoids, depform->refobjid))
+				CatalogTupleDelete(rel, &tup->t_self);
+			else if (oldcmoids == NULL)
+				cmoids = list_append_unique_oid(cmoids, depform->refobjid);
+		}
+	}
+
+	systable_endscan(scan);
+	table_close(rel, lock);
+
+	return cmoids;
+}
+
+/*
+ * Remove the attribute dependency on the old compression methods given in the
+ * cmoids list.
+ */
+static void
+remove_old_dependencies(Oid attrelid, AttrNumber attnum, List *cmoids)
+{
+	lookup_attribute_compression(attrelid, attnum, cmoids);
+}
+
+/*
+ * Check whether the given compression method oid is supported by
+ * the target attribue.
+ */
+bool
+IsCompressionSupported(Form_pg_attribute att, Oid cmoid)
+{
+	List	   *cmoids = NIL;
+
+	/* Check whether it is same as the current compression oid */
+	if (cmoid == att->attcompression)
+		return true;
+
+	/* Check the oid in all preserved compresion methods */
+	cmoids = lookup_attribute_compression(att->attrelid, att->attnum, NULL);
+	if (list_member_oid(cmoids, cmoid))
+		return true;
+	else
+		return false;
+}
+
+/*
+ * Get the compression method oid based on the compression method name.  When
+ * compression is not specified returns default attribute compression.  It is
+ * possible case for CREATE TABLE and ADD COLUMN commands where COMPRESSION
+ * syntax is optional.
+ *
+ * For ALTER command, check all the supported compression methods for the
+ * attribute and if the preserve list is not passed or some of the old
+ * compression methods are not given in the preserved list then delete
+ * dependency from the old compression methods and force the table rewrite.
+ */
+Oid
+GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression,
+						bool *need_rewrite)
+{
+	Oid			cmoid;
+	ListCell   *cell;
+
+	/* no compression for the plain storage */
+	if (att->attstorage == TYPSTORAGE_PLAIN)
+		return InvalidOid;
+
+	/* fallback to default compression if it's not specified */
+	if (compression == NULL)
+		return DefaultCompressionOid;
+
+	cmoid = GetCompressionOid(compression->cmname);
+	if (!OidIsValid(cmoid))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("compression type \"%s\" not recognized", compression->cmname)));
+
+	/*
+	 * Determine if the column needs rewrite or not. Rewrite conditions: SET
+	 * COMPRESSION without PRESERVE - SET COMPRESSION with PRESERVE but not
+	 * with full list of previous access methods.
+	 */
+	if (need_rewrite != NULL)
+	{
+		List	   *previous_cmoids = NIL;
+
+		*need_rewrite = false;
+
+		/* If we have preserved all then rewrite is not required */
+		if (compression->preserve_all)
+			return cmoid;
+
+		previous_cmoids =
+			lookup_attribute_compression(att->attrelid, att->attnum, NULL);
+
+		if (compression->preserve != NIL)
+		{
+			foreach(cell, compression->preserve)
+			{
+				char	   *cmname_p = strVal(lfirst(cell));
+				Oid			cmoid_p = GetCompressionOid(cmname_p);
+
+				if (!list_member_oid(previous_cmoids, cmoid_p))
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							 errmsg("\"%s\" compression access method cannot be preserved", cmname_p),
+							 errhint("use \"pg_column_compression\" function for list of compression methods")));
+
+				/*
+				 * Remove from previous list, also protect from multiple
+				 * mentions of one access method in PRESERVE list
+				 */
+				previous_cmoids = list_delete_oid(previous_cmoids, cmoid_p);
+			}
+		}
+
+		/* delete the current cmoid from the list */
+		previous_cmoids = list_delete_oid(previous_cmoids, cmoid);
+
+		/*
+		 * If the list of previous Oids is not empty after deletions then
+		 * we need to rewrite tuples in the table.
+		 */
+		if (list_length(previous_cmoids) != 0)
+		{
+			remove_old_dependencies(att->attrelid, att->attnum,
+									previous_cmoids);
+			*need_rewrite = true;
+		}
+
+		/* Cleanup */
+		list_free(previous_cmoids);
+	}
+
+	return cmoid;
+}
+
+/*
+ * Construct ColumnCompression node from the compression method oid.
+ */
+ColumnCompression *
+MakeColumnCompression(Oid attcompression)
+{
+	ColumnCompression *node;
+
+	if (!OidIsValid(attcompression))
+		return NULL;
+
+	node = makeNode(ColumnCompression);
+	node->cmname = GetCompressionNameFromOid(attcompression);
+
+	return node;
+}
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index 8bb17c34f5..26fe640e8e 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -1053,6 +1053,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_COMPRESSION:
 			return true;
 
 			/*
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 43f79242e8..80bdb79530 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -391,6 +391,7 @@ static bool check_for_column_name_collision(Relation rel, const char *colname,
 											bool if_not_exists);
 static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
 static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid);
+static void add_column_compression_dependency(Oid relid, int32 attnum, Oid cmoid);
 static void ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
 static ObjectAddress ATExecDropNotNull(Relation rel, const char *colName, LOCKMODE lockmode);
 static void ATPrepSetNotNull(List **wqueue, Relation rel,
@@ -531,7 +532,9 @@ static void ATExecEnableRowSecurity(Relation rel);
 static void ATExecDisableRowSecurity(Relation rel);
 static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls);
 static ObjectAddress ATExecSetCompression(AlteredTableInfo *tab, Relation rel,
-					 const char *column, Node *newValue, LOCKMODE lockmode);
+										  const char *column,
+										  ColumnCompression *compression,
+										  LOCKMODE lockmode);
 
 static void index_copy_data(Relation rel, RelFileNode newrnode);
 static const char *storage_name(char c);
@@ -562,8 +565,6 @@ static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx,
 								  Relation partitionTbl);
 static List *GetParentedForeignKeyRefs(Relation partition);
 static void ATDetachCheckNoForeignKeyRefs(Relation partition);
-static Oid GetAttributeCompressionMethod(Form_pg_attribute att,
-										 char *compression);
 
 /* ----------------------------------------------------------------
  *		DefineRelation
@@ -588,6 +589,7 @@ ObjectAddress
 DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			   ObjectAddress *typaddress, const char *queryString)
 {
+	int			i;
 	char		relname[NAMEDATALEN];
 	Oid			namespaceId;
 	Oid			relationId;
@@ -866,7 +868,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			(relkind == RELKIND_RELATION ||
 			 relkind == RELKIND_PARTITIONED_TABLE))
 			attr->attcompression =
-					GetAttributeCompressionMethod(attr, colDef->compression);
+					GetAttributeCompression(attr, colDef->compression, NULL);
 		else
 			attr->attcompression = InvalidOid;
 	}
@@ -936,6 +938,20 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	 */
 	rel = relation_open(relationId, AccessExclusiveLock);
 
+	/*
+	 * Specify relation Oid in attribute compression tuples, and create
+	 * dependencies.
+	 */
+	for (i = 0; i < (RelationGetDescr(rel))->natts; i++)
+	{
+		Form_pg_attribute attr;
+
+		attr = TupleDescAttr(RelationGetDescr(rel), i);
+		if (OidIsValid(attr->attcompression))
+			add_column_compression_dependency(attr->attrelid, attr->attnum,
+											  attr->attcompression);
+	}
+
 	/*
 	 * Now add any newly specified column default and generation expressions
 	 * to the new relation.  These are passed to us in the form of raw
@@ -2414,17 +2430,17 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				/* Copy/check compression parameter */
 				if (OidIsValid(attribute->attcompression))
 				{
-					char *compression =
-						GetCompressionNameFromOid(attribute->attcompression);
+					ColumnCompression *compression =
+							MakeColumnCompression(attribute->attcompression);
 
 					if (!def->compression)
 						def->compression = compression;
-					else if (strcmp(def->compression, compression))
+					else if (strcmp(def->compression->cmname, compression->cmname))
 						ereport(ERROR,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
 								 errmsg("column \"%s\" has a compression method conflict",
 										attributeName),
-								 errdetail("%s versus %s", def->compression, compression)));
+								 errdetail("%s versus %s", def->compression->cmname, compression->cmname)));
 				}
 
 				def->inhcount++;
@@ -2461,7 +2477,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->collOid = attribute->attcollation;
 				def->constraints = NIL;
 				def->location = -1;
-				def->compression = GetCompressionNameFromOid(attribute->attcompression);
+				def->compression = MakeColumnCompression(attribute->attcompression);
 				inhSchema = lappend(inhSchema, def);
 				newattmap->attnums[parent_attno - 1] = ++child_attno;
 			}
@@ -2712,12 +2728,12 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 					def->compression = newdef->compression;
 				else if (newdef->compression)
 				{
-					if (strcmp(def->compression, newdef->compression))
+					if (strcmp(def->compression->cmname, newdef->compression->cmname))
 						ereport(ERROR,
 							(errcode(ERRCODE_DATATYPE_MISMATCH),
 								errmsg("column \"%s\" has a compression method conflict",
 									attributeName),
-								errdetail("%s versus %s", def->compression, newdef->compression)));
+								errdetail("%s versus %s", def->compression->cmname, newdef->compression->cmname)));
 				}
 
 				/* Mark the column as locally defined */
@@ -4787,7 +4803,8 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
 			ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name);
 			break;
 		case AT_SetCompression:
-			address = ATExecSetCompression(tab, rel, cmd->name, cmd->def,
+			address = ATExecSetCompression(tab, rel, cmd->name,
+										   (ColumnCompression *) cmd->def,
 										   lockmode);
 			break;
 		default:				/* oops */
@@ -6280,8 +6297,8 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	 */
 	if (rel->rd_rel->relkind == RELKIND_RELATION ||
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-		attribute.attcompression = GetAttributeCompressionMethod(&attribute,
-																 colDef->compression);
+		attribute.attcompression = GetAttributeCompression(&attribute,
+										colDef->compression, NULL);
 	else
 		attribute.attcompression = InvalidOid;
 
@@ -6456,6 +6473,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	 */
 	add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid);
 	add_column_collation_dependency(myrelid, newattnum, attribute.attcollation);
+	add_column_compression_dependency(myrelid, newattnum, attribute.attcompression);
 
 	/*
 	 * Propagate to children as appropriate.  Unlike most other ALTER
@@ -6603,6 +6621,28 @@ add_column_collation_dependency(Oid relid, int32 attnum, Oid collid)
 	}
 }
 
+/*
+ * Install a dependency for compression on its column.
+ *
+ * This is used to determine connection between column and builtin
+ * compression in ALTER SET COMPRESSION command.
+ *
+ * If dependency is already there the whole thing is skipped.
+ */
+static void
+add_column_compression_dependency(Oid relid, int32 attnum, Oid cmoid)
+{
+	ObjectAddress acref,
+		attref;
+
+	Assert(relid > 0 && attnum > 0);
+
+	ObjectAddressSet(acref, CompressionRelationId, cmoid);
+	ObjectAddressSubSet(attref, RelationRelationId, relid, attnum);
+
+	recordDependencyOn(&attref, &acref, DEPENDENCY_NORMAL);
+}
+
 /*
  * ALTER TABLE ALTER COLUMN DROP NOT NULL
  */
@@ -11591,7 +11631,11 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 								   getObjectDescription(&foundObject, false),
 								   colName)));
 				break;
+			case OCLASS_COMPRESSION:
 
+				/* Just check that dependency is the right type */
+				Assert(foundDep->deptype == DEPENDENCY_NORMAL);
+				break;
 			case OCLASS_DEFAULT:
 
 				/*
@@ -11702,7 +11746,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 			  foundDep->refobjid == attTup->attcollation) &&
 			!(foundDep->refclassid == RelationRelationId &&
 			  foundDep->refobjid == RelationGetRelid(rel) &&
-			  foundDep->refobjsubid != 0)
+			  foundDep->refobjsubid != 0) &&
+			  foundDep->refclassid != CompressionRelationId
 			)
 			elog(ERROR, "found unexpected dependency for column: %s",
 				 getObjectDescription(&foundObject, false));
@@ -11814,6 +11859,11 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 	add_column_datatype_dependency(RelationGetRelid(rel), attnum, targettype);
 	add_column_collation_dependency(RelationGetRelid(rel), attnum, targetcollid);
 
+	/* Create dependency for new attribute compression */
+	if (OidIsValid(attTup->attcompression))
+		add_column_compression_dependency(RelationGetRelid(rel), attnum,
+										  attTup->attcompression);
+
 	/*
 	 * Drop any pg_statistic entry for the column, since it's now wrong type
 	 */
@@ -14987,24 +15037,21 @@ static ObjectAddress
 ATExecSetCompression(AlteredTableInfo *tab,
 					 Relation rel,
 					 const char *column,
-					 Node *newValue,
+					 ColumnCompression *compression,
 					 LOCKMODE lockmode)
 {
 	Relation	attrel;
 	HeapTuple	atttuple;
 	Form_pg_attribute atttableform;
 	AttrNumber	attnum;
-	char	   *compression;
 	Oid			cmoid;
+	bool		need_rewrite;
 	Datum		values[Natts_pg_attribute];
 	bool		nulls[Natts_pg_attribute];
 	bool		replace[Natts_pg_attribute];
 	ObjectAddress address;
 	ListCell   *lc;
 
-	Assert(IsA(newValue, String));
-	compression = strVal(newValue);
-
 	attrel = table_open(AttributeRelationId, RowExclusiveLock);
 
 	atttuple = SearchSysCacheAttName(RelationGetRelid(rel), column);
@@ -15035,9 +15082,12 @@ ATExecSetCompression(AlteredTableInfo *tab,
 	memset(replace, false, sizeof(replace));
 
 	/* Get the attribute compression method. */
-	cmoid = GetAttributeCompressionMethod(atttableform, compression);
+	cmoid = GetAttributeCompression(atttableform, compression, &need_rewrite);
 
 	if (atttableform->attcompression != cmoid)
+		add_column_compression_dependency(atttableform->attrelid,
+										  atttableform->attnum, cmoid);
+	if (need_rewrite)
 		tab->rewrite |= AT_REWRITE_ALTER_COMPRESSION;
 
 	atttableform->attcompression = cmoid;
@@ -17790,27 +17840,3 @@ ATDetachCheckNoForeignKeyRefs(Relation partition)
 		table_close(rel, NoLock);
 	}
 }
-
-/*
- * Get compression method for the attribute from compression name.
- */
-static Oid
-GetAttributeCompressionMethod(Form_pg_attribute att, char *compression)
-{
-	Oid cmoid;
-
-	/* no compression for the plain storage */
-	if (att->attstorage == TYPSTORAGE_PLAIN)
-		return InvalidOid;
-
-	/* fallback to default compression if it's not specified */
-	if (compression == NULL)
-		return DefaultCompressionOid;
-
-	cmoid = GetCompressionOid(compression);
-	if (!OidIsValid(cmoid))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("compression type \"%s\" not recognized", compression)));
-	return cmoid;
-}
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 0fb5a1bbe5..5257aeb66b 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -45,6 +45,7 @@
 #include "access/toast_internals.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
+#include "commands/defrem.h"
 #include "commands/trigger.h"
 #include "executor/execPartition.h"
 #include "executor/executor.h"
@@ -1936,6 +1937,7 @@ CompareCompressionMethodAndDecompress(TupleTableSlot *slot,
 	int			natts = slot->tts_tupleDescriptor->natts;
 	bool		isnull = false;
 	bool		decompressed_any = false;
+	Oid			cmoid = InvalidOid;
 	TupleDesc	tupleDesc = slot->tts_tupleDescriptor;
 
 	if (natts == 0)
@@ -1968,8 +1970,9 @@ CompareCompressionMethodAndDecompress(TupleTableSlot *slot,
 			 * Get the compression method stored in the toast header and
 			 * compare with the compression method of the target.
 			 */
-			if (targetTupDesc->attrs[i].attcompression !=
-				GetCompressionOidFromCompressionId(TOAST_COMPRESS_METHOD(new_value)))
+			cmoid = GetCompressionOidFromCompressionId(
+									TOAST_COMPRESS_METHOD(new_value));
+			if (!IsCompressionSupported(&targetTupDesc->attrs[i], cmoid))
 			{
 				new_value = detoast_attr(new_value);
 				slot->tts_values[attnum - 1] = PointerGetDatum(new_value);
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 1791f3dc10..2d1f830d9d 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2928,7 +2928,7 @@ _copyColumnDef(const ColumnDef *from)
 
 	COPY_STRING_FIELD(colname);
 	COPY_NODE_FIELD(typeName);
-	COPY_STRING_FIELD(compression);
+	COPY_NODE_FIELD(compression);
 	COPY_SCALAR_FIELD(inhcount);
 	COPY_SCALAR_FIELD(is_local);
 	COPY_SCALAR_FIELD(is_not_null);
@@ -2948,6 +2948,18 @@ _copyColumnDef(const ColumnDef *from)
 	return newnode;
 }
 
+static ColumnCompression *
+_copyColumnCompression(const ColumnCompression *from)
+{
+	ColumnCompression *newnode = makeNode(ColumnCompression);
+
+	COPY_STRING_FIELD(cmname);
+	COPY_SCALAR_FIELD(preserve_all);
+	COPY_NODE_FIELD(preserve);
+
+	return newnode;
+}
+
 static Constraint *
 _copyConstraint(const Constraint *from)
 {
@@ -5629,6 +5641,9 @@ copyObjectImpl(const void *from)
 		case T_ColumnDef:
 			retval = _copyColumnDef(from);
 			break;
+		case T_ColumnCompression:
+			retval = _copyColumnCompression(from);
+			break;
 		case T_Constraint:
 			retval = _copyConstraint(from);
 			break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index ab893ec26a..365957cfe7 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2588,7 +2588,7 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 {
 	COMPARE_STRING_FIELD(colname);
 	COMPARE_NODE_FIELD(typeName);
-	COMPARE_STRING_FIELD(compression);
+	COMPARE_NODE_FIELD(compression);
 	COMPARE_SCALAR_FIELD(inhcount);
 	COMPARE_SCALAR_FIELD(is_local);
 	COMPARE_SCALAR_FIELD(is_not_null);
@@ -2608,6 +2608,16 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 	return true;
 }
 
+static bool
+_equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b)
+{
+	COMPARE_STRING_FIELD(cmname);
+	COMPARE_SCALAR_FIELD(preserve_all);
+	COMPARE_NODE_FIELD(preserve);
+
+	return true;
+}
+
 static bool
 _equalConstraint(const Constraint *a, const Constraint *b)
 {
@@ -3683,6 +3693,9 @@ equal(const void *a, const void *b)
 		case T_ColumnDef:
 			retval = _equalColumnDef(a, b);
 			break;
+		case T_ColumnCompression:
+			retval = _equalColumnCompression(a, b);
+			break;
 		case T_Constraint:
 			retval = _equalConstraint(a, b);
 			break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index b97a87c413..2ef640b3b7 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2832,7 +2832,7 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
 
 	WRITE_STRING_FIELD(colname);
 	WRITE_NODE_FIELD(typeName);
-	WRITE_STRING_FIELD(compression);
+	WRITE_NODE_FIELD(compression);
 	WRITE_INT_FIELD(inhcount);
 	WRITE_BOOL_FIELD(is_local);
 	WRITE_BOOL_FIELD(is_not_null);
@@ -2850,6 +2850,16 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
 	WRITE_LOCATION_FIELD(location);
 }
 
+static void
+_outColumnCompression(StringInfo str, const ColumnCompression *node)
+{
+	WRITE_NODE_TYPE("COLUMNCOMPRESSION");
+
+	WRITE_STRING_FIELD(cmname);
+	WRITE_BOOL_FIELD(preserve_all);
+	WRITE_NODE_FIELD(preserve);
+}
+
 static void
 _outTypeName(StringInfo str, const TypeName *node)
 {
@@ -4198,6 +4208,9 @@ outNode(StringInfo str, const void *obj)
 			case T_ColumnDef:
 				_outColumnDef(str, obj);
 				break;
+			case T_ColumnCompression:
+				_outColumnCompression(str, obj);
+				break;
 			case T_TypeName:
 				_outTypeName(str, obj);
 				break;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 289c8c97ee..a1755c6fc7 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -601,7 +601,9 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <list>		hash_partbound
 %type <defelt>		hash_partbound_elem
 
-%type <str>	optColumnCompression
+%type <node>	optColumnCompression alterColumnCompression
+%type <str>		compressionClause
+%type <list>	optCompressionPreserve
 
 /*
  * Non-keyword token types.  These are hard-wired into the "flex" lexer.
@@ -2267,12 +2269,12 @@ alter_table_cmd:
 					$$ = (Node *)n;
 				}
 			/* ALTER TABLE <name> ALTER [COLUMN] <colname> SET (COMPRESSION <cm>) */
-			| ALTER opt_column ColId SET optColumnCompression
+			| ALTER opt_column ColId SET alterColumnCompression
 				{
 					AlterTableCmd *n = makeNode(AlterTableCmd);
 					n->subtype = AT_SetCompression;
 					n->name = $3;
-					n->def = (Node *) makeString($5);;
+					n->def = $5;
 					$$ = (Node *)n;
 				}
 			/* ALTER TABLE <name> DROP [COLUMN] IF EXISTS <colname> [RESTRICT|CASCADE] */
@@ -3387,7 +3389,7 @@ columnDef:	ColId Typename optColumnCompression create_generic_options ColQualLis
 					ColumnDef *n = makeNode(ColumnDef);
 					n->colname = $1;
 					n->typeName = $2;
-					n->compression = $3;
+					n->compression = (ColumnCompression *) $3;
 					n->inhcount = 0;
 					n->is_local = true;
 					n->is_not_null = false;
@@ -3442,13 +3444,43 @@ columnOptions:	ColId ColQualList
 				}
 		;
 
+optCompressionPreserve:
+			PRESERVE '(' name_list ')' { $$ = $3; }
+			| /*EMPTY*/ { $$ = NULL; }
+		;
+
+compressionClause:
+			COMPRESSION name { $$ = pstrdup($2); }
+		;
+
 optColumnCompression:
-					COMPRESSION name
-					{
-						$$ = $2;
-					}
-					| /*EMPTY*/	{ $$ = NULL; }
-				;
+			compressionClause
+				{
+					ColumnCompression *n = makeNode(ColumnCompression);
+					n->cmname = $1;
+					n->preserve = NIL;
+					$$ = (Node *) n;
+				}
+			| /*EMPTY*/	{ $$ = NULL; }
+		;
+
+alterColumnCompression:
+			compressionClause optCompressionPreserve
+				{
+					ColumnCompression *n = makeNode(ColumnCompression);
+					n->cmname = $1;
+					n->preserve = (List *) $2;
+					$$ = (Node *) n;
+				}
+			|	compressionClause PRESERVE ALL
+				{
+					ColumnCompression *n = makeNode(ColumnCompression);
+					n->cmname = $1;
+					n->preserve_all = true;
+					n->preserve = NIL;
+					$$ = (Node *) n;
+				}
+		;
 
 ColQualList:
 			ColQualList ColConstraint				{ $$ = lappend($1, $2); }
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 4f04c128df..db7d381b53 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1069,7 +1069,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		/* Likewise, copy compression if requested */
 		if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION
 			&& OidIsValid(attribute->attcompression))
-			def->compression = GetCompressionNameFromOid(attribute->attcompression);
+			def->compression = MakeColumnCompression(attribute->attcompression);
 		else
 			def->compression = NULL;
 
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index a8f7e9965b..bd5a2e3834 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -125,10 +125,11 @@ typedef enum ObjectClass
 	OCLASS_PUBLICATION,			/* pg_publication */
 	OCLASS_PUBLICATION_REL,		/* pg_publication_rel */
 	OCLASS_SUBSCRIPTION,		/* pg_subscription */
-	OCLASS_TRANSFORM			/* pg_transform */
+	OCLASS_TRANSFORM,			/* pg_transform */
+	OCLASS_COMPRESSION			/* pg_compression */
 } ObjectClass;
 
-#define LAST_OCLASS		OCLASS_TRANSFORM
+#define LAST_OCLASS		OCLASS_COMPRESSION
 
 /* flag bits for performDeletion/performMultipleDeletions: */
 #define PERFORM_DELETION_INTERNAL			0x0001	/* internal action */
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 7a079ef07f..936b072c76 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -145,6 +145,13 @@ extern Oid	get_table_am_oid(const char *amname, bool missing_ok);
 extern Oid	get_am_oid(const char *amname, bool missing_ok);
 extern char *get_am_name(Oid amOid);
 
+/* commands/compressioncmds.c */
+extern Oid GetAttributeCompression(Form_pg_attribute att,
+								   ColumnCompression *compression,
+								   bool *need_rewrite);
+extern ColumnCompression *MakeColumnCompression(Oid atttcompression);
+extern bool IsCompressionSupported(Form_pg_attribute att, Oid cmoid);
+
 /* support routines in commands/define.c */
 
 extern char *defGetString(DefElem *def);
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 5a1ca9fada..65ef987d67 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -479,6 +479,7 @@ typedef enum NodeTag
 	T_PartitionBoundSpec,
 	T_PartitionRangeDatum,
 	T_PartitionCmd,
+	T_ColumnCompression,
 	T_VacuumRelation,
 
 	/*
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 4656bb7e58..3c1ee794b5 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -624,6 +624,20 @@ typedef struct RangeTableSample
 	int			location;		/* method name location, or -1 if unknown */
 } RangeTableSample;
 
+/*
+ * ColumnCompression - compression parameters for some attribute
+ *
+ * This represents compression information defined using clause:
+ * .. COMPRESSION <compression method> PRESERVE <compression methods>
+ */
+typedef struct ColumnCompression
+{
+	NodeTag		type;
+	char	   *cmname;
+	bool		preserve_all;
+	List	   *preserve;
+} ColumnCompression;
+
 /*
  * ColumnDef - column definition (used in various creates)
  *
@@ -647,7 +661,7 @@ typedef struct ColumnDef
 	NodeTag		type;
 	char	   *colname;		/* name of column */
 	TypeName   *typeName;		/* type of column */
-	char	   *compression;	/* compression method for column */
+	ColumnCompression *compression;	/* column compression */
 	int			inhcount;		/* number of times column is inherited */
 	bool		is_local;		/* column has local (non-inherited) def'n */
 	bool		is_not_null;	/* NOT NULL constraint specified? */
diff --git a/src/test/regress/expected/create_cm.out b/src/test/regress/expected/create_cm.out
index 3de5fab8c9..1b7b5bd6e5 100644
--- a/src/test/regress/expected/create_cm.out
+++ b/src/test/regress/expected/create_cm.out
@@ -94,4 +94,13 @@ ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION lz4;
 --------+------+-----------+----------+---------+----------+-------------+--------------+-------------
  f1     | text |           |          |         | extended | lz4         |              | 
 
+-- preserve old compression method
+ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (lz4);
+INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
+\d+ cmmove2
+                                        Table "public.cmmove2"
+ Column | Type | Collation | Nullable | Default | Storage  | Compression | Stats target | Description 
+--------+------+-----------+----------+---------+----------+-------------+--------------+-------------
+ f1     | text |           |          |         | extended | pglz        |              | 
+
 DROP TABLE cmmove1, cmmove2, cmmove3, lz4test;
diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out
index 6ace7662ee..a5abdcaf85 100644
--- a/src/test/regress/expected/create_index.out
+++ b/src/test/regress/expected/create_index.out
@@ -2064,6 +2064,7 @@ WHERE classid = 'pg_class'::regclass AND
   ORDER BY 1, 2;
                    obj                    |                           objref                           | deptype 
 ------------------------------------------+------------------------------------------------------------+---------
+ column c2 of table concur_reindex_tab    | compression pglz                                           | n
  index concur_reindex_ind1                | constraint concur_reindex_ind1 on table concur_reindex_tab | i
  index concur_reindex_ind2                | column c2 of table concur_reindex_tab                      | a
  index concur_reindex_ind3                | column c1 of table concur_reindex_tab                      | a
@@ -2072,7 +2073,7 @@ WHERE classid = 'pg_class'::regclass AND
  index concur_reindex_ind4                | column c2 of table concur_reindex_tab                      | a
  materialized view concur_reindex_matview | schema public                                              | n
  table concur_reindex_tab                 | schema public                                              | n
-(8 rows)
+(9 rows)
 
 REINDEX INDEX CONCURRENTLY concur_reindex_ind1;
 REINDEX TABLE CONCURRENTLY concur_reindex_tab;
@@ -2091,6 +2092,7 @@ WHERE classid = 'pg_class'::regclass AND
   ORDER BY 1, 2;
                    obj                    |                           objref                           | deptype 
 ------------------------------------------+------------------------------------------------------------+---------
+ column c2 of table concur_reindex_tab    | compression pglz                                           | n
  index concur_reindex_ind1                | constraint concur_reindex_ind1 on table concur_reindex_tab | i
  index concur_reindex_ind2                | column c2 of table concur_reindex_tab                      | a
  index concur_reindex_ind3                | column c1 of table concur_reindex_tab                      | a
@@ -2099,7 +2101,7 @@ WHERE classid = 'pg_class'::regclass AND
  index concur_reindex_ind4                | column c2 of table concur_reindex_tab                      | a
  materialized view concur_reindex_matview | schema public                                              | n
  table concur_reindex_tab                 | schema public                                              | n
-(8 rows)
+(9 rows)
 
 -- Check that comments are preserved
 CREATE TABLE testcomment (i int);
diff --git a/src/test/regress/sql/create_cm.sql b/src/test/regress/sql/create_cm.sql
index 8b75a0640d..3549fb35f0 100644
--- a/src/test/regress/sql/create_cm.sql
+++ b/src/test/regress/sql/create_cm.sql
@@ -46,5 +46,9 @@ ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION lz4;
 ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION lz4;
 \d+ cmmove2
 
+-- preserve old compression method
+ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (lz4);
+INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
+\d+ cmmove2
 
 DROP TABLE cmmove1, cmmove2, cmmove3, lz4test;
-- 
2.23.0