From 3a381237b22cd139f861c8085641e9e4a1d3697a Mon Sep 17 00:00:00 2001
From: Ildus Kurbangaliev <i.kurbangaliev@gmail.com>
Date: Mon, 18 Jun 2018 15:26:13 +0300
Subject: [PATCH 2/8] Add compression catalog tables and the basic
 infrastructure

Signed-off-by: Ildus Kurbangaliev <i.kurbangaliev@gmail.com>
---
 src/backend/access/Makefile                   |   4 +-
 src/backend/access/common/indextuple.c        |  25 +-
 src/backend/access/common/reloptions.c        |  89 +++
 src/backend/access/compression/Makefile       |  17 +
 src/backend/access/compression/cmapi.c        |  96 +++
 src/backend/access/heap/heapam.c              |   7 +-
 src/backend/access/heap/rewriteheap.c         |   2 +-
 src/backend/access/heap/tuptoaster.c          | 475 +++++++++++--
 src/backend/access/index/amapi.c              |  47 +-
 src/backend/bootstrap/bootstrap.c             |   1 +
 src/backend/catalog/Makefile                  |   2 +-
 src/backend/catalog/dependency.c              |  11 +-
 src/backend/catalog/heap.c                    |   3 +
 src/backend/catalog/index.c                   |   2 +
 src/backend/catalog/objectaddress.c           |  57 ++
 src/backend/commands/Makefile                 |   4 +-
 src/backend/commands/alter.c                  |   1 +
 src/backend/commands/amcmds.c                 |  84 +++
 src/backend/commands/compressioncmds.c        | 649 ++++++++++++++++++
 src/backend/commands/event_trigger.c          |   1 +
 src/backend/commands/foreigncmds.c            |  46 +-
 src/backend/commands/tablecmds.c              | 336 ++++++++-
 src/backend/nodes/copyfuncs.c                 |  16 +
 src/backend/nodes/equalfuncs.c                |  14 +
 src/backend/nodes/nodeFuncs.c                 |  10 +
 src/backend/nodes/outfuncs.c                  |  14 +
 src/backend/parser/parse_utilcmd.c            |   7 +
 .../replication/logical/reorderbuffer.c       |   2 +-
 src/backend/utils/adt/pseudotypes.c           |   1 +
 src/backend/utils/cache/syscache.c            |  12 +
 src/include/access/cmapi.h                    |  80 +++
 src/include/access/reloptions.h               |   3 +-
 src/include/access/tuptoaster.h               |  42 +-
 src/include/catalog/dependency.h              |   5 +-
 src/include/catalog/indexing.h                |   5 +
 src/include/catalog/pg_attr_compression.dat   |  23 +
 src/include/catalog/pg_attr_compression.h     |  53 ++
 src/include/catalog/pg_attribute.h            |   7 +-
 src/include/catalog/pg_class.dat              |   2 +-
 src/include/catalog/pg_proc.dat               |  10 +
 src/include/catalog/pg_type.dat               |   5 +
 src/include/catalog/toasting.h                |   1 +
 src/include/commands/defrem.h                 |  14 +
 src/include/nodes/nodes.h                     |   3 +-
 src/include/postgres.h                        |  26 +-
 src/include/utils/syscache.h                  |   1 +
 src/tools/pgindent/typedefs.list              |   3 +
 47 files changed, 2124 insertions(+), 194 deletions(-)
 create mode 100644 src/backend/access/compression/Makefile
 create mode 100644 src/backend/access/compression/cmapi.c
 create mode 100644 src/backend/commands/compressioncmds.c
 create mode 100644 src/include/access/cmapi.h
 create mode 100644 src/include/catalog/pg_attr_compression.dat
 create mode 100644 src/include/catalog/pg_attr_compression.h

diff --git a/src/backend/access/Makefile b/src/backend/access/Makefile
index bd93a6a8d1..4cd4fd9715 100644
--- a/src/backend/access/Makefile
+++ b/src/backend/access/Makefile
@@ -8,7 +8,7 @@ subdir = src/backend/access
 top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
-SUBDIRS	    = brin common gin gist hash heap index nbtree rmgrdesc spgist \
-			  tablesample transam
+SUBDIRS	    = brin common compression gin gist hash heap index nbtree \
+			  rmgrdesc spgist tablesample transam
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index aa52a96259..9a1cebbe2f 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -77,13 +77,30 @@ index_form_tuple(TupleDesc tupleDescriptor,
 
 		/*
 		 * If value is stored EXTERNAL, must fetch it so we are not depending
-		 * on outside storage.  This should be improved someday.
+		 * on outside storage.  This should be improved someday. If value also
+		 * was compressed by custom compression method then we should
+		 * decompress it too.
 		 */
 		if (VARATT_IS_EXTERNAL(DatumGetPointer(values[i])))
+		{
+			struct varatt_external toast_pointer;
+
+			VARATT_EXTERNAL_GET_POINTER(toast_pointer, DatumGetPointer(values[i]));
+			if (VARATT_EXTERNAL_IS_CUSTOM_COMPRESSED(toast_pointer))
+				untoasted_values[i] =
+					PointerGetDatum(heap_tuple_untoast_attr((struct varlena *)
+															DatumGetPointer(values[i])));
+			else
+				untoasted_values[i] =
+					PointerGetDatum(heap_tuple_fetch_attr((struct varlena *)
+														  DatumGetPointer(values[i])));
+			untoasted_free[i] = true;
+		}
+		else if (VARATT_IS_CUSTOM_COMPRESSED(DatumGetPointer(values[i])))
 		{
 			untoasted_values[i] =
-				PointerGetDatum(heap_tuple_fetch_attr((struct varlena *)
-													  DatumGetPointer(values[i])));
+				PointerGetDatum(heap_tuple_untoast_attr((struct varlena *)
+														DatumGetPointer(values[i])));
 			untoasted_free[i] = true;
 		}
 
@@ -95,7 +112,7 @@ index_form_tuple(TupleDesc tupleDescriptor,
 			VARSIZE(DatumGetPointer(untoasted_values[i])) > TOAST_INDEX_TARGET &&
 			(att->attstorage == 'x' || att->attstorage == 'm'))
 		{
-			Datum		cvalue = toast_compress_datum(untoasted_values[i]);
+			Datum		cvalue = toast_compress_datum(untoasted_values[i], InvalidOid);
 
 			if (DatumGetPointer(cvalue) != NULL)
 			{
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index db84da0678..bf836305a9 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -962,11 +962,100 @@ untransformRelOptions(Datum options)
 			val = (Node *) makeString(pstrdup(p));
 		}
 		result = lappend(result, makeDefElem(pstrdup(s), val, -1));
+		pfree(s);
 	}
 
 	return result;
 }
 
+/*
+ * helper function for qsort to compare DefElem
+ */
+static int
+compare_options(const void *a, const void *b)
+{
+	DefElem    *da = (DefElem *) lfirst(*(ListCell **) a);
+	DefElem    *db = (DefElem *) lfirst(*(ListCell **) b);
+
+	return strcmp(da->defname, db->defname);
+}
+
+/*
+ * Convert a DefElem list to the text array format that is used in
+ * pg_foreign_data_wrapper, pg_foreign_server, pg_user_mapping,
+ * pg_foreign_table and pg_attr_compression
+ *
+ * Returns the array in the form of a Datum, or PointerGetDatum(NULL)
+ * if the list is empty.
+ *
+ * Note: The array is usually stored to database without further
+ * processing, hence any validation should be done before this
+ * conversion.
+ */
+Datum
+optionListToArray(List *options, bool sorted)
+{
+	ArrayBuildState *astate = NULL;
+	ListCell   *cell;
+	List	   *resoptions = NIL;
+	int			len = list_length(options);
+
+	/* sort by option name if needed */
+	if (sorted && len > 1)
+		resoptions = list_qsort(options, compare_options);
+	else
+		resoptions = options;
+
+	foreach(cell, resoptions)
+	{
+		DefElem    *def = lfirst(cell);
+		const char *value;
+		Size		len;
+		text	   *t;
+
+		value = defGetString(def);
+		len = VARHDRSZ + strlen(def->defname) + 1 + strlen(value);
+		t = palloc(len + 1);
+		SET_VARSIZE(t, len);
+		sprintf(VARDATA(t), "%s=%s", def->defname, value);
+
+		astate = accumArrayResult(astate, PointerGetDatum(t),
+								  false, TEXTOID,
+								  CurrentMemoryContext);
+	}
+
+	/* free if the the modified version of list was used */
+	if (resoptions != options)
+		list_free(resoptions);
+
+	if (astate)
+		return makeArrayResult(astate, CurrentMemoryContext);
+
+	return PointerGetDatum(NULL);
+}
+
+/*
+ * Return human readable list of reloptions
+ */
+char *
+formatRelOptions(List *options)
+{
+	StringInfoData buf;
+	ListCell   *cell;
+
+	initStringInfo(&buf);
+
+	foreach(cell, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(cell);
+
+		appendStringInfo(&buf, "%s%s=%s", buf.len > 0 ? ", " : "",
+						 def->defname, defGetString(def));
+	}
+
+	return buf.data;
+}
+
 /*
  * Extract and parse reloptions from a pg_class tuple.
  *
diff --git a/src/backend/access/compression/Makefile b/src/backend/access/compression/Makefile
new file mode 100644
index 0000000000..a09dc787ed
--- /dev/null
+++ b/src/backend/access/compression/Makefile
@@ -0,0 +1,17 @@
+#-------------------------------------------------------------------------
+#
+# Makefile--
+#    Makefile for access/compression
+#
+# IDENTIFICATION
+#    src/backend/access/compression/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/backend/access/compression
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+OBJS = cmapi.o
+
+include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/compression/cmapi.c b/src/backend/access/compression/cmapi.c
new file mode 100644
index 0000000000..504da0638f
--- /dev/null
+++ b/src/backend/access/compression/cmapi.c
@@ -0,0 +1,96 @@
+/*-------------------------------------------------------------------------
+ *
+ * compression/cmapi.c
+ *	  Functions for compression access methods
+ *
+ * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/compression/cmapi.c
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+#include "fmgr.h"
+#include "access/cmapi.h"
+#include "access/htup_details.h"
+#include "access/reloptions.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression.h"
+#include "commands/defrem.h"
+#include "utils/syscache.h"
+
+/*
+ * InvokeCompressionAmHandler - call the specified access method handler routine to get
+ * its CompressionAmRoutine struct, which will be palloc'd in the caller's context.
+ */
+CompressionAmRoutine *
+InvokeCompressionAmHandler(Oid amhandler)
+{
+	Datum		datum;
+	CompressionAmRoutine *routine;
+
+	datum = OidFunctionCall0(amhandler);
+	routine = (CompressionAmRoutine *) DatumGetPointer(datum);
+
+	if (routine == NULL || !IsA(routine, CompressionAmRoutine))
+		elog(ERROR, "compression method handler function %u "
+			 "did not return an CompressionAmRoutine struct",
+			 amhandler);
+
+	return routine;
+}
+
+/*
+ * GetAttrCompressionOptions
+ *
+ * Parse array of attribute compression options and return it as a list.
+ */
+List *
+GetAttrCompressionOptions(Oid acoid)
+{
+	HeapTuple	tuple;
+	List	   *result = NIL;
+	bool		isnull;
+	Datum		acoptions;
+
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for an attribute compression %u", acoid);
+
+	/* options could be NULL, so we can't use form struct */
+	acoptions = SysCacheGetAttr(ATTCOMPRESSIONOID, tuple,
+								Anum_pg_attr_compression_acoptions, &isnull);
+
+	if (!isnull)
+		result = untransformRelOptions(acoptions);
+
+	ReleaseSysCache(tuple);
+	return result;
+}
+
+/*
+ * GetAttrCompressionAmOid
+ *
+ * Return access method Oid by attribute compression Oid
+ */
+Oid
+GetAttrCompressionAmOid(Oid acoid)
+{
+	Oid			result;
+	HeapTuple	tuple;
+	Form_pg_attr_compression acform;
+
+	/* extract access method Oid */
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+	acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+	result = get_compression_am_oid(NameStr(acform->acname), false);
+	ReleaseSysCache(tuple);
+
+	return result;
+}
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index fb63471a0e..d963cc50e8 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2702,8 +2702,9 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
 		Assert(!HeapTupleHasExternal(tup));
 		return tup;
 	}
-	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
-		return toast_insert_or_update(relation, tup, NULL, options);
+	else if (HeapTupleHasExternal(tup)
+			 || tup->t_len > TOAST_TUPLE_THRESHOLD)
+		return toast_insert_or_update(relation, tup, NULL, options, NULL);
 	else
 		return tup;
 }
@@ -4142,7 +4143,7 @@ l2:
 		if (need_toast)
 		{
 			/* Note we always use WAL and FSM during updates */
-			heaptup = toast_insert_or_update(relation, newtup, &oldtup, 0);
+			heaptup = toast_insert_or_update(relation, newtup, &oldtup, 0, NULL);
 			newtupsize = MAXALIGN(heaptup->t_len);
 		}
 		else
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index 7127788964..f479a70c2f 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -667,7 +667,7 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
 			options |= HEAP_INSERT_NO_LOGICAL;
 
 		heaptup = toast_insert_or_update(state->rs_new_rel, tup, NULL,
-										 options);
+										 options, NULL);
 	}
 	else
 		heaptup = tup;
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index cd42c50b09..785efc113f 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -30,17 +30,25 @@
 #include <unistd.h>
 #include <fcntl.h>
 
+#include "access/cmapi.h"
 #include "access/genam.h"
 #include "access/heapam.h"
+#include "access/reloptions.h"
 #include "access/tuptoaster.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
+#include "catalog/pg_am.h"
+#include "commands/defrem.h"
 #include "common/pg_lzcompress.h"
 #include "miscadmin.h"
 #include "utils/expandeddatum.h"
 #include "utils/fmgroids.h"
+#include "utils/hsearch.h"
+#include "utils/inval.h"
+#include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
+#include "utils/syscache.h"
 #include "utils/typcache.h"
 #include "utils/tqual.h"
 
@@ -53,19 +61,51 @@
 typedef struct toast_compress_header
 {
 	int32		vl_len_;		/* varlena header (do not touch directly!) */
-	int32		rawsize;
+	uint32		info;			/* flags (2 bits) and rawsize */
 } toast_compress_header;
 
