From 1b65367daa02d81926ff9c18a609265a0d66d14b Mon Sep 17 00:00:00 2001
From: dilipkumar <dilipbalaut@gmail.com>
Date: Sun, 27 Sep 2020 14:32:13 +0530
Subject: [PATCH v3 3/3] 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       |   7 +-
 src/backend/access/common/tupdesc.c     |   8 +-
 src/backend/access/table/toast_helper.c |   4 +-
 src/backend/bootstrap/bootstrap.c       |   7 +-
 src/backend/catalog/heap.c              |   4 +-
 src/backend/commands/Makefile           |   1 +
 src/backend/commands/compressioncmds.c  | 138 ++++++++++++++++++++++++
 src/backend/commands/tablecmds.c        | 120 +++++++++------------
 src/backend/executor/nodeModifyTable.c  |  11 +-
 src/backend/nodes/copyfuncs.c           |  16 ++-
 src/backend/nodes/equalfuncs.c          |  14 ++-
 src/backend/nodes/outfuncs.c            |  14 ++-
 src/backend/parser/gram.y               |  44 ++++++--
 src/backend/parser/parse_utilcmd.c      |   6 +-
 src/include/catalog/pg_attribute.h      |  10 +-
 src/include/commands/defrem.h           |   7 ++
 src/include/nodes/nodes.h               |   1 +
 src/include/nodes/parsenodes.h          |  15 ++-
 src/test/regress/expected/create_cm.out |   9 ++
 src/test/regress/sql/create_cm.sql      |   4 +
 20 files changed, 334 insertions(+), 106 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..152fb8de93 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>) ]
     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 ]
@@ -391,7 +391,10 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
     <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 partial
+      list of compression methods the table will be rewritten.
      </para>
     </listitem>
    </varlistentry>
diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c
index ad39fd81a8..aa5b18c0c9 100644
--- a/src/backend/access/common/tupdesc.c
+++ b/src/backend/access/common/tupdesc.c
@@ -24,6 +24,7 @@
 #include "access/tupdesc_details.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_type.h"
+#include "commands/defrem.h"
 #include "common/hashfn.h"
 #include "miscadmin.h"
 #include "parser/parse_type.h"
@@ -471,7 +472,7 @@ equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2)
 			return false;
 		if (attr1->attcollation != attr2->attcollation)
 			return false;
-		if (attr1->attcompression != attr2->attcompression)
+		if (strcmp(NameStr(attr1->attcompression), NameStr(attr2->attcompression)) != 0)
 			return false;
 		/* attacl, attoptions and attfdwoptions are not even present... */
 	}
@@ -667,8 +668,9 @@ TupleDescInitEntry(TupleDesc desc,
 	att->attalign = typeForm->typalign;
 	att->attstorage = typeForm->typstorage;
 	att->attcollation = typeForm->typcollation;
-	att->attcompression = (att->attstorage == TYPSTORAGE_PLAIN) ?
-		InvalidCompressionMethod : DefaultCompressionMethod;
+
+	/* initialize the attribute compression field based on the storage type */
+	InitAttributeCompression(att);
 
 	ReleaseSysCache(tuple);
 }
diff --git a/src/backend/access/table/toast_helper.c b/src/backend/access/table/toast_helper.c
index b33c925a4b..829ebb29d0 100644
--- a/src/backend/access/table/toast_helper.c
+++ b/src/backend/access/table/toast_helper.c
@@ -54,7 +54,9 @@ toast_tuple_init(ToastTupleContext *ttc)
 
 		ttc->ttc_attr[i].tai_colflags = 0;
 		ttc->ttc_attr[i].tai_oldexternal = NULL;