+/*
+ * If the compression method were used, then data also contains
+ * Oid of compression options
+ */
+typedef struct toast_compress_header_custom
+{
+	int32		vl_len_;		/* varlena header (do not touch directly!) */
+	uint32		info;			/* flags (2 high bits) and rawsize */
+	Oid			cmid;			/* Oid from pg_attr_compression */
+}			toast_compress_header_custom;
+
+static HTAB *amoptions_cache = NULL;
+static MemoryContext amoptions_cache_mcxt = NULL;
+
+#define RAWSIZEMASK (0x3FFFFFFFU)
+
 /*
  * Utilities for manipulation of header information for compressed
  * toast entries.
+ *
+ * Since version 11 TOAST_COMPRESS_SET_RAWSIZE also marks compressed
+ * varlenas as custom compressed. Such varlenas will contain 0x02 (0b10) in
+ * two highest bits.
  */
-#define TOAST_COMPRESS_HDRSZ		((int32) sizeof(toast_compress_header))
-#define TOAST_COMPRESS_RAWSIZE(ptr) (((toast_compress_header *) (ptr))->rawsize)
+#define TOAST_COMPRESS_HDRSZ			((int32) sizeof(toast_compress_header))
+#define TOAST_COMPRESS_HDRSZ_CUSTOM		((int32) sizeof(toast_compress_header_custom))
+#define TOAST_COMPRESS_RAWSIZE(ptr) (((toast_compress_header *) (ptr))->info & RAWSIZEMASK)
 #define TOAST_COMPRESS_RAWDATA(ptr) \
 	(((char *) (ptr)) + TOAST_COMPRESS_HDRSZ)
 #define TOAST_COMPRESS_SET_RAWSIZE(ptr, len) \
-	(((toast_compress_header *) (ptr))->rawsize = (len))
+do { \
+	Assert(len > 0 && len <= RAWSIZEMASK); \
+	((toast_compress_header *) (ptr))->info = (len) | (0x02 << 30); \
+} while (0)
+#define TOAST_COMPRESS_SET_CMID(ptr, oid) \
+	(((toast_compress_header_custom *) (ptr))->cmid = (oid))
+
+#define VARATT_EXTERNAL_SET_CUSTOM(toast_pointer) \
+do { \
+	((toast_pointer).va_extinfo |= (1 << 31)); \
+	((toast_pointer).va_extinfo &= ~(1 << 30)); \
+} while (0)
 
 static void toast_delete_datum(Relation rel, Datum value, bool is_speculative);
 static Datum toast_save_datum(Relation rel, Datum value,
@@ -83,7 +123,8 @@ static int toast_open_indexes(Relation toastrel,
 static void toast_close_indexes(Relation *toastidxs, int num_indexes,
 					LOCKMODE lock);
 static void init_toast_snapshot(Snapshot toast_snapshot);
-
+static void init_amoptions_cache(void);
+static bool attr_compression_options_are_equal(Oid acoid1, Oid acoid2);
 
 /* ----------
  * heap_tuple_fetch_attr -
@@ -421,7 +462,7 @@ toast_datum_size(Datum value)
 		struct varatt_external toast_pointer;
 
 		VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
-		result = toast_pointer.va_extsize;
+		result = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
 	}
 	else if (VARATT_IS_EXTERNAL_INDIRECT(attr))
 	{
@@ -511,6 +552,98 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
 	}
 }
 
+/* ----------
+ * toast_prepare_varlena
+ *
+ * Untoast or fetch varlena if needed. This function could return
+ * compressed varlena.
+ */
+static struct varlena *
+toast_prepare_varlena(Form_pg_attribute att, struct varlena *value,
+					  List *preserved_amoids, bool *modified)
+{
+	struct varlena *tmpval = NULL;
+
+	*modified = false;
+
+	/*
+	 * We took care of UPDATE, so any external value we find still in the
+	 * tuple must be someone else's that we cannot reuse (this includes the
+	 * case of an out-of-line in-memory datum). Fetch it back (without
+	 * decompression, unless we are forcing PLAIN storage or it has a custom
+	 * compression).  If necessary, we'll push it out as a new external value
+	 * below.
+	 */
+	if (VARATT_IS_EXTERNAL(value))
+	{
+		*modified = true;
+		if (att->attstorage == 'p')
+		{
+			value = heap_tuple_untoast_attr(value);
+
+			/* untoasted value does not need any further work */
+			return value;
+		}
+		else
+			value = heap_tuple_fetch_attr(value);
+	}
+
+	/*
+	 * Process custom compressed datum.
+	 *
+	 * 1) If destination column has identical compression move the data as is
+	 * and only change attribute compression Oid. 2) If it's rewrite from
+	 * ALTER command check list of preserved compression access methods. 3) In
+	 * other cases just untoast the datum.
+	 */
+	if (VARATT_IS_CUSTOM_COMPRESSED(value))
+	{
+		bool		storage_ok = (att->attstorage == 'm' || att->attstorage == 'x');
+		toast_compress_header_custom *hdr;
+
+		hdr = (toast_compress_header_custom *) value;
+
+		/* nothing to do */
+		if (storage_ok && hdr->cmid == att->attcompression)
+			return value;
+
+		if (storage_ok &&
+			OidIsValid(att->attcompression) &&
+			attr_compression_options_are_equal(att->attcompression, hdr->cmid))
+		{
+			/* identical compression, just change Oid to new one */
+			tmpval = palloc(VARSIZE(value));
+			memcpy(tmpval, value, VARSIZE(value));
+			TOAST_COMPRESS_SET_CMID(tmpval, att->attcompression);
+		}
+		else if (preserved_amoids != NULL)
+		{
+			Oid			amoid = GetAttrCompressionAmOid(hdr->cmid);
+
+			/* decompress the value if it's not in preserved list */
+			if (!list_member_oid(preserved_amoids, amoid))
+				tmpval = heap_tuple_untoast_attr(value);
+		}
+		else
+			/* just decompress the value */
+			tmpval = heap_tuple_untoast_attr(value);
+
+	}
+	else if (OidIsValid(att->attcompression))
+		tmpval = heap_tuple_untoast_attr(value);
+
+	if (tmpval && tmpval != value)
+	{
+		/* if value was external we need to free fetched data */
+		if (*modified)
+			pfree(value);
+
+		value = tmpval;
+		*modified = true;
+	}
+
+	return value;
+}
 
 /* ----------
  * toast_insert_or_update -
@@ -522,6 +655,8 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
  *	newtup: the candidate new tuple to be inserted
  *	oldtup: the old row version for UPDATE, or NULL for INSERT
  *	options: options to be passed to heap_insert() for toast rows
+ *	preserved_am_info: hash table with attnum as key, for lists of compression
+ *		methods which should be preserved on decompression.
  * Result:
  *	either newtup if no toasting is needed, or a palloc'd modified tuple
  *	that is what should actually get stored
@@ -532,7 +667,7 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
  */
 HeapTuple
 toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
-					   int options)
+					   int options, HTAB *preserved_am_info)
 {
 	HeapTuple	result_tuple;
 	TupleDesc	tupleDesc;
@@ -667,27 +802,37 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		 */
 		if (att->attlen == -1)
 		{
+			bool		modified = false;
+			List	   *preserved_amoids = NIL;
+
 			/*
 			 * If the table's attribute says PLAIN always, force it so.
 			 */
 			if (att->attstorage == 'p')
 				toast_action[i] = 'p';
 
+			if (VARATT_IS_EXTERNAL(new_value))
+				toast_oldexternal[i] = new_value;
+
 			/*
-			 * We took care of UPDATE above, so any external value we find
-			 * still in the tuple must be someone else's that we cannot reuse
-			 * (this includes the case of an out-of-line in-memory datum).
-			 * Fetch it back (without decompression, unless we are forcing
-			 * PLAIN storage).  If necessary, we'll push it out as a new
-			 * external value below.
+			 * If it's ALTER SET COMPRESSION command we should check that
+			 * access method of compression in preserved list.
 			 */
-			if (VARATT_IS_EXTERNAL(new_value))
+			if (preserved_am_info != NULL)
+			{
+				bool		found;
+				AttrCmPreservedInfo *pinfo;
+
+				pinfo = hash_search(preserved_am_info, &att->attnum,
+									HASH_FIND, &found);
+				if (found)
+					preserved_amoids = pinfo->preserved_amoids;
+			}
+
+			new_value = toast_prepare_varlena(att,
+											  new_value, preserved_amoids, &modified);
+			if (modified)
 			{
-				toast_oldexternal[i] = new_value;
-				if (att->attstorage == 'p')
-					new_value = heap_tuple_untoast_attr(new_value);
-				else
-					new_value = heap_tuple_fetch_attr(new_value);
 				toast_values[i] = PointerGetDatum(new_value);
 				toast_free[i] = true;
 				need_change = true;
@@ -741,12 +886,14 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		Datum		old_value;
 		Datum		new_value;
 
+		Form_pg_attribute att;
+
 		/*
 		 * Search for the biggest yet unprocessed internal attribute
 		 */
 		for (i = 0; i < numAttrs; i++)
 		{
-			Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
+			char		attstorage;
 
 			if (toast_action[i] != ' ')
 				continue;
@@ -754,7 +901,9 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 				continue;		/* can't happen, toast_action would be 'p' */
 			if (VARATT_IS_COMPRESSED(DatumGetPointer(toast_values[i])))
 				continue;
-			if (att->attstorage != 'x' && att->attstorage != 'e')
+
+			attstorage = (TupleDescAttr(tupleDesc, i))->attstorage;
+			if (attstorage != 'x' && attstorage != 'e')
 				continue;
 			if (toast_sizes[i] > biggest_size)
 			{
@@ -770,10 +919,11 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		 * Attempt to compress it inline, if it has attstorage 'x'
 		 */
 		i = biggest_attno;
-		if (TupleDescAttr(tupleDesc, i)->attstorage == 'x')
+		att = TupleDescAttr(tupleDesc, i);
+		if (att->attstorage == 'x')
 		{
 			old_value = toast_values[i];
-			new_value = toast_compress_datum(old_value);
+			new_value = toast_compress_datum(old_value, att->attcompression);
 
 			if (DatumGetPointer(new_value) != NULL)
 			{
@@ -914,7 +1064,8 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
 		 */
 		i = biggest_attno;
 		old_value = toast_values[i];
-		new_value = toast_compress_datum(old_value);
+		new_value = toast_compress_datum(old_value,
+										 TupleDescAttr(tupleDesc, i)->attcompression);
 
 		if (DatumGetPointer(new_value) != NULL)
 		{
@@ -1353,6 +1504,18 @@ toast_build_flattened_tuple(TupleDesc tupleDesc,
 	return new_tuple;
 }
 
+/* ---------
+ * toast_set_compressed_datum_info
+ *
+ * Save metadata in compressed datum. This information will required
+ * for decompression.
+ */
+void
+toast_set_compressed_datum_info(struct varlena *val, Oid cmid, int32 rawsize)
+{
+	TOAST_COMPRESS_SET_RAWSIZE(val, rawsize);
+	TOAST_COMPRESS_SET_CMID(val, cmid);
+}
 
 /* ----------
  * toast_compress_datum -
@@ -1368,54 +1531,47 @@ toast_build_flattened_tuple(TupleDesc tupleDesc,
  * ----------
  */
 Datum
-toast_compress_datum(Datum value)
+toast_compress_datum(Datum value, Oid acoid)
 {
-	struct varlena *tmp;
-	int32		valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
-	int32		len;
+	struct varlena *tmp = NULL;
+	int32		valsize;
+	CompressionAmOptions *cmoptions = NULL;
 
 	Assert(!VARATT_IS_EXTERNAL(DatumGetPointer(value)));
 	Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value)));
 
-	/*
-	 * No point in wasting a palloc cycle if value size is out of the allowed
-	 * range for compression
-	 */
-	if (valsize < PGLZ_strategy_default->min_input_size ||
-		valsize > PGLZ_strategy_default->max_input_size)
-		return PointerGetDatum(NULL);
+	/* Fallback to default compression if not specified */
+	if (!OidIsValid(acoid))
+		acoid = DefaultCompressionOid;
 
-	tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) +
-									TOAST_COMPRESS_HDRSZ);
+	cmoptions = lookup_compression_am_options(acoid);
+	tmp = cmoptions->amroutine->cmcompress(cmoptions, (const struct varlena *) value);
+	if (!tmp)
+		return PointerGetDatum(NULL);
 
 	/*
-	 * We recheck the actual size even if pglz_compress() reports success,
-	 * because it might be satisfied with having saved as little as one byte
-	 * in the compressed data --- which could turn into a net loss once you
-	 * consider header and alignment padding.  Worst case, the compressed
-	 * format might require three padding bytes (plus header, which is
-	 * included in VARSIZE(tmp)), whereas the uncompressed format would take
-	 * only one header byte and no padding if the value is short enough.  So
-	 * we insist on a savings of more than 2 bytes to ensure we have a gain.
+	 * We recheck the actual size even if compression function reports
+	 * success, because it might be satisfied with having saved as little as
+	 * one byte in the compressed data --- which could turn into a net loss
+	 * once you consider header and alignment padding.  Worst case, the
+	 * compressed format might require three padding bytes (plus header, which
+	 * is included in VARSIZE(tmp)), whereas the uncompressed format would
+	 * take only one header byte and no padding if the value is short enough.
+	 * So we insist on a savings of more than 2 bytes to ensure we have a
+	 * gain.
 	 */
-	len = pglz_compress(VARDATA_ANY(DatumGetPointer(value)),
-						valsize,
-						TOAST_COMPRESS_RAWDATA(tmp),
-						PGLZ_strategy_default);
-	if (len >= 0 &&
-		len + TOAST_COMPRESS_HDRSZ < valsize - 2)
+	valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
+
+	if (VARSIZE(tmp) < valsize - 2)
 	{
-		TOAST_COMPRESS_SET_RAWSIZE(tmp, valsize);
-		SET_VARSIZE_COMPRESSED(tmp, len + TOAST_COMPRESS_HDRSZ);
 		/* successful compression */
+		toast_set_compressed_datum_info(tmp, cmoptions->acoid, valsize);
 		return PointerGetDatum(tmp);
 	}
-	else
-	{
-		/* incompressible data */
-		pfree(tmp);
-		return PointerGetDatum(NULL);
-	}
+
+	/* incompressible data */
+	pfree(tmp);
+	return PointerGetDatum(NULL);
 }
 
 
@@ -1510,19 +1666,20 @@ toast_save_datum(Relation rel, Datum value,
 									&num_indexes);
 
 	/*
-	 * Get the data pointer and length, and compute va_rawsize and va_extsize.
+	 * Get the data pointer and length, and compute va_rawsize and va_extinfo.
 	 *
 	 * va_rawsize is the size of the equivalent fully uncompressed datum, so
 	 * we have to adjust for short headers.
 	 *
-	 * va_extsize is the actual size of the data payload in the toast records.
+	 * va_extinfo contains the actual size of the data payload in the toast
+	 * records.
 	 */
 	if (VARATT_IS_SHORT(dval))
 	{
 		data_p = VARDATA_SHORT(dval);
 		data_todo = VARSIZE_SHORT(dval) - VARHDRSZ_SHORT;
 		toast_pointer.va_rawsize = data_todo + VARHDRSZ;	/* as if not short */
-		toast_pointer.va_extsize = data_todo;
+		toast_pointer.va_extinfo = data_todo;	/* no flags */
 	}
 	else if (VARATT_IS_COMPRESSED(dval))
 	{
@@ -1530,7 +1687,10 @@ toast_save_datum(Relation rel, Datum value,
 		data_todo = VARSIZE(dval) - VARHDRSZ;
 		/* rawsize in a compressed datum is just the size of the payload */
 		toast_pointer.va_rawsize = VARRAWSIZE_4B_C(dval) + VARHDRSZ;
-		toast_pointer.va_extsize = data_todo;
+		toast_pointer.va_extinfo = data_todo;
+		if (VARATT_IS_CUSTOM_COMPRESSED(dval))
+			VARATT_EXTERNAL_SET_CUSTOM(toast_pointer);
+
 		/* Assert that the numbers look like it's compressed */
 		Assert(VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer));
 	}
@@ -1539,7 +1699,7 @@ toast_save_datum(Relation rel, Datum value,
 		data_p = VARDATA(dval);
 		data_todo = VARSIZE(dval) - VARHDRSZ;
 		toast_pointer.va_rawsize = VARSIZE(dval);
-		toast_pointer.va_extsize = data_todo;
+		toast_pointer.va_extinfo = data_todo;	/* no flags */
 	}
 
 	/*
@@ -1899,7 +2059,7 @@ toast_fetch_datum(struct varlena *attr)
 	/* Must copy to access aligned fields */
 	VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
 
-	ressize = toast_pointer.va_extsize;
+	ressize = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
 	numchunks = ((ressize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
 
 	result = (struct varlena *) palloc(ressize + VARHDRSZ);
@@ -2084,7 +2244,7 @@ toast_fetch_datum_slice(struct varlena *attr, int32 sliceoffset, int32 length)
 	 */
 	Assert(!VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer));
 
-	attrsize = toast_pointer.va_extsize;
+	attrsize = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
 	totalchunks = ((attrsize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
 
 	if (sliceoffset >= attrsize)
@@ -2280,15 +2440,26 @@ toast_decompress_datum(struct varlena *attr)
 
 	Assert(VARATT_IS_COMPRESSED(attr));
 
-	result = (struct varlena *)
-		palloc(TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
-	SET_VARSIZE(result, TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+	if (VARATT_IS_CUSTOM_COMPRESSED(attr))
+	{
+		CompressionAmOptions *cmoptions;
+		toast_compress_header_custom *hdr;
 
-	if (pglz_decompress(TOAST_COMPRESS_RAWDATA(attr),
-						VARSIZE(attr) - TOAST_COMPRESS_HDRSZ,
-						VARDATA(result),
-						TOAST_COMPRESS_RAWSIZE(attr)) < 0)
-		elog(ERROR, "compressed data is corrupted");
+		hdr = (toast_compress_header_custom *) attr;
+		cmoptions = lookup_compression_am_options(hdr->cmid);
+		result = cmoptions->amroutine->cmdecompress(cmoptions, attr);
+	}
+	else
+	{
+		result = (struct varlena *)
+			palloc(TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+		SET_VARSIZE(result, TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+		if (pglz_decompress(TOAST_COMPRESS_RAWDATA(attr),
+							VARSIZE(attr) - TOAST_COMPRESS_HDRSZ,
+							VARDATA(result),
+							TOAST_COMPRESS_RAWSIZE(attr)) < 0)
+			elog(ERROR, "compressed data is corrupted");
+	}
 
 	return result;
 }
@@ -2390,3 +2561,159 @@ init_toast_snapshot(Snapshot toast_snapshot)
 
 	InitToastSnapshot(*toast_snapshot, snapshot->lsn, snapshot->whenTaken);
 }
+
+/* ----------
+ * remove_cached_amoptions
+ *
+ * Remove specified compression options from cache
+ */
+static void
+remove_cached_amoptions(CompressionAmOptions *entry)
+{
+	MemoryContextDelete(entry->mcxt);
+
+	if (hash_search(amoptions_cache, (void *) &entry->acoid,
+					HASH_REMOVE, NULL) == NULL)
+		elog(ERROR, "hash table corrupted");
+}
+
+/* ----------
+ * invalidate_amoptions_cache
+ *
+ * Flush cache entries when pg_attr_compression is updated.
+ */
+static void
+invalidate_amoptions_cache(Datum arg, int cacheid, uint32 hashvalue)
+{
+	HASH_SEQ_STATUS status;
+	CompressionAmOptions *entry;
+
+	hash_seq_init(&status, amoptions_cache);
+	while ((entry = (CompressionAmOptions *) hash_seq_search(&status)) != NULL)
+		remove_cached_amoptions(entry);
+}
+
+/* ----------
+ * init_amoptions_cache
+ *
+ * Initialize a local cache for attribute compression options.
+ */
+static void
+init_amoptions_cache(void)
+{
+	HASHCTL		ctl;
+
+	amoptions_cache_mcxt = AllocSetContextCreate(TopMemoryContext,
+												 "attr compression cache",
+												 ALLOCSET_DEFAULT_SIZES);
+	MemSet(&ctl, 0, sizeof(ctl));
+	ctl.keysize = sizeof(Oid);
+	ctl.entrysize = sizeof(CompressionAmOptions);
+	ctl.hcxt = amoptions_cache_mcxt;
+	amoptions_cache = hash_create("attr compression cache", 100, &ctl,
+								  HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+
+	/* Set up invalidation callback on pg_attr_compression. */
+	CacheRegisterSyscacheCallback(ATTCOMPRESSIONOID,
+								  invalidate_amoptions_cache,
+								  (Datum) 0);
+}
+
+/* ----------
+ * lookup_compression_am_options
+ *
+ * Return or generate cache record for attribute compression.
+ */
+CompressionAmOptions *
+lookup_compression_am_options(Oid acoid)
+{
+	bool		found,
+				optisnull;
+	CompressionAmOptions *result;
+	Datum		acoptions;
+	Form_pg_attr_compression acform;
+	HeapTuple	tuple;
+	Oid			amoid;
+	regproc		amhandler;
+
+	/*
+	 * This call could invalidate system cache so we need to call it before
+	 * we're putting something to our cache.
+	 */
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+	acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+	amoid = get_compression_am_oid(NameStr(acform->acname), false);
+	amhandler = get_am_handler_oid(amoid, AMTYPE_COMPRESSION, false);
+	acoptions = SysCacheGetAttr(ATTCOMPRESSIONOID, tuple,
+					Anum_pg_attr_compression_acoptions, &optisnull);
+
+	if (!amoptions_cache)
+		init_amoptions_cache();
+
+	result = hash_search(amoptions_cache, &acoid, HASH_ENTER, &found);
+	if (!found)
+	{
+		MemoryContext oldcxt = CurrentMemoryContext;
+
+		result->mcxt = AllocSetContextCreate(amoptions_cache_mcxt,
+											 "compression am options",
+											 ALLOCSET_DEFAULT_SIZES);
+
+		PG_TRY();
+		{
+			result->acoid = acoid;
+			result->amoid = amoid;
+
+			MemoryContextSwitchTo(result->mcxt);
+			result->amroutine =	InvokeCompressionAmHandler(amhandler);
+			if (optisnull)
+				result->acoptions = NIL;
+			else
+				result->acoptions = untransformRelOptions(acoptions);
+			result->acstate = result->amroutine->cminitstate ?
+				result->amroutine->cminitstate(acoid, result->acoptions) : NULL;
+		}
+		PG_CATCH();
+		{
+			MemoryContextSwitchTo(oldcxt);
+			remove_cached_amoptions(result);
+			PG_RE_THROW();
+		}
+		PG_END_TRY();
+		MemoryContextSwitchTo(oldcxt);
+	}
+
+	ReleaseSysCache(tuple);
+
+	if (!result->amroutine)
+		/* should not happen but need to check if something goes wrong */
+		elog(ERROR, "compression access method routine is NULL");
+
+	return result;
+}
+
+/* ----------
+ * attr_compression_options_are_equal
+ *
+ * Compare two attribute compression records
+ */
+static bool
+attr_compression_options_are_equal(Oid acoid1, Oid acoid2)
+{
+	CompressionAmOptions *a;
+	CompressionAmOptions *b;
+
+	a = lookup_compression_am_options(acoid1);
+	b = lookup_compression_am_options(acoid2);
+
+	if (a->amoid != b->amoid)
+		return false;
+
+	if (list_length(a->acoptions) != list_length(b->acoptions))
+		return false;
+
+	return equal(a->acoptions, b->acoptions);
+}
diff --git a/src/backend/access/index/amapi.c b/src/backend/access/index/amapi.c
index f395cb1ab4..353cd4bc39 100644
--- a/src/backend/access/index/amapi.c
+++ b/src/backend/access/index/amapi.c
@@ -17,6 +17,7 @@
 #include "access/htup_details.h"
 #include "catalog/pg_am.h"
 #include "catalog/pg_opclass.h"
+#include "commands/defrem.h"
 #include "utils/builtins.h"
 #include "utils/syscache.h"
 
@@ -55,52 +56,12 @@ GetIndexAmRoutine(Oid amhandler)
 IndexAmRoutine *
 GetIndexAmRoutineByAmId(Oid amoid, bool noerror)
 {
-	HeapTuple	tuple;
-	Form_pg_am	amform;
 	regproc		amhandler;
 
 	/* Get handler function OID for the access method */
-	tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(amoid));
-	if (!HeapTupleIsValid(tuple))
-	{
-		if (noerror)
-			return NULL;
-		elog(ERROR, "cache lookup failed for access method %u",
-			 amoid);
-	}
-	amform = (Form_pg_am) GETSTRUCT(tuple);
-
-	/* Check if it's an index access method as opposed to some other AM */
-	if (amform->amtype != AMTYPE_INDEX)
-	{
-		if (noerror)
-		{
-			ReleaseSysCache(tuple);
-			return NULL;
-		}
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-				 errmsg("access method \"%s\" is not of type %s",
-						NameStr(amform->amname), "INDEX")));
-	}
-
-	amhandler = amform->amhandler;
-
-	/* Complain if handler OID is invalid */
-	if (!RegProcedureIsValid(amhandler))
-	{
-		if (noerror)
-		{
-			ReleaseSysCache(tuple);
-			return NULL;
-		}
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-				 errmsg("index access method \"%s\" does not have a handler",
-						NameStr(amform->amname))));
-	}
-
-	ReleaseSysCache(tuple);
+	amhandler = get_am_handler_oid(amoid, AMTYPE_INDEX, noerror);
+	if (noerror && !RegProcedureIsValid(amhandler))
+		return NULL;
 
 	/* And finally, call the handler function to get the API struct. */
 	return GetIndexAmRoutine(amhandler);
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 578af2e66d..1343efe76d 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -748,6 +748,7 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
 	attrtypes[attnum]->attcacheoff = -1;
 	attrtypes[attnum]->atttypmod = -1;
 	attrtypes[attnum]->attislocal = true;
+	attrtypes[attnum]->attcompression = InvalidOid;
 
 	if (nullness == BOOTCOL_NULL_FORCE_NOT_NULL)
 	{
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 0865240f11..5560fc05b8 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -34,7 +34,7 @@ CATALOG_HEADERS := \
 	pg_attrdef.h pg_constraint.h pg_inherits.h pg_index.h pg_operator.h \
 	pg_opfamily.h pg_opclass.h pg_am.h pg_amop.h pg_amproc.h \
 	pg_language.h pg_largeobject_metadata.h pg_largeobject.h pg_aggregate.h \
-	pg_statistic_ext.h \
+	pg_statistic_ext.h pg_attr_compression.h \
 	pg_statistic.h pg_rewrite.h pg_trigger.h pg_event_trigger.h pg_description.h \
 	pg_cast.h pg_enum.h pg_namespace.h pg_conversion.h pg_depend.h \
 	pg_database.h pg_db_role_setting.h pg_tablespace.h pg_pltemplate.h \
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 7dfa3278a5..0d3bf0dd05 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -23,6 +23,7 @@
 #include "catalog/pg_am.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
 #include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
@@ -170,7 +171,8 @@ static const Oid object_classes[] = {
 	PublicationRelationId,		/* OCLASS_PUBLICATION */
 	PublicationRelRelationId,	/* OCLASS_PUBLICATION_REL */
 	SubscriptionRelationId,		/* OCLASS_SUBSCRIPTION */
-	TransformRelationId			/* OCLASS_TRANSFORM */
+	TransformRelationId,		/* OCLASS_TRANSFORM */
+	AttrCompressionRelationId	/* OCLASS_ATTR_COMPRESSION */
 };
 
 
@@ -1280,6 +1282,10 @@ doDeletion(const ObjectAddress *object, int flags)
 			DropTransformById(object->objectId);
 			break;
 
+		case OCLASS_ATTR_COMPRESSION:
+			RemoveAttributeCompression(object->objectId);
+			break;
+
 			/*
 			 * These global object types are not supported here.
 			 */
@@ -2538,6 +2544,9 @@ getObjectClass(const ObjectAddress *object)
 
 		case TransformRelationId:
 			return OCLASS_TRANSFORM;
+
+		case AttrCompressionRelationId:
+			return OCLASS_ATTR_COMPRESSION;
 	}
 
 	/* shouldn't get here */
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 3c9c03c997..ce29d0d070 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -689,6 +689,7 @@ InsertPgAttributeTuple(Relation pg_attribute_rel,
 	values[Anum_pg_attribute_attislocal - 1] = BoolGetDatum(new_attribute->attislocal);
 	values[Anum_pg_attribute_attinhcount - 1] = Int32GetDatum(new_attribute->attinhcount);
 	values[Anum_pg_attribute_attcollation - 1] = ObjectIdGetDatum(new_attribute->attcollation);
+	values[Anum_pg_attribute_attcompression - 1] = ObjectIdGetDatum(new_attribute->attcompression);
 
 	/* start out with empty permissions and empty options */
 	nulls[Anum_pg_attribute_attacl - 1] = true;
@@ -1661,6 +1662,8 @@ RemoveAttributeById(Oid relid, AttrNumber attnum)
 		/* We don't want to keep stats for it anymore */
 		attStruct->attstattarget = 0;
 
+		attStruct->attcompression = InvalidOid;
+
 		/*
 		 * Change the column name to something that isn't likely to conflict
 		 */
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 5885899c9b..150cb91839 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -410,6 +410,7 @@ ConstructTupleDescriptor(Relation heapRelation,
 			to->attstorage = typeTup->typstorage;
 			to->attalign = typeTup->typalign;
 			to->atttypmod = exprTypmod(indexkey);
+			to->attcompression = InvalidOid;
 
 			ReleaseSysCache(tuple);
 
@@ -496,6 +497,7 @@ ConstructTupleDescriptor(Relation heapRelation,
 			to->attbyval = typeTup->typbyval;
 			to->attalign = typeTup->typalign;
 			to->attstorage = typeTup->typstorage;
+			to->attcompression = InvalidOid;
 
 			ReleaseSysCache(tuple);
 		}
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 593e6f7022..2429e9b678 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -23,6 +23,7 @@
 #include "catalog/pg_am.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
 #include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
@@ -489,6 +490,18 @@ static const ObjectPropertyType ObjectProperty[] =
 		InvalidAttrNumber,		/* no ACL (same as relation) */
 		OBJECT_STATISTIC_EXT,
 		true
+	},
+	{
+		AttrCompressionRelationId,
+		AttrCompressionIndexId,
+		ATTCOMPRESSIONOID,
+		-1,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		-1,
+		false
 	}
 };
 
@@ -3548,6 +3561,37 @@ getObjectDescription(const ObjectAddress *object)
 				break;
 			}
 
+		case OCLASS_ATTR_COMPRESSION:
+			{
+				char	   *attname;
+				HeapTuple	tup;
+				Form_pg_attr_compression acform;
+
+				tup = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(tup))
+					elog(ERROR, "cache lookup failed for attribute compression %u", object->objectId);
+
+				acform = (Form_pg_attr_compression) GETSTRUCT(tup);
+				if (OidIsValid(acform->acrelid))
+				{
+					appendStringInfo(&buffer, _("compression on "));
+					getRelationDescription(&buffer, acform->acrelid);
+
+					attname = get_attname(acform->acrelid, acform->acattnum, true);
+					if (attname)
+					{
+						appendStringInfo(&buffer, _(" column %s"), attname);
+						pfree(attname);
+					}
+				}
+				else
+					appendStringInfo(&buffer, _("attribute compression %u"), object->objectId);
+
+				ReleaseSysCache(tup);
+
+				break;
+			}
+
 		case OCLASS_TRANSFORM:
 			{
 				HeapTuple	trfTup;
@@ -4072,6 +4116,10 @@ getObjectTypeDescription(const ObjectAddress *object)
 			appendStringInfoString(&buffer, "transform");
 			break;
 
+		case OCLASS_ATTR_COMPRESSION:
+			appendStringInfoString(&buffer, "attribute compression");
+			break;
+
 			/*
 			 * There's intentionally no default: case here; we want the
 			 * compiler to warn if a new OCLASS hasn't been handled above.
@@ -4616,6 +4664,15 @@ getObjectIdentityParts(const ObjectAddress *object,
 				break;
 			}
 
+		case OCLASS_ATTR_COMPRESSION:
+			{
+				appendStringInfo(&buffer, "%u",
+								 object->objectId);
+				if (objname)
+					*objname = list_make1(psprintf("%u", object->objectId));
+				break;
+			}
+
 		case OCLASS_REWRITE:
 			{
 				Relation	ruleDesc;
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 4a6c99e090..e23abf64f1 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -13,8 +13,8 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \
-	collationcmds.o constraint.o conversioncmds.o copy.o createas.o \
-	dbcommands.o define.o discard.o dropcmds.o \
+	collationcmds.o compressioncmds.o constraint.o conversioncmds.o copy.o \
+	createas.o dbcommands.o define.o discard.o dropcmds.o \
 	event_trigger.o explain.o extension.o foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o matview.o operatorcmds.o opclasscmds.o \
 	policy.o portalcmds.o prepare.o proclang.o publicationcmds.o \
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index eff325cc7d..eb49cfc54a 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -631,6 +631,7 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_ATTR_COMPRESSION:
 			/* ignore object types that don't have schema-qualified names */
 			break;
 
diff --git a/src/backend/commands/amcmds.c b/src/backend/commands/amcmds.c
index f2173450ad..8f69f9e33b 100644
--- a/src/backend/commands/amcmds.c
+++ b/src/backend/commands/amcmds.c
@@ -175,6 +175,70 @@ get_am_type_oid(const char *amname, char amtype, bool missing_ok)
 	return oid;
 }
 
+/*
+ * get_am_handler_oid - return access method handler Oid with additional
+ * type checking.
+ *
+ * If noerror is false, throw an error if access method not found or its
+ * type not equal to specified type, or its handler is not valid.
+ *
+ * If amtype is not '\0', an error is raised if the AM found is not of the
+ * given type.
+ */
+regproc
+get_am_handler_oid(Oid amOid, char amtype, bool noerror)
+{
+	HeapTuple	tup;
+	Form_pg_am	amform;
+	regproc		amhandler;
+
+	/* Get handler function OID for the access method */
+	tup = SearchSysCache1(AMOID, ObjectIdGetDatum(amOid));
+	if (!HeapTupleIsValid(tup))
+	{
+		if (noerror)
+			return InvalidOid;
+
+		elog(ERROR, "cache lookup failed for access method %u", amOid);
+	}
+
+	amform = (Form_pg_am) GETSTRUCT(tup);
+	amhandler = amform->amhandler;
+
+	if (amtype != '\0' && amform->amtype != amtype)
+	{
+		if (noerror)
+		{
+			ReleaseSysCache(tup);
+			return InvalidOid;
+		}
+
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("access method \"%s\" is not of type %s",
+						NameStr(amform->amname),
+						get_am_type_string(amtype))));
+	}
+
+	/* Complain if handler OID is invalid */
+	if (!RegProcedureIsValid(amhandler))
+	{
+		if (noerror)
+		{
+			ReleaseSysCache(tup);
+			return InvalidOid;
+		}
+
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("access method \"%s\" does not have a handler",
+						NameStr(amform->amname))));
+	}
+
+	ReleaseSysCache(tup);
+	return amhandler;
+}
+
 /*
  * get_index_am_oid - given an access method name, look up its OID
  *		and verify it corresponds to an index AM.
@@ -185,6 +249,16 @@ get_index_am_oid(const char *amname, bool missing_ok)
 	return get_am_type_oid(amname, AMTYPE_INDEX, missing_ok);
 }
 
+/*
+ * get_index_am_oid - given an access method name, look up its OID
+ *		and verify it corresponds to an index AM.
+ */
+Oid
+get_compression_am_oid(const char *amname, bool missing_ok)
+{
+	return get_am_type_oid(amname, AMTYPE_COMPRESSION, missing_ok);
+}
+
 /*
  * get_am_oid - given an access method name, look up its OID.
  *		The type is not checked.
@@ -225,6 +299,8 @@ get_am_type_string(char amtype)
 	{
 		case AMTYPE_INDEX:
 			return "INDEX";
+		case AMTYPE_COMPRESSION:
+			return "COMPRESSION";
 		default:
 			/* shouldn't happen */
 			elog(ERROR, "invalid access method type '%c'", amtype);