-		ttc->ttc_attr[i].tai_compression = att->attcompression;
+
+		/* Current compression method is stored in the first character */
+		ttc->ttc_attr[i].tai_compression = *(NameStr(att->attcompression));
 
 		if (ttc->ttc_oldvalues != NULL)
 		{
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 909705d86b..079921eda0 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -28,6 +28,7 @@
 #include "catalog/index.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_type.h"
+#include "commands/defrem.h"
 #include "common/link-canary.h"
 #include "libpq/pqsignal.h"
 #include "miscadmin.h"
@@ -732,9 +733,9 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
 	attrtypes[attnum]->attcacheoff = -1;
 	attrtypes[attnum]->atttypmod = -1;
 	attrtypes[attnum]->attislocal = true;
-	attrtypes[attnum]->attcompression =
-		(attrtypes[attnum]->attstorage == TYPSTORAGE_PLAIN) ?
-		InvalidCompressionMethod : DefaultCompressionMethod;
+
+	/* initialize the attribute compression field based on the storage type */
+	InitAttributeCompression(attrtypes[attnum]);
 
 	if (nullness == BOOTCOL_NULL_FORCE_NOT_NULL)
 	{
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index d9d3012e65..6080ba74ef 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -781,7 +781,7 @@ InsertPgAttributeTuples(Relation pg_attribute_rel,
 		slot[slotCount]->tts_values[Anum_pg_attribute_attislocal - 1] = BoolGetDatum(attrs->attislocal);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attinhcount - 1] = Int32GetDatum(attrs->attinhcount);
 		slot[slotCount]->tts_values[Anum_pg_attribute_attcollation - 1] = ObjectIdGetDatum(attrs->attcollation);
-		slot[slotCount]->tts_values[Anum_pg_attribute_attcompression - 1] = CharGetDatum(attrs->attcompression);
+		slot[slotCount]->tts_values[Anum_pg_attribute_attcompression - 1] = NameGetDatum(&attrs->attcompression);
 		if (attoptions && attoptions[natts] != (Datum) 0)
 			slot[slotCount]->tts_values[Anum_pg_attribute_attoptions - 1] = attoptions[natts];
 		else
@@ -1706,7 +1706,7 @@ RemoveAttributeById(Oid relid, AttrNumber attnum)
 		/* Unset this so no one tries to look up the generation expression */
 		attStruct->attgenerated = '\0';
 
-		attStruct->attcompression = InvalidCompressionMethod;
+		MemSet(NameStr(attStruct->attcompression), 0, NAMEDATALEN);
 
 		/*
 		 * Change the column name to something that isn't likely to conflict
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/compressioncmds.c b/src/backend/commands/compressioncmds.c
new file mode 100644
index 0000000000..2c0f38673b
--- /dev/null
+++ b/src/backend/commands/compressioncmds.c
@@ -0,0 +1,138 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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/cmapi.h"
+#include "commands/defrem.h"
+
+
+/*
+ * CheckCompressionMethodsPreserved - Compare new compression method list with
+ * 									  the existing list.
+ *
+ * Check whether all the previous compression methods are preserved in the new
+ * compression methods or not.
+ */
+static bool
+CheckCompressionMethodsPreserved(char *previous_cm, char *new_cm)
+{
+	int i = 0;
+
+	while (previous_cm[i] != 0)
+	{
+		if (strchr(new_cm, previous_cm[i]) == NULL)
+			return false;
+		i++;
+	}
+
+	return true;
+}
+
+/*
+ * GetAttributeCompression - Get compression for given attribute
+ */
+char *
+GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression,
+						bool *need_rewrite)
+{
+	ListCell   *cell;
+	char	   *cm;
+	int			preserve_idx = 0;
+
+	cm = palloc0(NAMEDATALEN);
+
+	/* no compression for the plain storage */
+	if (att->attstorage == TYPSTORAGE_PLAIN)
+		return NULL;
+
+	/* fallback to default compression if it's not specified */
+	if (compression == NULL)
+	{
+		cm[preserve_idx] = DefaultCompressionMethod;
+		return cm;
+	}
+
+	/* current compression method */
+	cm[preserve_idx] = GetCompressionMethod(compression->cmname);
+	if (!IsValidCompression(cm[preserve_idx]))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("compression type \"%s\" not recognized", compression->cmname)));
+	preserve_idx++;
+
+	foreach (cell, compression->preserve)
+	{
+		char *cmname_p = strVal(lfirst(cell));
+		char cm_p = GetCompressionMethod(cmname_p);
+
+		/*
+		 * The compression method given in the preserve list must present in
+		 * the existing compression methods for the attribute.
+		 */
+		if (strchr(NameStr(att->attcompression), cm_p) == NULL)
+			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")));
+		cm[preserve_idx++] = cm_p;
+	}
+
+	/*
+	 * If all the previous compression methods are preserved then we don't need
+	 * to rewrite the table otherwise we need to.
+	 */
+	if (need_rewrite)
+	{
+		if (CheckCompressionMethodsPreserved(NameStr(att->attcompression), cm))
+			*need_rewrite = false;
+		else
+			*need_rewrite = true;
+	}
+
+	return cm;
+}
+
+/*
+ * InitAttributeCompression - initialize compression method in pg_attribute
+ *
+ * If attribute storage is plain then initialize with the invalid compression
+ * otherwise initialize it with the default compression method.
+ */
+void
+InitAttributeCompression(Form_pg_attribute att)
+{
+	char *attcompression = NameStr(att->attcompression);
+
+	MemSet(NameStr(att->attcompression), 0, NAMEDATALEN);
+	if (att->attstorage != TYPSTORAGE_PLAIN)
+		attcompression[0] = DefaultCompressionMethod;
+}
+
+/*
+ * MakeColumnCompression - Construct ColumnCompression node.
+ */
+ColumnCompression *
+MakeColumnCompression(NameData *compression)
+{
+	ColumnCompression *node;
+	char	cm = *(NameStr(*compression));
+
+	if (!IsValidCompression(cm))
+		return NULL;
+
+	node = makeNode(ColumnCompression);
+	node->cmname = GetCompressionName(cm);
+
+	return node;
+}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 6c35211945..1574e9fa35 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -531,7 +531,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 +564,6 @@ static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx,
 								  Relation partitionTbl);
 static List *GetParentedForeignKeyRefs(Relation partition);
 static void ATDetachCheckNoForeignKeyRefs(Relation partition);
-static char GetAttributeCompressionMethod(Form_pg_attribute att,
-										  char *compression);
 
 /* ----------------------------------------------------------------
  *		DefineRelation
@@ -862,10 +862,16 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		if (!IsBinaryUpgrade &&
 			(relkind == RELKIND_RELATION ||
 			 relkind == RELKIND_PARTITIONED_TABLE))
-			attr->attcompression =
-					GetAttributeCompressionMethod(attr, colDef->compression);
+		{
+			char *cm = GetAttributeCompression(attr, colDef->compression, NULL);
+
+			if (cm == NULL)
+				MemSet(NameStr(attr->attcompression), 0, NAMEDATALEN);
+			else
+				namestrcpy(&attr->attcompression, cm);
+		}
 		else
-			attr->attcompression = InvalidCompressionMethod;
+			MemSet(NameStr(attr->attcompression), 0, NAMEDATALEN);
 	}
 
 	/*
@@ -2406,19 +2412,19 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 									   storage_name(attribute->attstorage))));
 
 				/* Copy/check compression parameter */
-				if (IsValidCompression(attribute->attcompression))
+				if (IsValidCompression(*(NameStr(attribute->attcompression))))
 				{
-					char *compression =
-						GetCompressionName(attribute->attcompression);
+					ColumnCompression *compression =
+						MakeColumnCompression(&attribute->attcompression);
 
-					if (!def->compression)
+					if (def->compression == NULL)
 						def->compression = compression;
-					else if (strcmp(def->compression, compression))
-							ereport(ERROR,
+					else if (strcmp(def->compression->cmname, compression->cmname))
+						ereport(ERROR,
 								(errcode(ERRCODE_DATATYPE_MISMATCH),
-									errmsg("column \"%s\" has a compression method conflict",
+								 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++;
@@ -2455,7 +2461,8 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->collOid = attribute->attcollation;
 				def->constraints = NIL;
 				def->location = -1;
-				def->compression = GetCompressionName(attribute->attcompression);
+				def->compression =
+						MakeColumnCompression(&attribute->attcompression);
 				inhSchema = lappend(inhSchema, def);
 				newattmap->attnums[parent_attno - 1] = ++child_attno;
 			}
@@ -2706,12 +2713,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 */
@@ -4779,7 +4786,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 */
@@ -6269,10 +6277,17 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	/* create attribute compresssion record */
 	if (rel->rd_rel->relkind == RELKIND_RELATION ||
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
-		attribute.attcompression = GetAttributeCompressionMethod(&attribute,
-																 colDef->compression);
+	{
+		char *compression =
+			GetAttributeCompression(&attribute, colDef->compression, NULL);
+
+		if (compression == NULL)
+			MemSet(NameStr(attribute.attcompression), 0, NAMEDATALEN);
+		else
+			namestrcpy(&(attribute.attcompression), compression);
+	}
 	else
-		attribute.attcompression = InvalidCompressionMethod;
+		MemSet(NameStr(attribute.attcompression), 0, NAMEDATALEN);
 
 	/* attribute.attacl is handled by InsertPgAttributeTuples() */
 
@@ -7698,9 +7713,9 @@ ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE loc
 						format_type_be(attrtuple->atttypid))));
 
 	/* use default compression if storage is not PLAIN */
-	if (!IsValidCompression(attrtuple->attcompression) &&
+	if (!IsValidCompression(*NameStr(attrtuple->attcompression)) &&
 		(newstorage != TYPSTORAGE_PLAIN))
-		attrtuple->attcompression = DefaultCompressionMethod;
+		InitAttributeCompression(attrtuple);
 
 	CatalogTupleUpdate(attrelation, &tuple->t_self, tuple);
 
@@ -11783,13 +11798,10 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
 	{
 		/* Setup attribute compression */
-		if (attTup->attstorage == TYPSTORAGE_PLAIN)
-			attTup->attcompression = InvalidCompressionMethod;
-		else if (!IsValidCompression(attTup->attcompression))
-			attTup->attcompression = DefaultCompressionMethod;
+		InitAttributeCompression(attTup);
 	}
 	else
-		attTup->attcompression = InvalidCompressionMethod;
+		MemSet(NameStr(attTup->attcompression), 0, NAMEDATALEN);
 
 	CatalogTupleUpdate(attrelation, &heapTup->t_self, heapTup);
 
@@ -14972,23 +14984,17 @@ 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;
-	char		cmethod;
-	Datum		values[Natts_pg_attribute];
-	bool		nulls[Natts_pg_attribute];
-	bool		replace[Natts_pg_attribute];
+	char	   *cm;
+	bool		need_rewrite;
 	ObjectAddress address;
 
-	Assert(IsA(newValue, String));
-	compression = strVal(newValue);
-
 	attrel = table_open(AttributeRelationId, RowExclusiveLock);
 
 	atttuple = SearchSysCacheAttName(RelationGetRelid(rel), column);
@@ -15007,24 +15013,20 @@ ATExecSetCompression(AlteredTableInfo *tab,
 				 errmsg("cannot alter system column \"%s\"", column)));
 
 	/* Prevent from altering untoastable attributes with PLAIN storage */
-	if (atttableform->attstorage == 'p' && !TypeIsToastable(atttableform->atttypid))
+	if (atttableform->attstorage == TYPSTORAGE_PLAIN &&
+		!TypeIsToastable(atttableform->atttypid))
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("column data type %s does not support compression",
 						format_type_be(atttableform->atttypid))));
 