@@ -263,6 +339,14 @@ lookup_index_am_handler_func(List *handler_name, char amtype)
 								NameListToString(handler_name),
 								"index_am_handler")));
 			break;
+		case AMTYPE_COMPRESSION:
+			if (get_func_rettype(handlerOid) != COMPRESSION_AM_HANDLEROID)
+				ereport(ERROR,
+						(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+						 errmsg("function %s must return type %s",
+								NameListToString(handler_name),
+								"compression_am_handler")));
+			break;
 		default:
 			elog(ERROR, "unrecognized access method type \"%c\"", amtype);
 	}
diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c
new file mode 100644
index 0000000000..7841c7700a
--- /dev/null
+++ b/src/backend/commands/compressioncmds.c
@@ -0,0 +1,649 @@
+/*-------------------------------------------------------------------------
+ *
+ * compressioncmds.c
+ *	  Routines for SQL commands for compression access 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 "miscadmin.h"
+
+#include "access/cmapi.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/reloptions.h"
+#include "access/xact.h"
+#include "catalog/catalog.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_attr_compression.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_depend.h"
+#include "catalog/pg_proc.h"
+#include "catalog/pg_type.h"
+#include "commands/defrem.h"
+#include "parser/parse_func.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+#include "utils/snapmgr.h"
+
+/*
+ * When conditions of compression satisfies one if builtin attribute
+ * compresssion tuples the compressed attribute will be linked to
+ * builtin compression without new record in pg_attr_compression.
+ * So the fact that the column has a builtin compression we only can find out
+ * by its dependency.
+ */
+static void
+lookup_builtin_dependencies(Oid attrelid, AttrNumber attnum,
+							List **amoids)
+{
+	LOCKMODE	lock = AccessShareLock;
+	Oid			amoid = InvalidOid;
+	HeapTuple	tup;
+	Relation	rel;
+	SysScanDesc scan;
+	ScanKeyData key[3];
+
+	rel = heap_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 == AttrCompressionRelationId)
+		{
+			Assert(IsBuiltinCompression(depform->refobjid));
+			amoid = GetAttrCompressionAmOid(depform->refobjid);
+			*amoids = list_append_unique_oid(*amoids, amoid);
+		}
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, lock);
+}
+
+/*
+ * Find identical attribute compression for reuse and fill the list with
+ * used compression access methods.
+ */
+static Oid
+lookup_attribute_compression(Oid attrelid, AttrNumber attnum,
+							 Oid amoid, Datum acoptions, List **previous_amoids)
+{
+	Relation	rel;
+	HeapTuple	tuple;
+	SysScanDesc scan;
+	FmgrInfo	arrayeq_info;
+	Oid			result = InvalidOid;
+	ScanKeyData key[2];
+
+	/* fill FmgrInfo for array_eq function */
+	fmgr_info(F_ARRAY_EQ, &arrayeq_info);
+
+	Assert((attrelid > 0 && attnum > 0) || (attrelid == 0 && attnum == 0));
+	if (previous_amoids)
+		lookup_builtin_dependencies(attrelid, attnum, previous_amoids);
+
+	rel = heap_open(AttrCompressionRelationId, AccessShareLock);
+	ScanKeyInit(&key[0],
+				Anum_pg_attr_compression_acrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(attrelid));
+
+	ScanKeyInit(&key[1],
+				Anum_pg_attr_compression_acattnum,
+				BTEqualStrategyNumber, F_INT2EQ,
+				Int16GetDatum(attnum));
+
+	scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+							  true, NULL, 2, key);
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+	{
+		Oid			acoid,
+					tup_amoid;
+		Datum		values[Natts_pg_attr_compression];
+		bool		nulls[Natts_pg_attr_compression];
+
+		heap_deform_tuple(tuple, RelationGetDescr(rel), values, nulls);
+		acoid = DatumGetObjectId(values[Anum_pg_attr_compression_acoid - 1]);
+		tup_amoid = get_am_oid(
+							   NameStr(*DatumGetName(values[Anum_pg_attr_compression_acname - 1])), false);
+
+		if (previous_amoids)
+			*previous_amoids = list_append_unique_oid(*previous_amoids, tup_amoid);
+
+		if (tup_amoid != amoid)
+			continue;
+
+		/*
+		 * even if we found the match, we still need to acquire all previous
+		 * access methods so don't break cycle if previous_amoids is not NULL
+		 */
+		if (nulls[Anum_pg_attr_compression_acoptions - 1])
+		{
+			if (DatumGetPointer(acoptions) == NULL)
+				result = acoid;
+		}
+		else
+		{
+			bool		equal;
+
+			/* check if arrays for WITH options are equal */
+			equal = DatumGetBool(CallerFInfoFunctionCall2(
+														  array_eq,
+														  &arrayeq_info,
+														  InvalidOid,
+														  acoptions,
+														  values[Anum_pg_attr_compression_acoptions - 1]));
+			if (equal)
+				result = acoid;
+		}
+
+		if (previous_amoids == NULL && OidIsValid(result))
+			break;
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, AccessShareLock);
+	return result;
+}
+
+/*
+ * Link compression with an attribute. Creates a row in pg_attr_compression
+ * if needed.
+ *
+ * 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.
+ *
+ * If any of builtin attribute compression tuples satisfies conditions
+ * returns it.
+ *
+ * For ALTER command check for previous attribute compression record with
+ * identical compression options and reuse it if found any.
+ *
+ * Note we create attribute compression for EXTERNAL storage too, so when
+ * storage is changed we can start compression on future tuples right away.
+ */
+Oid
+CreateAttributeCompression(Form_pg_attribute att,
+						   ColumnCompression *compression,
+						   bool *need_rewrite, List **preserved_amoids)
+{
+	Relation	rel;
+	HeapTuple	newtup;
+	Oid			acoid = InvalidOid,
+				amoid;
+	regproc		amhandler;
+	Datum		arropt;
+	Datum		values[Natts_pg_attr_compression];
+	bool		nulls[Natts_pg_attr_compression];
+
+	ObjectAddress myself,
+				amref;
+
+	CompressionAmRoutine *routine;
+
+	/* No compression for PLAIN storage. */
+	if (att->attstorage == 'p')
+		return InvalidOid;
+
+	/* Fallback to default compression if it's not specified */
+	if (compression == NULL)
+		return DefaultCompressionOid;
+
+	amoid = get_compression_am_oid(compression->amname, false);
+	if (compression->options)
+		arropt = optionListToArray(compression->options, true);
+	else
+		arropt = PointerGetDatum(NULL);
+
+	/* Try to find builtin compression first */
+	acoid = lookup_attribute_compression(0, 0, amoid, arropt, NULL);
+
+	/*
+	 * attrelid will be invalid on CREATE TABLE, no need for table rewrite
+	 * check.
+	 */
+	if (OidIsValid(att->attrelid))
+	{
+		Oid			attacoid;
+		List	   *previous_amoids = NIL;
+
+		/*
+		 * Try to find identical compression from previous tuples, and fill
+		 * the list of previous compresssion methods
+		 */
+		attacoid = lookup_attribute_compression(att->attrelid, att->attnum,
+												amoid, arropt, &previous_amoids);
+		if (!OidIsValid(acoid))
+			acoid = attacoid;
+
+		/*
+		 * 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)
+		{
+			/* no rewrite by default */
+			*need_rewrite = false;
+
+			Assert(preserved_amoids != NULL);
+
+			if (compression->preserve == NIL)
+			{
+				Assert(!IsBinaryUpgrade);
+				*need_rewrite = true;
+			}
+			else
+			{
+				ListCell   *cell;
+
+				foreach(cell, compression->preserve)
+				{
+					char	   *amname_p = strVal(lfirst(cell));
+					Oid			amoid_p = get_am_oid(amname_p, false);
+
+					if (!list_member_oid(previous_amoids, amoid_p))
+						ereport(ERROR,
+								(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+								 errmsg("\"%s\" compression access method cannot be preserved", amname_p),
+								 errhint("use \"pg_column_compression\" function for list of compression methods")
+								 ));
+
+					*preserved_amoids = list_append_unique_oid(*preserved_amoids, amoid_p);
+
+					/*
+					 * Remove from previous list, also protect from multiple
+					 * mentions of one access method in PRESERVE list
+					 */
+					previous_amoids = list_delete_oid(previous_amoids, amoid_p);
+				}
+
+				/*
+				 * If the list of previous Oids is not empty after deletions
+				 * then we need to rewrite tuples in the table.
+				 *
+				 * In binary upgrade list will not be free since it contains
+				 * Oid of builtin compression access method.
+				 */
+				if (!IsBinaryUpgrade && list_length(previous_amoids) != 0)
+					*need_rewrite = true;
+			}
+		}
+
+		/* Cleanup */
+		list_free(previous_amoids);
+	}
+
+	if (IsBinaryUpgrade && !OidIsValid(acoid))
+		elog(ERROR, "could not restore attribute compression data");
+
+	/* Return Oid if we already found identical compression on this column */
+	if (OidIsValid(acoid))
+	{
+		if (DatumGetPointer(arropt) != NULL)
+			pfree(DatumGetPointer(arropt));
+
+		return acoid;
+	}
+
+	/* Initialize buffers for new tuple values */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+
+	amhandler = get_am_handler_oid(amoid, AMTYPE_COMPRESSION, false);
+
+	rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+
+	acoid = GetNewOidWithIndex(rel, AttrCompressionIndexId,
+							   Anum_pg_attr_compression_acoid);
+	if (acoid < FirstNormalObjectId)
+	{
+		/* this is database initialization */
+		heap_close(rel, RowExclusiveLock);
+		return DefaultCompressionOid;
+	}
+
+	/* we need routine only to call cmcheck function */
+	routine = InvokeCompressionAmHandler(amhandler);
+	if (routine->cmcheck != NULL)
+		routine->cmcheck(att, compression->options);
+	pfree(routine);
+
+	values[Anum_pg_attr_compression_acoid - 1] = ObjectIdGetDatum(acoid);
+	values[Anum_pg_attr_compression_acname - 1] = CStringGetDatum(compression->amname);
+	values[Anum_pg_attr_compression_acrelid - 1] = ObjectIdGetDatum(att->attrelid);
+	values[Anum_pg_attr_compression_acattnum - 1] = Int32GetDatum(att->attnum);
+
+	if (compression->options)
+		values[Anum_pg_attr_compression_acoptions - 1] = arropt;
+	else
+		nulls[Anum_pg_attr_compression_acoptions - 1] = true;
+
+	newtup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+	CatalogTupleInsert(rel, newtup);
+	heap_freetuple(newtup);
+	heap_close(rel, RowExclusiveLock);
+
+	ObjectAddressSet(myself, AttrCompressionRelationId, acoid);
+	ObjectAddressSet(amref, AccessMethodRelationId, amoid);
+
+	recordDependencyOn(&myself, &amref, DEPENDENCY_NORMAL);
+	recordDependencyOnCurrentExtension(&myself, false);
+
+	/* Make the changes visible */
+	CommandCounterIncrement();
+	return acoid;
+}
+
+/*
+ * Remove the attribute compression record from pg_attr_compression.
+ */
+void
+RemoveAttributeCompression(Oid acoid)
+{
+	Relation	relation;
+	HeapTuple	tup;
+
+	tup = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+	/* check we're not trying to remove builtin attribute compression */
+	Assert(((Form_pg_attr_compression) GETSTRUCT(tup))->acrelid != 0);
+
+	/* delete the record from catalogs */
+	relation = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+	CatalogTupleDelete(relation, &tup->t_self);
+	heap_close(relation, RowExclusiveLock);
+	ReleaseSysCache(tup);
+}
+
+/*
+ * CleanupAttributeCompression
+ *
+ * Remove entries in pg_attr_compression except current attribute compression
+ * and related with specified list of access methods.
+ */
+void
+CleanupAttributeCompression(Oid relid, AttrNumber attnum, List *keepAmOids)
+{
+	Oid			acoid,
+				amoid;
+	Relation	rel;
+	SysScanDesc scan;
+	ScanKeyData key[3];
+	HeapTuple	tuple,
+				attrtuple;
+	Form_pg_attribute attform;
+	List	   *removed = NIL;
+	ListCell   *lc;
+
+	attrtuple = SearchSysCache2(ATTNUM,
+								ObjectIdGetDatum(relid),
+								Int16GetDatum(attnum));
+
+	if (!HeapTupleIsValid(attrtuple))
+		elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+			 attnum, relid);
+	attform = (Form_pg_attribute) GETSTRUCT(attrtuple);
+	acoid = attform->attcompression;
+	ReleaseSysCache(attrtuple);
+
+	Assert(relid > 0 && attnum > 0);
+
+	if (IsBinaryUpgrade)
+		goto builtin_removal;
+
+	rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_attr_compression_acrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relid));
+
+	ScanKeyInit(&key[1],
+				Anum_pg_attr_compression_acattnum,
+				BTEqualStrategyNumber, F_INT2EQ,
+				Int16GetDatum(attnum));
+
+	scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+							  true, NULL, 2, key);
+
+	/* Remove attribute compression tuples and collect removed Oids to list */
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+	{
+		Form_pg_attr_compression acform;
+
+		acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+		amoid = get_am_oid(NameStr(acform->acname), false);
+
+		/* skip current compression */
+		if (acform->acoid == acoid)
+			continue;
+
+		if (!list_member_oid(keepAmOids, amoid))
+		{
+			removed = lappend_oid(removed, acform->acoid);
+			CatalogTupleDelete(rel, &tuple->t_self);
+		}
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, RowExclusiveLock);
+
+	/* Now remove dependencies */
+	rel = heap_open(DependRelationId, RowExclusiveLock);
+	foreach(lc, removed)
+	{
+		Oid			tup_acoid = lfirst_oid(lc);
+
+		ScanKeyInit(&key[0],
+					Anum_pg_depend_classid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(AttrCompressionRelationId));
+		ScanKeyInit(&key[1],
+					Anum_pg_depend_objid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(tup_acoid));
+
+		scan = systable_beginscan(rel, DependDependerIndexId, true,
+								  NULL, 2, key);
+
+		while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+			CatalogTupleDelete(rel, &tuple->t_self);
+
+		systable_endscan(scan);
+	}
+	heap_close(rel, RowExclusiveLock);
+
+	/* Now remove dependencies with builtin compressions */
+	rel = heap_open(DependRelationId, RowExclusiveLock);
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_classid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationRelationId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_objid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relid));
+	ScanKeyInit(&key[2],
+				Anum_pg_depend_objsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum((int32) attnum));
+
+	scan = systable_beginscan(rel, DependDependerIndexId, true,
+							  NULL, 3, key);
+
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tuple);
+
+		if (depform->refclassid != AttrCompressionRelationId)
+			continue;
+
+		/* skip current compression */
+		if (depform->refobjid == acoid)
+			continue;
+
+		amoid = GetAttrCompressionAmOid(depform->refobjid);
+		if (!list_member_oid(keepAmOids, amoid))
+			CatalogTupleDelete(rel, &tuple->t_self);
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, RowExclusiveLock);
+}
+
+/*
+ * Construct ColumnCompression node by attribute compression Oid.
+ */
+ColumnCompression *
+MakeColumnCompression(Oid acoid)
+{
+	HeapTuple	tuple;
+	Form_pg_attr_compression acform;
+	ColumnCompression *node;
+
+	if (!OidIsValid(acoid))
+		return NULL;
+
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+	acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+	node = makeNode(ColumnCompression);
+	node->amname = pstrdup(NameStr(acform->acname));
+	ReleaseSysCache(tuple);
+
+	/*
+	 * fill attribute compression options too. We could've do it above but
+	 * it's easier to call this helper.
+	 */
+	node->options = GetAttrCompressionOptions(acoid);
+
+	return node;
+}
+
+/*
+ * Compare compression options for two columns.
+ */
+void
+CheckCompressionMismatch(ColumnCompression *c1, ColumnCompression *c2,
+						 const char *attributeName)
+{
+	if (strcmp(c1->amname, c2->amname))
+		ereport(ERROR,
+				(errcode(ERRCODE_DATATYPE_MISMATCH),
+				 errmsg("column \"%s\" has a compression method conflict",
+						attributeName),
+				 errdetail("%s versus %s", c1->amname, c2->amname)));
+
+	if (!equal(c1->options, c2->options))
+		ereport(ERROR,
+				(errcode(ERRCODE_DATATYPE_MISMATCH),
+				 errmsg("column \"%s\" has a compression options conflict",
+						attributeName),
+				 errdetail("(%s) versus (%s)",
+						   formatRelOptions(c1->options),
+						   formatRelOptions(c2->options))));
+}
+
+/*
+ * Return list of compression methods used in specified column.
+ */
+Datum
+pg_column_compression(PG_FUNCTION_ARGS)
+{
+	Oid			relOid = PG_GETARG_OID(0);
+	char	   *attname = TextDatumGetCString(PG_GETARG_TEXT_P(1));
+	Relation	rel;
+	HeapTuple	tuple;
+	AttrNumber	attnum;
+	List	   *amoids = NIL;
+	Oid			amoid;
+	ListCell   *lc;
+
+	ScanKeyData key[2];
+	SysScanDesc scan;
+	StringInfoData result;
+
+	attnum = get_attnum(relOid, attname);
+	if (attnum == InvalidAttrNumber)
+		PG_RETURN_NULL();
+
+	/* Collect related builtin compression access methods */
+	lookup_builtin_dependencies(relOid, attnum, &amoids);
+
+	/* Collect other related access methods */
+	rel = heap_open(AttrCompressionRelationId, AccessShareLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_attr_compression_acrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relOid));
+	ScanKeyInit(&key[1],
+				Anum_pg_attr_compression_acattnum,
+				BTEqualStrategyNumber, F_INT2EQ,
+				Int16GetDatum(attnum));
+
+	scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+							  true, NULL, 2, key);
+	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+	{
+		Form_pg_attr_compression acform;
+
+		acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+		amoid = get_am_oid(NameStr(acform->acname), false);
+		amoids = list_append_unique_oid(amoids, amoid);
+	}
+
+	systable_endscan(scan);
+	heap_close(rel, AccessShareLock);
+
+	if (!list_length(amoids))
+		PG_RETURN_NULL();
+
+	/* Construct the list separated by comma */
+	amoid = InvalidOid;
+	initStringInfo(&result);
+	foreach(lc, amoids)
+	{
+		if (OidIsValid(amoid))
+			appendStringInfoString(&result, ", ");
+
+		amoid = lfirst_oid(lc);
+		appendStringInfoString(&result, get_am_name(amoid));
+	}
+
+	PG_RETURN_TEXT_P(CStringGetTextDatum(result.data));
+}
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index 20a3a78692..20eeee60cc 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -1209,6 +1209,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_PUBLICATION_REL:
 		case OCLASS_SUBSCRIPTION:
 		case OCLASS_TRANSFORM:
+		case OCLASS_ATTR_COMPRESSION:
 			return true;
 
 			/*
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index e5dd9958a4..efd0007a04 100644
--- a/src/backend/commands/foreigncmds.c
+++ b/src/backend/commands/foreigncmds.c
@@ -48,50 +48,6 @@ typedef struct
 /* Internal functions */
 static void import_error_callback(void *arg);
 
-
-/*
- * Convert a DefElem list to the text array format that is used in
- * pg_foreign_data_wrapper, pg_foreign_server, pg_user_mapping, and
- * pg_foreign_table.
- *
- * Returns the array in the form of a Datum, or PointerGetDatum(NULL)
- * if the list is empty.
- *
- * Note: The array is usually stored to database without further
- * processing, hence any validation should be done before this
- * conversion.
- */
-static Datum
-optionListToArray(List *options)
-{
-	ArrayBuildState *astate = NULL;
-	ListCell   *cell;
-
-	foreach(cell, options)
-	{
-		DefElem    *def = lfirst(cell);
-		const char *value;
-		Size		len;
-		text	   *t;
-
-		value = defGetString(def);
-		len = VARHDRSZ + strlen(def->defname) + 1 + strlen(value);
-		t = palloc(len + 1);
-		SET_VARSIZE(t, len);
-		sprintf(VARDATA(t), "%s=%s", def->defname, value);
-
-		astate = accumArrayResult(astate, PointerGetDatum(t),
-								  false, TEXTOID,
-								  CurrentMemoryContext);
-	}
-
-	if (astate)
-		return makeArrayResult(astate, CurrentMemoryContext);
-
-	return PointerGetDatum(NULL);
-}
-
-
 /*
  * Transform a list of DefElem into text array format.  This is substantially
  * the same thing as optionListToArray(), except we recognize SET/ADD/DROP
@@ -178,7 +134,7 @@ transformGenericOptions(Oid catalogId,
 		}
 	}
 
-	result = optionListToArray(resultOptions);
+	result = optionListToArray(resultOptions, false);
 
 	if (OidIsValid(fdwvalidator))
 	{
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 153aec263e..f18b97add1 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -22,6 +22,7 @@
 #include "access/relscan.h"
 #include "access/sysattr.h"
 #include "access/tupconvert.h"
+#include "access/tuptoaster.h"
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "catalog/catalog.h"
@@ -33,6 +34,7 @@
 #include "catalog/objectaccess.h"
 #include "catalog/partition.h"
 #include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_depend.h"
@@ -40,6 +42,7 @@
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_opclass.h"
+#include "catalog/pg_proc.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_trigger.h"
 #include "catalog/pg_type.h"
@@ -89,6 +92,7 @@
 #include "storage/smgr.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
+#include "utils/datum.h"
 #include "utils/fmgroids.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
@@ -370,6 +374,8 @@ 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 acoid);
+static void set_column_compression_relid(Oid acoid, Oid relid);
 static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
 			  AlterTableCmd *cmd, LOCKMODE lockmode);
 static void ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
@@ -468,6 +474,8 @@ static void ATExecGenericOptions(Relation rel, List *options);
 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, ColumnCompression *compression, LOCKMODE lockmode);
 
 static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
 				   ForkNumber forkNum, char relpersistence);
@@ -522,6 +530,7 @@ ObjectAddress
 DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			   ObjectAddress *typaddress, const char *queryString)
 {
+	int			i;
 	char		relname[NAMEDATALEN];
 	Oid			namespaceId;
 	Oid			relationId;
@@ -696,6 +705,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 	 * We can set the atthasdef flags now in the tuple descriptor; this just
 	 * saves StoreAttrDefault from having to do an immediate update of the
 	 * pg_attribute rows.
+	 *
+	 * Also create attribute compression records if needed.
 	 */
 	rawDefaults = NIL;
 	cookedDefaults = NIL;