-	/* Initialize buffers for new tuple values */
-	memset(values, 0, sizeof(values));
-	memset(nulls, false, sizeof(nulls));
-	memset(replace, false, sizeof(replace));
-
 	/* Get the attribute compression method. */
-	cmethod = GetAttributeCompressionMethod(atttableform, compression);
-
-	if (atttableform->attcompression != cmethod)
+	cm = GetAttributeCompression(atttableform, compression, &need_rewrite);
+	if (need_rewrite)
 		tab->rewrite |= AT_REWRITE_ALTER_COMPRESSION;
 
-	atttableform->attcompression = cmethod;
+	namestrcpy(&atttableform->attcompression, cm);
+
 	CatalogTupleUpdate(attrel, &atttuple->t_self, atttuple);
 
 	InvokeObjectPostAlterHook(RelationRelationId,
@@ -17730,27 +17732,3 @@ ATDetachCheckNoForeignKeyRefs(Relation partition)
 		table_close(rel, NoLock);
 	}
 }
-
-/*
- * GetAttributeCompressionMethod - Get compression method from compression name
- */
-static char
-GetAttributeCompressionMethod(Form_pg_attribute att, char *compression)
-{
-	char cm;
-
-	/* no compression for the plain storage */
-	if (att->attstorage == TYPSTORAGE_PLAIN)
-		return InvalidCompressionMethod;
-
-	/* fallback to default compression if it's not specified */
-	if (compression == NULL)
-		return DefaultCompressionMethod;
-
-	cm = GetCompressionMethod(compression);
-	if (!IsValidCompression(cm))
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("compression type \"%s\" not recognized", compression)));
-	return cm;
-}
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 0c005624e4..d6745959f9 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"
@@ -2050,11 +2051,13 @@ ExecCompareCompressionMethod(TupleTableSlot *slot, TupleDesc targetTupDesc)
 				continue;
 
 			/*
-			 * Get the compression method stored in the toast header and
-			 * compare with the compression method of the target.
+			 * Check whether the source value's compression method is supported
+			 * in the target table's attribute or not.  If it is not supported
+			 * then decompress the value.
 			 */
-			if (targetTupDesc->attrs[i].attcompression !=
-				GetCompressionMethodFromCompressionId(TOAST_COMPRESS_METHOD(new_value)))
+			if (strchr(NameStr(targetTupDesc->attrs[i].attcompression),
+				GetCompressionMethodFromCompressionId(
+						TOAST_COMPRESS_METHOD(new_value))) == NULL)
 			{
 				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 37fe5c72e9..aac399427c 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2930,7 +2930,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);
@@ -2950,6 +2950,17 @@ _copyColumnDef(const ColumnDef *from)
 	return newnode;
 }
 