@@ -742,6 +753,13 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 
 		if (colDef->identity)
 			attr->attidentity = colDef->identity;
+
+		if (relkind == RELKIND_RELATION || relkind == RELKIND_PARTITIONED_TABLE)
+			attr->attcompression = CreateAttributeCompression(attr,
+															  colDef->compression,
+															  NULL, NULL);
+		else
+			attr->attcompression = InvalidOid;
 	}
 
 	/*
@@ -787,6 +805,23 @@ 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))
+		{
+			set_column_compression_relid(attr->attcompression, relationId);
+			add_column_compression_dependency(attr->attrelid, attr->attnum,
+											  attr->attcompression);
+		}
+	}
+
 	/* Process and store partition bound, if any. */
 	if (stmt->partbound)
 	{
@@ -2147,6 +2182,19 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 									   storage_name(def->storage),
 									   storage_name(attribute->attstorage))));
 
+				if (OidIsValid(attribute->attcompression))
+				{
+					ColumnCompression *compression =
+					MakeColumnCompression(attribute->attcompression);
+
+					if (!def->compression)
+						def->compression = compression;
+					else
+						CheckCompressionMismatch(def->compression,
+												 compression,
+												 attributeName);
+				}
+
 				def->inhcount++;
 				/* Merge of NOT NULL constraints = OR 'em together */
 				def->is_not_null |= attribute->attnotnull;
@@ -2174,6 +2222,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->collOid = attribute->attcollation;
 				def->constraints = NIL;
 				def->location = -1;
+				def->compression = MakeColumnCompression(attribute->attcompression);
 				inhSchema = lappend(inhSchema, def);
 				newattno[parent_attno - 1] = ++child_attno;
 			}
@@ -2383,6 +2432,13 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 									   storage_name(def->storage),
 									   storage_name(newdef->storage))));
 
+				if (!def->compression)
+					def->compression = newdef->compression;
+				else if (newdef->compression)
+					CheckCompressionMismatch(def->compression,
+											 newdef->compression,
+											 attributeName);
+
 				/* Mark the column as locally defined */
 				def->is_local = true;
 				/* Merge of NOT NULL constraints = OR 'em together */
@@ -3464,6 +3520,7 @@ AlterTableGetLockLevel(List *cmds)
 				 */
 			case AT_GenericOptions:
 			case AT_AlterColumnGenericOptions:
+			case AT_SetCompression:
 				cmd_lockmode = AccessExclusiveLock;
 				break;
 
@@ -3939,6 +3996,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
 		case AT_DisableRowSecurity:
 		case AT_ForceRowSecurity:
 		case AT_NoForceRowSecurity:
+		case AT_SetCompression:
 			ATSimplePermissions(rel, ATT_TABLE);
 			/* These commands never recurse */
 			/* No command-specific prep needed */
@@ -4313,6 +4371,11 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
 			Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
 			ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name);
 			break;
+		case AT_SetCompression:
+			address = ATExecSetCompression(tab, rel, cmd->name,
+										   (ColumnCompression *) cmd->def,
+										   lockmode);
+			break;
 		default:				/* oops */
 			elog(ERROR, "unrecognized alter table type: %d",
 				 (int) cmd->subtype);
@@ -4378,8 +4441,9 @@ ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode)
 
 		/*
 		 * We only need to rewrite the table if at least one column needs to
-		 * be recomputed, we are adding/removing the OID column, or we are
-		 * changing its persistence.
+		 * be recomputed, we are adding/removing the OID column, we are
+		 * changing its persistence, or we are changing compression type of a
+		 * column without preserving old ones.
 		 *
 		 * There are two reasons for requiring a rewrite when changing
 		 * persistence: on one hand, we need to ensure that the buffers
@@ -5540,6 +5604,16 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
 	attribute.attislocal = colDef->is_local;
 	attribute.attinhcount = colDef->inhcount;
 	attribute.attcollation = collOid;
+
+	/* create attribute compresssion record */
+	if (rel->rd_rel->relkind == RELKIND_RELATION ||
+		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+		attribute.attcompression = CreateAttributeCompression(&attribute,
+															  colDef->compression,
+															  NULL, NULL);
+	else
+		attribute.attcompression = InvalidOid;
+
 	/* attribute.attacl is handled by InsertPgAttributeTuple */
 
 	ReleaseSysCache(typeTuple);
@@ -5714,6 +5788,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
@@ -5836,6 +5911,30 @@ add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid)
 	recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
 }
 
+/* Update acrelid in pg_attr_compression */
+static void
+set_column_compression_relid(Oid acoid, Oid relid)
+{
+	Relation	rel;
+	HeapTuple	tuple;
+	Form_pg_attr_compression acform;
+
+	Assert(OidIsValid(relid));
+	if (IsBuiltinCompression(acoid))
+		return;
+
+	rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+	tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+	acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+	acform->acrelid = relid;
+	CatalogTupleUpdate(rel, &tuple->t_self, tuple);
+	ReleaseSysCache(tuple);
+	heap_close(rel, RowExclusiveLock);
+}
+
 /*
  * Install a column's dependency on its collation.
  */
@@ -5858,6 +5957,106 @@ add_column_collation_dependency(Oid relid, int32 attnum, Oid collid)
 	}
 }
 
+/*
+ * Install a dependency for compression options on its column.
+ *
+ * If it is builtin attribute compression then column depends on it. This
+ * is used to determine connection between column and builtin attribute
+ * 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 acoid)
+{
+	bool		found = false;
+	Relation	depRel;
+	ObjectAddress acref,
+				attref;
+	HeapTuple	depTup;
+	ScanKeyData key[3];
+	SysScanDesc scan;
+
+	if (!OidIsValid(acoid))
+		return;
+
+	Assert(relid > 0 && attnum > 0);
+	ObjectAddressSet(acref, AttrCompressionRelationId, acoid);
+	ObjectAddressSubSet(attref, RelationRelationId, relid, attnum);
+
+	depRel = heap_open(DependRelationId, AccessShareLock);
+
+	if (IsBuiltinCompression(acoid))
+	{
+		ScanKeyInit(&key[0],
+					Anum_pg_depend_classid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(RelationRelationId));
+		ScanKeyInit(&key[1],
+					Anum_pg_depend_objid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		ScanKeyInit(&key[2],
+					Anum_pg_depend_objsubid,
+					BTEqualStrategyNumber, F_INT4EQ,
+					Int32GetDatum((int32) attnum));
+
+		scan = systable_beginscan(depRel, DependDependerIndexId, true,
+								  NULL, 3, key);
+
+		while (HeapTupleIsValid(depTup = systable_getnext(scan)))
+		{
+			Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(depTup);
+
+			if (foundDep->refclassid == AttrCompressionRelationId &&
+				foundDep->refobjid == acoid)
+			{
+				found = true;
+				break;
+			}
+		}
+		systable_endscan(scan);
+
+		if (!found)
+			recordDependencyOn(&attref, &acref, DEPENDENCY_NORMAL);
+	}
+	else
+	{
+		ScanKeyInit(&key[0],
+					Anum_pg_depend_refclassid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(RelationRelationId));
+		ScanKeyInit(&key[1],
+					Anum_pg_depend_refobjid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+		ScanKeyInit(&key[2],
+					Anum_pg_depend_refobjsubid,
+					BTEqualStrategyNumber, F_INT4EQ,
+					Int32GetDatum((int32) attnum));
+
+		scan = systable_beginscan(depRel, DependReferenceIndexId, true,
+								  NULL, 3, key);
+
+		while (HeapTupleIsValid(depTup = systable_getnext(scan)))
+		{
+			Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(depTup);
+
+			if (foundDep->classid == AttrCompressionRelationId &&
+				foundDep->objid == acoid)
+			{
+				found = true;
+				break;
+			}
+		}
+		systable_endscan(scan);
+
+		if (!found)
+			recordDependencyOn(&acref, &attref, DEPENDENCY_INTERNAL);
+	}
+	heap_close(depRel, AccessShareLock);
+}
+
 /*
  * ALTER TABLE SET WITH OIDS
  *
@@ -6702,6 +6901,10 @@ ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE loc
 				 errmsg("column data type %s can only have storage PLAIN",
 						format_type_be(attrtuple->atttypid))));
 
+	/* use default compression if storage is not PLAIN */
+	if (!OidIsValid(attrtuple->attcompression) && (newstorage != 'p'))
+		attrtuple->attcompression = DefaultCompressionOid;
+
 	CatalogTupleUpdate(attrelation, &tuple->t_self, tuple);
 
 	InvokeObjectPostAlterHook(RelationRelationId,
@@ -9586,6 +9789,12 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 								   colName)));
 				break;
 
+			case OCLASS_ATTR_COMPRESSION:
+
+				/* Just check that dependency is the right type */
+				Assert(foundDep->deptype == DEPENDENCY_INTERNAL);
+				break;
+
 			case OCLASS_DEFAULT:
 
 				/*
@@ -9686,7 +9895,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 		if (!(foundDep->refclassid == TypeRelationId &&
 			  foundDep->refobjid == attTup->atttypid) &&
 			!(foundDep->refclassid == CollationRelationId &&
-			  foundDep->refobjid == attTup->attcollation))
+			  foundDep->refobjid == attTup->attcollation) &&
+			foundDep->refclassid != AttrCompressionRelationId)
 			elog(ERROR, "found unexpected dependency for column");
 
 		CatalogTupleDelete(depRel, &depTup->t_self);
@@ -9711,6 +9921,34 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 
 	ReleaseSysCache(typeTuple);
 
+	if (rel->rd_rel->relkind == RELKIND_RELATION ||
+		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		/* Setup attribute compression for new attribute */
+		if (OidIsValid(attTup->attcompression) && attTup->attstorage == 'p')
+		{
+			/* Set up rewrite of table */
+			attTup->attcompression = InvalidOid;
+			/* TODO: call the rewrite function here */
+		}
+		else if (OidIsValid(attTup->attcompression))
+		{
+			/* create new record */
+			ColumnCompression *compression = MakeColumnCompression(attTup->attcompression);
+
+			if (!IsBuiltinCompression(attTup->attcompression))
+				/* TODO: call the rewrite function here */;
+
+			attTup->attcompression = CreateAttributeCompression(attTup, compression, NULL, NULL);
+		}
+		else if (attTup->attstorage != 'p')
+			attTup->attcompression = DefaultCompressionOid;
+		else
+			attTup->attcompression = InvalidOid;
+	}
+	else
+		attTup->attcompression = InvalidOid;
+
 	CatalogTupleUpdate(attrelation, &heapTup->t_self, heapTup);
 
 	heap_close(attrelation, RowExclusiveLock);
@@ -9719,6 +9957,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
 	 */
@@ -12742,6 +12985,93 @@ ATExecGenericOptions(Relation rel, List *options)
 	heap_freetuple(tuple);
 }
 
+static ObjectAddress
+ATExecSetCompression(AlteredTableInfo *tab,
+					 Relation rel,
+					 const char *column,
+					 ColumnCompression *compression,
+					 LOCKMODE lockmode)
+{
+	Oid			acoid;
+	Relation	attrel;
+	HeapTuple	atttuple;
+	Form_pg_attribute atttableform;
+	AttrNumber	attnum;
+	bool		need_rewrite;
+	Datum		values[Natts_pg_attribute];
+	bool		nulls[Natts_pg_attribute];
+	bool		replace[Natts_pg_attribute];
+	List	   *preserved_amoids = NIL;
+	ObjectAddress address;
+
+	attrel = heap_open(AttributeRelationId, RowExclusiveLock);
+
+	atttuple = SearchSysCacheAttName(RelationGetRelid(rel), column);
+	if (!HeapTupleIsValid(atttuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_COLUMN),
+				 errmsg("column \"%s\" of relation \"%s\" does not exist",
+						column, RelationGetRelationName(rel))));
+
+	/* Prevent them from altering a system attribute */
+	atttableform = (Form_pg_attribute) GETSTRUCT(atttuple);
+	attnum = atttableform->attnum;
+	if (attnum <= 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("cannot alter system column \"%s\"", column)));
+
+	/* Prevent from altering untoastable attributes with PLAIN storage */
+	if (atttableform->attstorage == 'p' && !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));
+
+	/* create pg_attr_compression record and its dependencies */
+	acoid = CreateAttributeCompression(atttableform, compression,
+									   &need_rewrite, &preserved_amoids);
+	add_column_compression_dependency(atttableform->attrelid,
+									  atttableform->attnum, acoid);
+
+	/*
+	 * Save list of preserved access methods to cache which will be passed to
+	 * toast_insert_or_update
+	 */
+	if (need_rewrite)
+		/* TODO: set up rewrite rules here */;
+
+	atttableform->attcompression = acoid;
+	CatalogTupleUpdate(attrel, &atttuple->t_self, atttuple);
+
+	InvokeObjectPostAlterHook(RelationRelationId,
+							  RelationGetRelid(rel),
+							  atttableform->attnum);
+
+	ReleaseSysCache(atttuple);
+	heap_close(attrel, RowExclusiveLock);
+
+	/* make changes visible */
+	CommandCounterIncrement();
+
+	/*
+	 * Normally cleanup is done in rewrite but in binary upgrade we should do
+	 * it explicitly.
+	 */
+	if (IsBinaryUpgrade)
+		CleanupAttributeCompression(RelationGetRelid(rel),
+									attnum, preserved_amoids);
+
+	ObjectAddressSet(address, AttrCompressionRelationId, acoid);
+	return address;
+}
+
+
 /*
  * Preparation phase for SET LOGGED/UNLOGGED
  *
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index e8ea59e34a..120245084c 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2869,6 +2869,7 @@ _copyColumnDef(const ColumnDef *from)
 
 	COPY_STRING_FIELD(colname);
 	COPY_NODE_FIELD(typeName);
+	COPY_NODE_FIELD(compression);
 	COPY_SCALAR_FIELD(inhcount);
 	COPY_SCALAR_FIELD(is_local);
 	COPY_SCALAR_FIELD(is_not_null);
@@ -2888,6 +2889,18 @@ _copyColumnDef(const ColumnDef *from)
 	return newnode;
 }
 
+static ColumnCompression *
+_copyColumnCompression(const ColumnCompression *from)
+{
+	ColumnCompression *newnode = makeNode(ColumnCompression);
+
+	COPY_STRING_FIELD(amname);
+	COPY_NODE_FIELD(options);
+	COPY_NODE_FIELD(preserve);
+
+	return newnode;
+}
+
 static Constraint *
 _copyConstraint(const Constraint *from)
 {
@@ -5567,6 +5580,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 3bb91c9595..3ddfe063af 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2549,6 +2549,7 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 {
 	COMPARE_STRING_FIELD(colname);
 	COMPARE_NODE_FIELD(typeName);
+	COMPARE_NODE_FIELD(compression);
 	COMPARE_SCALAR_FIELD(inhcount);
 	COMPARE_SCALAR_FIELD(is_local);
 	COMPARE_SCALAR_FIELD(is_not_null);
@@ -2568,6 +2569,16 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
 	return true;
 }
 
+static bool
+_equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b)
+{
+	COMPARE_STRING_FIELD(amname);
+	COMPARE_NODE_FIELD(options);
+	COMPARE_NODE_FIELD(preserve);
+
+	return true;
+}
+
 static bool
 _equalConstraint(const Constraint *a, const Constraint *b)
 {
@@ -3628,6 +3639,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/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c
index a10014f755..07ec871980 100644
--- a/src/backend/nodes/nodeFuncs.c
+++ b/src/backend/nodes/nodeFuncs.c
@@ -3648,6 +3648,8 @@ raw_expression_tree_walker(Node *node,
 
 				if (walker(coldef->typeName, context))
 					return true;
+				if (walker(coldef->compression, context))
+					return true;
 				if (walker(coldef->raw_default, context))
 					return true;
 				if (walker(coldef->collClause, context))
@@ -3655,6 +3657,14 @@ raw_expression_tree_walker(Node *node,
 				/* for now, constraints are ignored */
 			}
 			break;