+static ColumnCompression *
+_copyColumnCompression(const ColumnCompression *from)
+{
+	ColumnCompression *newnode = makeNode(ColumnCompression);
+
+	COPY_STRING_FIELD(cmname);
+	COPY_NODE_FIELD(preserve);
+
+	return newnode;
+}
+
 static Constraint *
 _copyConstraint(const Constraint *from)
 {
@@ -5631,6 +5642,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..35efa064e5 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,15 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 	return true;
 }
 
+static bool
+_equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b)
+{
+	COMPARE_STRING_FIELD(cmname);
+	COMPARE_NODE_FIELD(preserve);
+
+	return true;
+}
+
 static bool
 _equalConstraint(const Constraint *a, const Constraint *b)
 {
@@ -3683,6 +3692,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 718c2ed7fd..a1b5302d5b 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2835,7 +2835,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);
@@ -2853,6 +2853,15 @@ _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_NODE_FIELD(preserve);
+}
+
 static void
 _outTypeName(StringInfo str, const TypeName *node)
 {
@@ -4201,6 +4210,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 fb8b597b18..29ff3ad0f2 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,35 @@ 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;
+				}
+		;
 
 ColQualList:
 			ColQualList ColConstraint				{ $$ = lappend($1, $2); }
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index c305ad3692..8faea29f17 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1065,9 +1065,9 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 			def->storage = 0;
 
 		/* Likewise, copy compression if requested */
-		if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION
-			&& IsValidCompression(attribute->attcompression))
-			def->compression = GetCompressionName(attribute->attcompression);
+		if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION)
+			def->compression =
+					MakeColumnCompression(&attribute->attcompression);
 		else
 			def->compression = NULL;
 
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index 5135992b5c..a7ba8aac9f 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -156,8 +156,12 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(75,
 	/* attribute's collation */
 	Oid			attcollation;
 
-	/* attribute's compression options */
-	char		attcompression BKI_DEFAULT('\0');
+	/*
+	 * Attribute's compression options, the first character represents the
+	 * current compression method and followed by all old preserved compression
+	 * methods.
+	 */
+	NameData 	attcompression BKI_DEFAULT('\0');
 
 #ifdef CATALOG_VARLEN			/* variable-length fields start here */
 	/* NOTE: The following fields are not present in tuple descriptors. */
@@ -186,7 +190,7 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(75,
  * can access fields beyond attcollation except in a real tuple!
  */
 #define ATTRIBUTE_FIXED_PART_SIZE \
-	(offsetof(FormData_pg_attribute,attcompression) + sizeof(Oid))
+	(offsetof(FormData_pg_attribute, attcompression) + NAMEDATALEN)
 
 /* ----------------
  *		Form_pg_attribute corresponds to a pointer to a tuple with
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 7a079ef07f..97e6130284 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 char *GetAttributeCompression(Form_pg_attribute att,
+									ColumnCompression *compression,
+									bool *need_rewrite);
+extern void InitAttributeCompression(Form_pg_attribute att);
+extern ColumnCompression *MakeColumnCompression(NameData *compression);
+
 /* 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 7ddd8c011b..fed9cb7994 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 1863bdcb93..be99691f4a 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -624,6 +624,19 @@ 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;
+	List	   *preserve;
+} ColumnCompression;
+
 /*
  * ColumnDef - column definition (used in various creates)
  *
@@ -647,7 +660,7 @@ typedef struct ColumnDef
 	NodeTag		type;
 	char	   *colname;		/* name of column */
 	TypeName   *typeName;		/* type of column */
-	char	   *compression;
+	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 0052085cb3..33c7132353 100644
--- a/src/test/regress/expected/create_cm.out
+++ b/src/test/regress/expected/create_cm.out
@@ -98,4 +98,13 @@ ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz;
 --------+------+-----------+----------+---------+----------+-------------+--------------+-------------
  f1     | text |           |          |         | extended | pglz        |              | 
 
+-- preserve old compression method
+ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION zlib PRESERVE (pglz);
+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 | zlib        |              | 
+
 DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest;
diff --git a/src/test/regress/sql/create_cm.sql b/src/test/regress/sql/create_cm.sql
index b8949c8630..abf34eede0 100644
--- a/src/test/regress/sql/create_cm.sql
+++ b/src/test/regress/sql/create_cm.sql
@@ -48,4 +48,8 @@ ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION zlib;
 ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz;
 \d+ cmmove2
 
+-- preserve old compression method
+ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION zlib PRESERVE (pglz);
+INSERT INTO cmmove2 VALUES (repeat('1234567890',1004));
+\d+ cmmove2
 DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest;
-- 
2.23.0