+		case T_ColumnCompression:
+			{
+				ColumnCompression *colcmp = (ColumnCompression *) node;
+
+				if (walker(colcmp->options, context))
+					return true;
+			}
+			break;
 		case T_IndexElem:
 			{
 				IndexElem  *indelem = (IndexElem *) node;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 69731ccdea..afe7319883 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2870,6 +2870,7 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
 
 	WRITE_STRING_FIELD(colname);
 	WRITE_NODE_FIELD(typeName);
+	WRITE_NODE_FIELD(compression);
 	WRITE_INT_FIELD(inhcount);
 	WRITE_BOOL_FIELD(is_local);
 	WRITE_BOOL_FIELD(is_not_null);
@@ -2887,6 +2888,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(amname);
+	WRITE_NODE_FIELD(options);
+	WRITE_NODE_FIELD(preserve);
+}
+
 static void
 _outTypeName(StringInfo str, const TypeName *node)
 {
@@ -4211,6 +4222,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/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index a6a2de94ea..e296a45299 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1062,6 +1062,13 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		else
 			def->storage = 0;
 
+		/* Likewise, copy compression if requested */
+		if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION
+			&& OidIsValid(attribute->attcompression))
+			def->compression = MakeColumnCompression(attribute->attcompression);
+		else
+			def->compression = NULL;
+
 		/* Likewise, copy comment if requested */
 		if ((table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS) &&
 			(comment = GetComment(attribute->attrelid,
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index bed63c768e..9eb8e05dbe 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -3123,7 +3123,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 				   VARSIZE(chunk) - VARHDRSZ);
 			data_done += VARSIZE(chunk) - VARHDRSZ;
 		}
-		Assert(data_done == toast_pointer.va_extsize);
+		Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
 
 		/* make sure its marked as compressed or not */
 		if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
diff --git a/src/backend/utils/adt/pseudotypes.c b/src/backend/utils/adt/pseudotypes.c
index dbe67cdb4c..4a97099cd0 100644
--- a/src/backend/utils/adt/pseudotypes.c
+++ b/src/backend/utils/adt/pseudotypes.c
@@ -418,3 +418,4 @@ PSEUDOTYPE_DUMMY_IO_FUNCS(internal);
 PSEUDOTYPE_DUMMY_IO_FUNCS(opaque);
 PSEUDOTYPE_DUMMY_IO_FUNCS(anyelement);
 PSEUDOTYPE_DUMMY_IO_FUNCS(anynonarray);
+PSEUDOTYPE_DUMMY_IO_FUNCS(compression_am_handler);
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 2b381782a3..0c97851cbd 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -27,6 +27,7 @@
 #include "catalog/pg_am.h"
 #include "catalog/pg_amop.h"
 #include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
 #include "catalog/pg_auth_members.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_cast.h"
@@ -187,6 +188,17 @@ static const struct cachedesc cacheinfo[] = {
 		},
 		16
 	},
+	{AttrCompressionRelationId, /* ATTCOMPRESSIONOID */
+		AttrCompressionIndexId,
+		1,
+		{
+			Anum_pg_attr_compression_acoid,
+			0,
+			0,
+			0
+		},
+		8,
+	},
 	{AttributeRelationId,		/* ATTNAME */
 		AttributeRelidNameIndexId,
 		2,
diff --git a/src/include/access/cmapi.h b/src/include/access/cmapi.h
new file mode 100644
index 0000000000..9e48f0d49f
--- /dev/null
+++ b/src/include/access/cmapi.h
@@ -0,0 +1,80 @@
+/*-------------------------------------------------------------------------
+ *
+ * cmapi.h
+ *	  API for Postgres compression AM.
+ *
+ * Copyright (c) 2015-2017, PostgreSQL Global Development Group
+ *
+ * src/include/access/cmapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef CMAPI_H
+#define CMAPI_H
+
+#include "postgres.h"
+#include "catalog/pg_attr_compression.h"
+#include "catalog/pg_attribute.h"
+#include "nodes/pg_list.h"
+
+#define IsBuiltinCompression(cmid)	((cmid) < FirstBootstrapObjectId)
+#define DefaultCompressionOid		(InvalidOid)
+
+typedef struct CompressionAmRoutine CompressionAmRoutine;
+
+/*
+ * CompressionAmOptions contains all information needed to compress varlena.
+ *
+ *  For optimization purposes it will be created once for each attribute
+ *  compression and stored in cache, until its renewal on global cache reset,
+ *  or until deletion of related attribute compression.
+ */
+typedef struct CompressionAmOptions
+{
+	Oid			acoid;			/* Oid of attribute compression,
+								   should go always first */
+	Oid			amoid;			/* Oid of compression access method */
+	List	   *acoptions;		/* Parsed options, used for comparison */
+	CompressionAmRoutine *amroutine;	/* compression access method routine */
+	MemoryContext	mcxt;
+
+	/* result of cminitstate function will be put here */
+	void	   *acstate;
+} CompressionAmOptions;
+
+typedef void (*cmcheck_function) (Form_pg_attribute att, List *options);
+typedef struct varlena *(*cmcompress_function)
+			(CompressionAmOptions *cmoptions, const struct varlena *value);
+typedef void *(*cminitstate_function) (Oid acoid, List *options);
+
+/*
+ * API struct for a compression AM.
+ *
+ * 'cmcheck' - called when attribute is linking with compression method.
+ *  This function should check compability of compression method with
+ *  the attribute and its options.
+ *
+ * 'cminitstate' - called when CompressionAmOptions instance is created.
+ *  Should return pointer to a memory in a caller memory context, or NULL.
+ *  Could be used to pass some internal state between compression function
+ *  calls, like internal structure for parsed compression options.
+ *
+ * 'cmcompress' and 'cmdecompress' - varlena compression functions.
+ */
+struct CompressionAmRoutine
+{
+	NodeTag		type;
+
+	cmcheck_function cmcheck;	/* can be NULL */
+	cminitstate_function cminitstate;	/* can be NULL */
+	cmcompress_function cmcompress;
+	cmcompress_function cmdecompress;
+};
+
+/* access/compression/cmapi.c */
+extern CompressionAmRoutine *InvokeCompressionAmHandler(Oid amhandler);
+extern List *GetAttrCompressionOptions(Oid acoid);
+extern Oid	GetAttrCompressionAmOid(Oid acoid);
+
+#endif							/* CMAPI_H */
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index 4022c14a83..c9302347e8 100644
--- a/src/include/access/reloptions.h
+++ b/src/include/access/reloptions.h
@@ -259,7 +259,6 @@ extern void add_string_reloption(bits32 kinds, const char *name, const char *des
 extern Datum transformRelOptions(Datum oldOptions, List *defList,
 					const char *namspace, char *validnsps[],
 					bool ignoreOids, bool isReset);
-extern List *untransformRelOptions(Datum options);
 extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
 				  amoptions_function amoptions);
 extern relopt_value *parseRelOptions(Datum options, bool validate,
@@ -270,6 +269,8 @@ extern void fillRelOptions(void *rdopts, Size basesize,
 			   relopt_value *options, int numoptions,
 			   bool validate,
 			   const relopt_parse_elt *elems, int nelems);
+extern char *formatRelOptions(List *options);
+extern List *untransformRelOptions(Datum options);
 
 extern bytea *default_reloptions(Datum reloptions, bool validate,
 				   relopt_kind kind);
diff --git a/src/include/access/tuptoaster.h b/src/include/access/tuptoaster.h
index f99291e30d..793367ff58 100644
--- a/src/include/access/tuptoaster.h
+++ b/src/include/access/tuptoaster.h
@@ -13,9 +13,11 @@
 #ifndef TUPTOASTER_H
 #define TUPTOASTER_H
 
+#include "access/cmapi.h"
 #include "access/htup_details.h"
 #include "storage/lockdefs.h"
 #include "utils/relcache.h"
+#include "utils/hsearch.h"
 
 /*
  * This enables de-toasting of index entries.  Needed until VACUUM is
@@ -101,6 +103,15 @@
 /* Size of an EXTERNAL datum that contains an indirection pointer */
 #define INDIRECT_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_indirect))
 
+/*
+ * va_extinfo in varatt_external contains actual length of the external data
+ * and optional flags
+ */
+#define VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer) \
+	((toast_pointer).va_extinfo & 0x3FFFFFFF)
+#define VARATT_EXTERNAL_IS_CUSTOM_COMPRESSED(toast_pointer) \
+	(((toast_pointer).va_extinfo >> 30) == 0x02)
+
 /*
  * Testing whether an externally-stored value is compressed now requires
  * comparing extsize (the actual length of the external data) to rawsize
@@ -109,7 +120,7 @@
  * saves space, so we expect either equality or less-than.
  */
 #define VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer) \
-	((toast_pointer).va_extsize < (toast_pointer).va_rawsize - VARHDRSZ)
+	(VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer) < (toast_pointer).va_rawsize - VARHDRSZ)
 
 /*
  * Macro to fetch the possibly-unaligned contents of an EXTERNAL datum
@@ -126,6 +137,16 @@ do { \
 	memcpy(&(toast_pointer), VARDATA_EXTERNAL(attre), sizeof(toast_pointer)); \
 } while (0)
 
+/*
+ * Used to pass to preserved compression access methods from
+ * ALTER TABLE SET COMPRESSION into toast_insert_or_update.
+ */
+typedef struct AttrCmPreservedInfo
+{
+	AttrNumber	attnum;
+	List	   *preserved_amoids;
+}			AttrCmPreservedInfo;
+
 /* ----------
  * toast_insert_or_update -
  *
@@ -134,7 +155,7 @@ do { \
  */
 extern HeapTuple toast_insert_or_update(Relation rel,
 					   HeapTuple newtup, HeapTuple oldtup,
-					   int options);
+					   int options, HTAB *preserved_cmlist);
 
 /* ----------
  * toast_delete -
@@ -210,7 +231,7 @@ extern HeapTuple toast_build_flattened_tuple(TupleDesc tupleDesc,
  *	Create a compressed version of a varlena datum, if possible
  * ----------
  */
-extern Datum toast_compress_datum(Datum value);
+extern Datum toast_compress_datum(Datum value, Oid cmoptoid);
 
 /* ----------
  * toast_raw_datum_size -
@@ -236,4 +257,19 @@ extern Size toast_datum_size(Datum value);
  */
 extern Oid	toast_get_valid_index(Oid toastoid, LOCKMODE lock);
 
+/*
+ * lookup_compression_am_options -
+ *
+ * Return cached CompressionAmOptions for specified attribute compression.
+ */
+extern CompressionAmOptions *lookup_compression_am_options(Oid acoid);
+
+/*
+ * toast_set_compressed_datum_info -
+ *
+ * Save metadata in compressed datum
+ */
+extern void toast_set_compressed_datum_info(struct varlena *val, Oid cmid,
+									 int32 rawsize);
+
 #endif							/* TUPTOASTER_H */
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index 46c271a46c..cb19cb3e41 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -180,10 +180,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_ATTR_COMPRESSION		/* pg_attr_compression */
 } ObjectClass;
 
-#define LAST_OCLASS		OCLASS_TRANSFORM
+#define LAST_OCLASS		OCLASS_ATTR_COMPRESSION
 
 /* flag bits for performDeletion/performMultipleDeletions: */
 #define PERFORM_DELETION_INTERNAL			0x0001	/* internal action */
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index 254fbef1f7..5f2270f93d 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -87,6 +87,11 @@ DECLARE_UNIQUE_INDEX(pg_attrdef_adrelid_adnum_index, 2656, on pg_attrdef using b
 DECLARE_UNIQUE_INDEX(pg_attrdef_oid_index, 2657, on pg_attrdef using btree(oid oid_ops));
 #define AttrDefaultOidIndexId  2657
 
+DECLARE_UNIQUE_INDEX(pg_attr_compression_index, 4003, on pg_attr_compression using btree(acoid oid_ops));
+#define AttrCompressionIndexId  4003
+DECLARE_INDEX(pg_attr_compression_relid_attnum_index, 4004, on pg_attr_compression using btree(acrelid oid_ops, acattnum int2_ops));
+#define AttrCompressionRelidAttnumIndexId  4004
+
 DECLARE_UNIQUE_INDEX(pg_attribute_relid_attnam_index, 2658, on pg_attribute using btree(attrelid oid_ops, attname name_ops));
 #define AttributeRelidNameIndexId  2658
 DECLARE_UNIQUE_INDEX(pg_attribute_relid_attnum_index, 2659, on pg_attribute using btree(attrelid oid_ops, attnum int2_ops));
diff --git a/src/include/catalog/pg_attr_compression.dat b/src/include/catalog/pg_attr_compression.dat
new file mode 100644
index 0000000000..30faae0de4
--- /dev/null
+++ b/src/include/catalog/pg_attr_compression.dat
@@ -0,0 +1,23 @@
+#----------------------------------------------------------------------
+#
+# pg_attr_compression.dat
+#    Initial contents of the pg_attr_compression system relation.
+#
+# Predefined compression options for builtin compression access methods.
+# It is safe to use Oids that equal to Oids of access methods, since
+# these are just numbers and not system Oids.
+#
+# Note that predefined options should have 0 in acrelid and acattnum.
+#
+# Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/include/catalog/pg_attr_compression.dat
+#
+#----------------------------------------------------------------------
+
+[
+
+
+
+]
diff --git a/src/include/catalog/pg_attr_compression.h b/src/include/catalog/pg_attr_compression.h
new file mode 100644
index 0000000000..563aa65a2a
--- /dev/null
+++ b/src/include/catalog/pg_attr_compression.h
@@ -0,0 +1,53 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_attr_compression.h
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_attr_compression.h
+ *
+ * NOTES
+ *	  The Catalog.pm module reads this file and derives schema
+ *	  information.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_ATTR_COMPRESSION_H
+#define PG_ATTR_COMPRESSION_H
+
+#include "catalog/genbki.h"
+#include "catalog/pg_am_d.h"
+#include "catalog/pg_attr_compression_d.h"
+
+/* ----------------
+ *		pg_attr_compression definition.  cpp turns this into
+ *		typedef struct FormData_pg_attr_compression
+ * ----------------
+ */
+CATALOG(pg_attr_compression,4001,AttrCompressionRelationId) BKI_WITHOUT_OIDS
+{
+	Oid			acoid;						/* attribute compression oid */
+	NameData	acname;						/* name of compression AM */
+	Oid			acrelid BKI_DEFAULT(0);		/* attribute relation */
+	int16		acattnum BKI_DEFAULT(0);	/* attribute number in the relation */
+
+#ifdef CATALOG_VARLEN			/* variable-length fields start here */
+	text		acoptions[1] BKI_DEFAULT(_null_);	/* specific options from WITH */
+#endif
+} FormData_pg_attr_compression;
+
+/* ----------------
+ *		Form_pg_attr_compresssion corresponds to a pointer to a tuple with
+ *		the format of pg_attr_compression relation.
+ * ----------------
+ */
+typedef FormData_pg_attr_compression *Form_pg_attr_compression;
+
+#ifdef EXPOSE_TO_CLIENT_CODE
+/* builtin attribute compression Oids */
+#define PGLZ_AC_OID (PGLZ_COMPRESSION_AM_OID)
+#define ZLIB_AC_OID (ZLIB_COMPRESSION_AM_OID)
+#endif
+
+#endif							/* PG_ATTR_COMPRESSION_H */
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index dc36753ede..08e074ea4d 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -160,6 +160,9 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_WITHOUT_OIDS BK
 	/* attribute's collation */
 	Oid			attcollation;
 
+	/* attribute's compression options  or InvalidOid */
+	Oid			attcompression BKI_DEFAULT(0);
+
 #ifdef CATALOG_VARLEN			/* variable-length fields start here */
 	/* NOTE: The following fields are not present in tuple descriptors. */
 
@@ -184,10 +187,10 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_WITHOUT_OIDS BK
  * ATTRIBUTE_FIXED_PART_SIZE is the size of the fixed-layout,
  * guaranteed-not-null part of a pg_attribute row.  This is in fact as much
  * of the row as gets copied into tuple descriptors, so don't expect you
- * can access fields beyond attcollation except in a real tuple!
+ * can access fields beyond attcompression except in a real tuple!
  */
 #define ATTRIBUTE_FIXED_PART_SIZE \
-	(offsetof(FormData_pg_attribute,attcollation) + sizeof(Oid))
+	(offsetof(FormData_pg_attribute,attcompression) + sizeof(Oid))
 
 /* ----------------
  *		Form_pg_attribute corresponds to a pointer to a tuple with
diff --git a/src/include/catalog/pg_class.dat b/src/include/catalog/pg_class.dat
index 9fffdef379..49643b934e 100644
--- a/src/include/catalog/pg_class.dat
+++ b/src/include/catalog/pg_class.dat
@@ -36,7 +36,7 @@
   reloftype => '0', relowner => 'PGUID', relam => '0', relfilenode => '0',
   reltablespace => '0', relpages => '0', reltuples => '0', relallvisible => '0',
   reltoastrelid => '0', relhasindex => 'f', relisshared => 'f',
-  relpersistence => 'p', relkind => 'r', relnatts => '24', relchecks => '0',
+  relpersistence => 'p', relkind => 'r', relnatts => '25', relchecks => '0',
   relhasoids => 'f', relhasrules => 'f', relhastriggers => 'f',
   relhassubclass => 'f', relrowsecurity => 'f', relforcerowsecurity => 'f',
   relispopulated => 't', relreplident => 'n', relispartition => 'f',
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 4026018ba9..02431b45ef 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -6740,6 +6740,9 @@
   descr => 'bytes required to store the value, perhaps with compression',
   proname => 'pg_column_size', provolatile => 's', prorettype => 'int4',
   proargtypes => 'any', prosrc => 'pg_column_size' },
+{ oid => '4008', descr => 'list of compression methods used by the column',
+  proname => 'pg_column_compression', provolatile => 's', prorettype => 'text',
+  proargtypes => 'regclass text', prosrc => 'pg_column_compression' },
 { oid => '2322',
   descr => 'total disk space usage for the specified tablespace',
   proname => 'pg_tablespace_size', provolatile => 'v', prorettype => 'int8',
@@ -6909,6 +6912,13 @@
 { oid => '3312', descr => 'I/O',
   proname => 'tsm_handler_out', prorettype => 'cstring',
   proargtypes => 'tsm_handler', prosrc => 'tsm_handler_out' },
+{ oid => '4006', descr => 'I/O',
+  proname => 'compression_am_handler_in', proisstrict => 'f',
+  prorettype => 'compression_am_handler', proargtypes => 'cstring',
+  prosrc => 'compression_am_handler_in' },
+{ oid => '4007', descr => 'I/O',
+  proname => 'compression_am_handler_out', prorettype => 'cstring',
+  proargtypes => 'compression_am_handler', prosrc => 'compression_am_handler_out' },
 
 # tablesample method handlers
 { oid => '3313', descr => 'BERNOULLI tablesample method handler',
diff --git a/src/include/catalog/pg_type.dat b/src/include/catalog/pg_type.dat
index d295eae1b9..0fb34cb680 100644
--- a/src/include/catalog/pg_type.dat
+++ b/src/include/catalog/pg_type.dat
@@ -574,6 +574,11 @@
   typcategory => 'P', typinput => 'index_am_handler_in',
   typoutput => 'index_am_handler_out', typreceive => '-', typsend => '-',
   typalign => 'i' },
+{ oid => '4005',
+  typname => 'compression_am_handler', typlen => '4', typbyval => 't', typtype => 'p',
+  typcategory => 'P', typinput => 'compression_am_handler_in',
+  typoutput => 'compression_am_handler_out', typreceive => '-', typsend => '-',
+  typalign => 'i' },
 { oid => '3310',
   descr => 'pseudo-type for the result of a tablesample method function',
   typname => 'tsm_handler', typlen => '4', typbyval => 't', typtype => 'p',
diff --git a/src/include/catalog/toasting.h b/src/include/catalog/toasting.h
index f259890e43..e5408e2763 100644
--- a/src/include/catalog/toasting.h
+++ b/src/include/catalog/toasting.h
@@ -71,6 +71,7 @@ DECLARE_TOAST(pg_trigger, 2336, 2337);
 DECLARE_TOAST(pg_ts_dict, 4169, 4170);
 DECLARE_TOAST(pg_type, 4171, 4172);
 DECLARE_TOAST(pg_user_mapping, 4173, 4174);
+DECLARE_TOAST(pg_attr_compression, 4187, 4188);
 
 /* shared catalogs */
 DECLARE_TOAST(pg_authid, 4175, 4176);
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 1d05a4bcdc..988986afce 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -147,6 +147,7 @@ extern Oid	RemoveUserMapping(DropUserMappingStmt *stmt);
 extern void RemoveUserMappingById(Oid umId);
 extern void CreateForeignTable(CreateForeignTableStmt *stmt, Oid relid);
 extern void ImportForeignSchema(ImportForeignSchemaStmt *stmt);
+extern Datum optionListToArray(List *options, bool sorted);
 extern Datum transformGenericOptions(Oid catalogId,
 						Datum oldOptions,
 						List *options,
@@ -156,8 +157,21 @@ extern Datum transformGenericOptions(Oid catalogId,
 extern ObjectAddress CreateAccessMethod(CreateAmStmt *stmt);
 extern void RemoveAccessMethodById(Oid amOid);
 extern Oid	get_index_am_oid(const char *amname, bool missing_ok);
+extern Oid	get_compression_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);
+extern regproc get_am_handler_oid(Oid amOid, char amtype, bool noerror);
+
+/* commands/compressioncmds.c */
+extern ColumnCompression *MakeColumnCompression(Oid acoid);
+extern Oid CreateAttributeCompression(Form_pg_attribute attr,
+						   ColumnCompression *compression,
+						   bool *need_rewrite,
+						   List **preserved_amoids);
+extern void RemoveAttributeCompression(Oid acoid);
+extern void CheckCompressionMismatch(ColumnCompression *c1,
+						 ColumnCompression *c2, const char *attributeName);
+void		CleanupAttributeCompression(Oid relid, AttrNumber attnum, List *keepAmOids);
 
 /* support routines in commands/define.c */
 
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index a392fcdd2c..95f1f935e7 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -506,7 +506,8 @@ typedef enum NodeTag
 	T_IndexAmRoutine,			/* in access/amapi.h */
 	T_TsmRoutine,				/* in access/tsmapi.h */
 	T_ForeignKeyCacheInfo,		/* in utils/rel.h */
-	T_CallContext				/* in nodes/parsenodes.h */
+	T_CallContext,				/* in nodes/parsenodes.h */
+	T_CompressionAmRoutine		/* in access/cmapi.h */
 } NodeTag;
 
 /*
diff --git a/src/include/postgres.h b/src/include/postgres.h
index b596fcb513..916c75c7d8 100644
--- a/src/include/postgres.h
+++ b/src/include/postgres.h
@@ -55,7 +55,7 @@
 /*
  * struct varatt_external is a traditional "TOAST pointer", that is, the
  * information needed to fetch a Datum stored out-of-line in a TOAST table.
- * The data is compressed if and only if va_extsize < va_rawsize - VARHDRSZ.
+ * The data is compressed if and only if size in va_extinfo < va_rawsize - VARHDRSZ.
  * This struct must not contain any padding, because we sometimes compare
  * these pointers using memcmp.
  *
@@ -67,7 +67,8 @@
 typedef struct varatt_external
 {
 	int32		va_rawsize;		/* Original data size (includes header) */
-	int32		va_extsize;		/* External saved size (doesn't) */
+	uint32		va_extinfo;		/* External saved size (without header) and
+								 * flags */
 	Oid			va_valueid;		/* Unique ID of value within TOAST table */
 	Oid			va_toastrelid;	/* RelID of TOAST table containing it */
 }			varatt_external;
@@ -145,9 +146,18 @@ typedef union
 	struct						/* Compressed-in-line format */
 	{
 		uint32		va_header;
-		uint32		va_rawsize; /* Original data size (excludes header) */
+		uint32		va_info;	/* Original data size (excludes header) and
+								 * flags */
 		char		va_data[FLEXIBLE_ARRAY_MEMBER]; /* Compressed data */
 	}			va_compressed;
+	struct						/* Compressed-in-line format */
+	{
+		uint32		va_header;
+		uint32		va_info;	/* Original data size (excludes header) and
+								 * flags */
+		Oid			va_cmid;	/* Oid of compression options */
+		char		va_data[FLEXIBLE_ARRAY_MEMBER]; /* Compressed data */
+	}			va_custom_compressed;
 } varattrib_4b;
 
 typedef struct
@@ -280,8 +290,14 @@ typedef struct
 #define VARDATA_1B(PTR)		(((varattrib_1b *) (PTR))->va_data)
 #define VARDATA_1B_E(PTR)	(((varattrib_1b_e *) (PTR))->va_data)
 
+/* va_info in va_compress contains raw size of datum and optional flags */
 #define VARRAWSIZE_4B_C(PTR) \
-	(((varattrib_4b *) (PTR))->va_compressed.va_rawsize)
+	(((varattrib_4b *) (PTR))->va_compressed.va_info & 0x3FFFFFFF)
+#define VARFLAGS_4B_C(PTR) \
+	(((varattrib_4b *) (PTR))->va_compressed.va_info >> 30)
+
+#define VARHDRSZ_CUSTOM_COMPRESSED	\
+	(offsetof(varattrib_4b, va_custom_compressed.va_data))
 
 /* Externally visible macros */
 
@@ -310,6 +326,8 @@ typedef struct
 #define VARDATA_EXTERNAL(PTR)				VARDATA_1B_E(PTR)
 
 #define VARATT_IS_COMPRESSED(PTR)			VARATT_IS_4B_C(PTR)
+#define VARATT_IS_CUSTOM_COMPRESSED(PTR)	(VARATT_IS_4B_C(PTR) && \
+											 (VARFLAGS_4B_C(PTR) == 0x02))
 #define VARATT_IS_EXTERNAL(PTR)				VARATT_IS_1B_E(PTR)
 #define VARATT_IS_EXTERNAL_ONDISK(PTR) \
 	(VARATT_IS_EXTERNAL(PTR) && VARTAG_EXTERNAL(PTR) == VARTAG_ONDISK)
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index 4f333586ee..48051b8294 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -37,6 +37,7 @@ enum SysCacheIdentifier
 	AMOPOPID,
 	AMOPSTRATEGY,
 	AMPROCNUM,
+	ATTCOMPRESSIONOID,
 	ATTNAME,
 	ATTNUM,
 	AUTHMEMMEMROLE,
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 9fe950b29d..1b0d714c36 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -351,6 +351,7 @@ CollectedCommand
 CollectedCommandType
 ColorTrgm
 ColorTrgmInfo
+ColumnCompression
 ColumnCompareData
 ColumnDef
 ColumnIOData
@@ -376,6 +377,8 @@ CompositeIOData
 CompositeTypeStmt
 CompoundAffixFlag
 CompressionAlgorithm
+CompressionAmOptions
+CompressionAmRoutine
 CompressorState
 ConditionVariable
 ConditionalStack
-- 
2.19.1

